99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
|
"""The reference journey, crossing two surfaces.
|
||
|
|
|
||
|
|
Sharing happens in the browser UI, where a person would do it. Setup and
|
||
|
|
revocation go through the API. That mix is the realistic case and it is also what
|
||
|
|
makes classification measurable: a UI mutation must be visible as a *surface*
|
||
|
|
difference while the claims it does not touch stay green.
|
||
|
|
|
||
|
|
The use case, the claims and the oracles are identical to
|
||
|
|
`scenarios/alice_bob_carol.py`. Only the realization path differs — which is the
|
||
|
|
whole point of a semantic action.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from testdriver import (
|
||
|
|
Actor, Cast, DirectDriver, Oracle, Scenario, SemanticAction, StateObserver,
|
||
|
|
Step, VerificationAsset, World,
|
||
|
|
)
|
||
|
|
from testdriver.agentic import DiscoveryRuntime
|
||
|
|
from testdriver.browser import BrowserDriver
|
||
|
|
from testdriver.drivers import CompositeDriver
|
||
|
|
from lab.mutations import ObservationChannel, build_lab
|
||
|
|
from scenarios.alice_bob_carol import RESOURCE, USE_CASE
|
||
|
|
from scenarios.browser_grant import lab_server # re-exported for callers
|
||
|
|
|
||
|
|
API = frozenset({"api"})
|
||
|
|
BROWSER = frozenset({"browser"})
|
||
|
|
|
||
|
|
__all__ = ["build_journey", "lab_server", "RESOURCE", "USE_CASE"]
|
||
|
|
|
||
|
|
|
||
|
|
def build_journey(app, tokens, base_url, runtime=None):
|
||
|
|
cast = Cast()
|
||
|
|
for name in ("alice", "bob", "carol"):
|
||
|
|
cast.add(Actor(name, name.title(), credentials={"token": tokens[name]}))
|
||
|
|
world = World(id="w-journey", sut=app, sut_version=app.version, cast=cast)
|
||
|
|
|
||
|
|
from testdriver.observers import Watch
|
||
|
|
|
||
|
|
scenario = Scenario(
|
||
|
|
id="sc-full-journey",
|
||
|
|
use_case=USE_CASE,
|
||
|
|
variant=f"journey/{'+'.join(app.applied_mutations) or 'baseline'}",
|
||
|
|
watches=(Watch("bob", RESOURCE), Watch("carol", RESOURCE)),
|
||
|
|
steps=(
|
||
|
|
Step("s1-create", "alice", SemanticAction(
|
||
|
|
"create_resource",
|
||
|
|
{"resource_id": RESOURCE, "content": "the secret"},
|
||
|
|
permitted_surfaces=API,
|
||
|
|
postcondition=lambda obs: "audit:R" in obs,
|
||
|
|
)),
|
||
|
|
Step("s2-grant", "alice", SemanticAction(
|
||
|
|
"grant_access",
|
||
|
|
{"subject_id": "bob", "permission": "READ"},
|
||
|
|
permitted_surfaces=BROWSER,
|
||
|
|
postcondition=lambda obs: obs["state_permission:bob:R"] == "READ",
|
||
|
|
)),
|
||
|
|
Step("s3-revoke", "alice", SemanticAction(
|
||
|
|
"revoke_access",
|
||
|
|
{"resource_id": RESOURCE, "subject_id": "bob"},
|
||
|
|
permitted_surfaces=API,
|
||
|
|
postcondition=lambda obs: obs["state_permission:bob:R"] is None,
|
||
|
|
)),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
driver = CompositeDriver(
|
||
|
|
{
|
||
|
|
"api": DirectDriver(app, tokens),
|
||
|
|
"browser": BrowserDriver(
|
||
|
|
base_url, tokens, runtime or DiscoveryRuntime(), RESOURCE
|
||
|
|
),
|
||
|
|
},
|
||
|
|
default="api",
|
||
|
|
)
|
||
|
|
observer = StateObserver(ObservationChannel(app), scenario.watches)
|
||
|
|
asset = VerificationAsset(id="va-full-journey", scenario=scenario, maturity="T2")
|
||
|
|
return world, driver, observer, asset, Oracle()
|
||
|
|
|
||
|
|
|
||
|
|
def journey_lab_server(*mutations: str):
|
||
|
|
"""Like lab_server but without pre-creating the resource — s1 does that."""
|
||
|
|
import threading
|
||
|
|
from contextlib import contextmanager
|
||
|
|
from lab.http_api import serve
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def _run():
|
||
|
|
app, tokens = build_lab(*mutations)
|
||
|
|
server = serve(app)
|
||
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||
|
|
try:
|
||
|
|
yield app, tokens, f"http://127.0.0.1:{server.server_address[1]}"
|
||
|
|
finally:
|
||
|
|
server.shutdown()
|
||
|
|
server.server_close()
|
||
|
|
|
||
|
|
return _run()
|