tegwick
cf682281d7
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e76f-be98-7ae3-965d-e0b31290a4c4
134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
"""Synthetic workflow lab, implemented from usecases/generalisation-spec.md."""
|
|
from copy import deepcopy
|
|
from dataclasses import dataclass
|
|
|
|
from testdriver.actions import Surface
|
|
from testdriver.drivers import Realization
|
|
|
|
|
|
class WorkflowDriver:
|
|
"""A completed denial is a protocol response, not a failed realization."""
|
|
|
|
surface = Surface("workflow", "direct")
|
|
|
|
def __init__(self, lab):
|
|
self.lab = lab
|
|
|
|
def realize(self, actor, action):
|
|
action.check_surface(self.surface.id)
|
|
response = self.lab.request(actor.credentials["token"], action.name, **action.args)
|
|
return Realization(self.surface.id, {"operation": action.name, "response": response})
|
|
|
|
|
|
class ApprovalLab:
|
|
tokens = {"requester": "fixture-requester", "reviewer": "fixture-reviewer",
|
|
"delegate": "fixture-delegate"}
|
|
|
|
def __init__(self, defect=None):
|
|
self.defect = defect
|
|
self.state = "draft"
|
|
self.reviewer = "reviewer"
|
|
self.approved_by = None
|
|
self.receipts = []
|
|
|
|
def may_execute(self, actor):
|
|
return actor == "requester" and (
|
|
self.state == "approved" or self.defect == "premature-execution"
|
|
)
|
|
|
|
def request(self, token, operation, **args):
|
|
actor = next((a for a, t in self.tokens.items() if t == token), None)
|
|
allowed = False
|
|
if operation == "submit":
|
|
allowed = actor == "requester" and self.state == "draft"
|
|
if allowed:
|
|
self.state = "pending"
|
|
elif operation == "delegate":
|
|
allowed = (actor == self.reviewer and self.state == "pending"
|
|
and args["to"] == "delegate")
|
|
if allowed:
|
|
self.reviewer = args["to"]
|
|
elif operation == "approve":
|
|
allowed = self.state == "pending" and (
|
|
actor == self.reviewer
|
|
or (actor == "requester" and self.defect == "self-approval")
|
|
or (actor == "reviewer" and self.defect == "stale-delegation")
|
|
)
|
|
if allowed:
|
|
self.state, self.approved_by = "approved", actor
|
|
elif operation == "execute":
|
|
allowed = self.may_execute(actor)
|
|
if allowed:
|
|
self.state = "executed"
|
|
else:
|
|
raise ValueError(operation)
|
|
receipt = {"actor": actor, "operation": operation, "allowed": allowed}
|
|
self.receipts.append(receipt)
|
|
return deepcopy(receipt)
|
|
|
|
|
|
@dataclass
|
|
class ApprovalObserver:
|
|
lab: ApprovalLab
|
|
name: str = "approval-observer"
|
|
|
|
def snapshot(self):
|
|
return deepcopy({
|
|
"state": self.lab.state,
|
|
"reviewer": self.lab.reviewer,
|
|
"approved_by": self.lab.approved_by,
|
|
"receipts": self.lab.receipts,
|
|
"execution_gate": {actor: self.lab.may_execute(actor) for actor in self.lab.tokens},
|
|
})
|
|
|
|
|
|
class TenantLab:
|
|
tokens = {"admin-a": "fixture-admin-a", "admin-b": "fixture-admin-b"}
|
|
tenants = {"admin-a": "A", "admin-b": "B"}
|
|
|
|
def __init__(self, defect=None):
|
|
self.defect = defect
|
|
self.records = {}
|
|
self.deleted = {}
|
|
self.receipts = []
|
|
|
|
def read(self, actor, tenant, resource):
|
|
if self.tenants.get(actor) != tenant and self.defect != "foreign-read":
|
|
return {"allowed": False, "content": None}
|
|
return {"allowed": (tenant, resource) in self.records,
|
|
"content": self.records.get((tenant, resource))}
|
|
|
|
def request(self, token, operation, tenant, resource="R", content=None):
|
|
actor = next((a for a, t in self.tokens.items() if t == token), None)
|
|
allowed = self.tenants.get(actor) == tenant
|
|
key = (tenant, resource)
|
|
if operation == "create":
|
|
allowed = allowed and key not in self.records
|
|
if allowed:
|
|
self.records[key] = (self.deleted.get(key, content)
|
|
if self.defect == "resurrect-content" else content)
|
|
elif operation == "delete":
|
|
allowed = (allowed or self.defect == "foreign-delete") and key in self.records
|
|
if allowed:
|
|
self.deleted[key] = self.records.pop(key)
|
|
if self.defect == "unscoped-delete":
|
|
self.records = {k: v for k, v in self.records.items() if k[1] != resource}
|
|
else:
|
|
raise ValueError(operation)
|
|
receipt = {"actor": actor, "operation": operation, "tenant": tenant, "allowed": allowed}
|
|
self.receipts.append(receipt)
|
|
return deepcopy(receipt)
|
|
|
|
|
|
@dataclass
|
|
class TenantObserver:
|
|
lab: TenantLab
|
|
name: str = "tenant-observer"
|
|
|
|
def snapshot(self):
|
|
return deepcopy({
|
|
"records": {tenant: self.lab.records.get((tenant, "R")) for tenant in ("A", "B")},
|
|
"reads": {f"{actor}:{tenant}": self.lab.read(actor, tenant, "R")
|
|
for actor in self.lab.tokens for tenant in ("A", "B")},
|
|
"receipts": self.lab.receipts,
|
|
})
|