77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
|
"""Drivers realize semantic actions against a surface.
|
||
|
|
|
||
|
|
A driver knows mechanics. It emits S1 evidence describing *how* it acted and
|
||
|
|
reports which surface it used, but it never reports whether the action was
|
||
|
|
correct — that is S2/S3 and belongs to the observer and the oracle.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Any, Protocol
|
||
|
|
|
||
|
|
from .actions import SemanticAction, Surface
|
||
|
|
from .world import Actor
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True, slots=True)
|
||
|
|
class Realization:
|
||
|
|
"""What a driver did, mechanically. Pure S1."""
|
||
|
|
|
||
|
|
surface_id: str
|
||
|
|
mechanics: dict[str, Any]
|
||
|
|
raised: str | None = None
|
||
|
|
|
||
|
|
|
||
|
|
class Driver(Protocol):
|
||
|
|
surface: Surface
|
||
|
|
|
||
|
|
def realize(self, actor: Actor, action: SemanticAction) -> Realization: ...
|
||
|
|
|
||
|
|
|
||
|
|
class UnsupportedAction(Exception):
|
||
|
|
"""The driver has no mechanical implementation for this semantic action."""
|
||
|
|
|
||
|
|
|
||
|
|
class DirectDriver:
|
||
|
|
"""Deterministic driver against the lab's enforcement path.
|
||
|
|
|
||
|
|
This is the T5 Deterministic end of the maturity continuum: a fixed mapping
|
||
|
|
from semantic action to mechanics, with no discovery and no model.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, lab: Any, tokens: dict[str, str]) -> None:
|
||
|
|
self._lab = lab
|
||
|
|
self._tokens = tokens
|
||
|
|
self.surface = Surface(
|
||
|
|
id="api", kind="http-like", description="lab enforcement path"
|
||
|
|
)
|
||
|
|
|
||
|
|
_MAPPING = {
|
||
|
|
"create_resource": ("create_resource", ("resource_id", "content")),
|
||
|
|
"grant_access": ("grant", ("resource_id", "subject_id", "permission")),
|
||
|
|
"revoke_access": ("revoke", ("resource_id", "subject_id")),
|
||
|
|
"read_resource": ("read_resource", ("resource_id",)),
|
||
|
|
}
|
||
|
|
|
||
|
|
def realize(self, actor: Actor, action: SemanticAction) -> Realization:
|
||
|
|
if action.name not in self._MAPPING:
|
||
|
|
raise UnsupportedAction(action.name)
|
||
|
|
action.check_surface(self.surface.id)
|
||
|
|
|
||
|
|
op, arg_names = self._MAPPING[action.name]
|
||
|
|
args = {name: action.args[name] for name in arg_names if name in action.args}
|
||
|
|
token = self._tokens[actor.id]
|
||
|
|
|
||
|
|
mechanics: dict[str, Any] = {
|
||
|
|
"operation": op,
|
||
|
|
"arguments": args,
|
||
|
|
"actor": actor.id,
|
||
|
|
}
|
||
|
|
try:
|
||
|
|
result = self._lab.request(token, op, **args)
|
||
|
|
except Exception as exc: # the SUT refusing is data, not a framework error
|
||
|
|
return Realization(self.surface.id, mechanics, raised=f"{type(exc).__name__}: {exc}")
|
||
|
|
mechanics["result"] = result
|
||
|
|
return Realization(self.surface.id, mechanics)
|