"""The scenario DSL: a catalogue of scenarios, the run context a scenario acts through, and the response a
door answers with — its assertions included. Runner-agnostic: the in-process and the live runner both
implement `Runner` and hand a scenario the same `Run`.
"""
import json
import re
import threading
import time
import uuid
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from typing import Any, Callable, Iterable, Optional
MODES = ("inprocess", "live")
# Tags with a meaning to the runners. Any other tag is the catalogue's own (permissions, concurrency, public…).
INPROCESS_ONLY_TAGS = {"failure": "fault injection is not possible against a deployment"}
CLOCK_TAG = "clock"
LIVE_MAX_CONCURRENCY = 10
class ScenarioFailed(AssertionError):
"""An expectation did not hold. Carries the response it was about, for the report."""
def __init__(self, message: str, response: "Optional[Response]" = None):
super().__init__(message)
self.response = response
class Skip(Exception):
"""This scenario cannot run in this mode — reported as SKIPPED with the reason, never as passed."""
# ── definition ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Scenario:
id: str
title: str
fn: Callable[["Run"], Any]
stories: tuple = ()
tags: tuple = ()
personas: tuple = ()
modes: tuple = MODES
generated: bool = False # made by a probe generator, not written by hand
def runs_in(self, mode: str) -> Optional[str]:
"""None if it runs in `mode`, else the reason it does not."""
if mode not in self.modes:
return f"declared {'/'.join(self.modes)} only"
if mode == "live":
for tag, why in INPROCESS_ONLY_TAGS.items():
if tag in self.tags:
return f"in-process only: {why}"
return None
_ID = re.compile(r"[A-Z][A-Z0-9]*(-[A-Z0-9]+)*-\d{2,3}[a-z]?|PROBE-[A-Z0-9-]+")
class Catalogue:
"""A brick's or a composition's scenarios, and how to stand the composition up for them.
CAT = Catalogue("orders", compose=HERE / "compose.yaml", roles=HERE / "roles.yaml",
ticks=["POST /orders/tick"], tick_as="ticker", clock="orders")
@CAT.scenario("SC-OR-01", "an order is due three working days after it is placed",
stories=["OR-01"], tags=["clock"], personas=["clerk"])
def due_date(s): ...
compose the compose.yaml (or a Composition) — the bricks, exactly as they ship
roles the role catalogue: each pool role → its permissions (the file the pool is seeded from)
ticks the composition's tick routes ("POST /orders/tick"); `s.tick()` calls each, as `tick_as`
clock the brick that mounts the demo clock (erp_platform.democlock) — live clock steps go through it;
`clock_as` the persona that may set it (default: the first role holding <brick>:clock:set)
zone the zone a bare wall-clock time is read in (s.clock.set("2026-10-23T09:00")); default Europe/London
storage {brick: wire(engine) -> undo} for a brick that does not follow the repository convention
(see erp_platform.scenarios.inprocess.wire_storage)
"""
def __init__(self, name: str, *, compose: Any, roles: Any, ticks: Iterable[str] = (), tick_as: Optional[str] = None,
clock: Optional[str] = None, clock_as: Optional[str] = None, zone: str = "Europe/London",
storage: Optional[dict] = None, calendar: Any = None):
self.name, self.compose, self.roles_path = name, compose, roles
self.ticks, self.tick_as = tuple(ticks), tick_as
self.clock, self.clock_as, self.zone = clock, clock_as, zone
self.storage = dict(storage or {})
self.calendar = calendar
self.scenarios: list = []
self._generators: list = []
self._roles = None
@property
def roles(self):
from erp_platform.scenarios.personas import RoleCatalogue
if self._roles is None:
self._roles = RoleCatalogue.load(self.roles_path)
return self._roles
def scenario(self, id: str, title: str, *, stories: Iterable[str] = (), tags: Iterable[str] = (),
personas: Iterable[str] = (), modes: Iterable[str] = MODES):
def register(fn):
self.add(Scenario(id, title, fn, tuple(stories), tuple(tags), tuple(personas), tuple(modes)))
return fn
return register
def add(self, sc: Scenario) -> None:
if not sc.generated and not _ID.fullmatch(sc.id):
raise ValueError(f"{sc.id!r}: a scenario id is SC-<MODULE>-nn (SC-HR-08, SC-CT-22)")
if any(x.id == sc.id for x in self.scenarios):
raise ValueError(f"{sc.id}: already in the catalogue")
if not (sc.fn.__doc__ or sc.title):
raise ValueError(f"{sc.id}: a scenario needs a title")
bad = set(sc.modes) - set(MODES)
if bad:
raise ValueError(f"{sc.id}: modes are {MODES}, not {sorted(bad)}")
self.scenarios.append(sc)
def generator(self, fn: Callable[["Catalogue", Any], list]) -> None:
"""A probe generator: called with (catalogue, composition) when a runner starts; returns scenarios."""
self._generators.append(fn)
def expand(self, comp) -> list:
"""The hand-written scenarios plus every generated one, for this composition."""
out = list(self.scenarios)
for gen in self._generators:
out.extend(gen(self, comp))
return out
# probe generators, attached here so a catalogue reads top to bottom (erp_platform.scenarios.probes)
def permission_matrix(self, **kw) -> None:
from erp_platform.scenarios import probes
self.generator(probes.permission_matrix(**kw))
def idempotency(self, **kw) -> None:
from erp_platform.scenarios import probes
self.generator(probes.idempotency(**kw))
def mass_assignment(self, **kw) -> None:
from erp_platform.scenarios import probes
self.generator(probes.mass_assignment(**kw))
# ── what a door answered ───────────────────────────────────────────────────────────────────
_MISSING = object()
_MCP_REFUSAL = re.compile(r"(?:Error executing tool \S+: )?(\d{3}): (.*)", re.S)
def _walk(body: Any, path: str) -> Any:
cur = body
for part in path.split(".") if path else []:
if isinstance(cur, dict):
if part not in cur:
return _MISSING
cur = cur[part]
elif isinstance(cur, list):
try:
cur = cur[int(part)]
except (ValueError, IndexError):
return _MISSING
else:
return _MISSING
return cur
def _show(v: Any) -> str:
return json.dumps(v, default=str)[:300] if v is not _MISSING else "(absent)"
def _same(actual: Any, expected: Any) -> Optional[str]:
"""None when equal, else why not. Strict about type: 1 is not "1", True is not 1, and money is exact."""
if isinstance(expected, Decimal):
if isinstance(actual, float):
return f"money came back as a float ({actual!r}); it leaves both doors as an exact string"
if isinstance(actual, bool) or not isinstance(actual, (str, int, Decimal)):
return f"expected the amount {expected} and got {_show(actual)}"
try:
got = Decimal(str(actual))
except InvalidOperation:
return f"expected the amount {expected} and got {_show(actual)}"
if got != expected or got.as_tuple().exponent != expected.as_tuple().exponent:
return f"expected exactly {expected} and got {actual}"
return None
if isinstance(expected, bool) or isinstance(actual, bool):
return None if type(actual) is type(expected) and actual == expected else \
f"expected {_show(expected)} and got {_show(actual)}"
if isinstance(expected, (int, float)) and isinstance(actual, (int, float)):
return None if actual == expected else f"expected {expected} and got {actual}"
if isinstance(expected, (datetime, date)):
expected = expected.isoformat()
if type(actual) is not type(expected) and not (isinstance(actual, list) and isinstance(expected, tuple)):
return f"expected {_show(expected)} ({type(expected).__name__}) and got {_show(actual)} ({type(actual).__name__})"
if isinstance(expected, tuple):
expected = list(expected)
return None if actual == expected else f"expected {_show(expected)} and got {_show(actual)}"
@dataclass
class Exchange:
"""One request and its response, as the report shows them (secrets redacted when rendered)."""
door: str # http | mcp | public
persona: str
method: str
target: str # path, tool name or public path
request: Any
status: int
body: Any
duration_ms: int
phase: str = ""
class Response:
"""What a door answered. Every assertion returns the response, so they chain:
r.ok().field("total", Decimal("99.99")).field("status", "open")
"""
def __init__(self, exchange: Exchange, headers: Optional[dict] = None, text: str = ""):
self.exchange = exchange
self.status_code = exchange.status
self.body = exchange.body
self.headers = {k.lower(): v for k, v in (headers or {}).items()}
self.text = text
self.door = exchange.door
def __repr__(self) -> str:
return f"<{self.door} {self.exchange.method} {self.exchange.target} → {self.status_code}>"
def fail(self, message: str) -> None:
raise ScenarioFailed(f"{self.exchange.method} {self.exchange.target} as {self.exchange.persona} "
f"({self.door}): {message}", self)
# status and refusal
@property
def reason(self) -> str:
if isinstance(self.body, dict) and isinstance(self.body.get("detail"), str):
return self.body["detail"]
return ""
def status(self, code: int) -> "Response":
if self.status_code != code:
self.fail(f"expected status {code}, got {self.status_code} {_show(self.body)}")
return self
def ok(self) -> "Response":
if not 200 <= self.status_code < 300:
self.fail(f"expected success, got {self.status_code} {_show(self.body)}")
return self
def refused(self, code: int, reason: Optional[str] = None) -> "Response":
"""Refused with this status — and, if given, a reason containing these words."""
self.status(code)
if reason is not None and reason.lower() not in self.reason.lower():
self.fail(f"refused {code} as expected, but the reason was {self.reason!r}, not one saying {reason!r}")
return self
def not_found(self) -> "Response":
"""A row this persona may not see: exactly 404 — a 403 would confirm it exists."""
if self.status_code != 404:
self.fail(f"expected 404 (a row outside this persona's visibility does not exist for them), "
f"got {self.status_code} {_show(self.body)}")
return self
def forbidden(self, permission: Optional[str] = None) -> "Response":
return self.refused(403, f"missing permission: {permission}" if permission else None)
def unauthorized(self) -> "Response":
return self.status(401)
# fields
def value(self, path: str) -> Any:
v = _walk(self.body, path)
if v is _MISSING:
self.fail(f"no field {path!r} in {_show(self.body)}")
return v
__getitem__ = value
def field(self, path: str, expected: Any) -> "Response":
"""Exact: `Decimal("99.99")` compares money to the minor unit (and refuses a float); 1 is not "1"."""
why = _same(_walk(self.body, path), expected)
if why:
self.fail(f"{path}: {why}")
return self
def fields(self, expected: dict) -> "Response":
for k, v in expected.items():
self.field(k, v)
return self
def has(self, path: str) -> "Response":
self.value(path)
return self
def absent(self, path: str) -> "Response":
if _walk(self.body, path) is not _MISSING:
self.fail(f"field {path!r} should not be there: {_show(_walk(self.body, path))}")
return self
def matches(self, path: str, predicate: Callable[[Any], bool], what: str = "") -> "Response":
v = self.value(path)
if not predicate(v):
self.fail(f"{path} = {_show(v)} is not {what or 'what was expected'}")
return self
def contains(self, path: str, item: Any) -> "Response":
v = self.value(path)
if item not in (v or []):
self.fail(f"{path} does not contain {_show(item)}: {_show(v)}")
return self
def length(self, path: str, n: int) -> "Response":
v = self.value(path)
if len(v) != n:
self.fail(f"{path} has {len(v)} items, expected {n}: {_show(v)}")
return self
def same_as(self, before: "Response", ignore: Iterable[str] = ()) -> "Response":
"""The same as before — status AND body, field for field (idempotency: a replay answers what the first
call answered). `ignore` names top-level fields that may differ (a request id, a read time)."""
skip = set(ignore)
if self.status_code != before.status_code:
self.fail(f"a replay answered {self.status_code}, the first call {before.status_code}")
a, b = self.body, before.body
if isinstance(a, dict) and isinstance(b, dict):
a = {k: v for k, v in a.items() if k not in skip}
b = {k: v for k, v in b.items() if k not in skip}
if a != b:
diff = sorted(k for k in set(a) | set(b) if a.get(k) != b.get(k)) if isinstance(a, dict) and isinstance(b, dict) else []
self.fail("a replay answered differently from the first call"
+ (f" in {', '.join(diff)}: {_show({k: a.get(k) for k in diff})} vs {_show({k: b.get(k) for k in diff})}"
if diff else f": {_show(a)} vs {_show(b)}"))
return self
class Responses(list):
"""The answers to concurrent requests."""
def statuses(self) -> Counter:
return Counter(r.status_code for r in self)
def expect(self, counts: dict, run: "Optional[Run]" = None) -> "Responses":
"""Exactly these many of each status: `{200: 1, 409: 4}`."""
got = self.statuses()
if dict(got) != {k: v for k, v in counts.items() if v}:
raise ScenarioFailed(f"{len(self)} concurrent requests answered {dict(sorted(got.items()))}, "
f"expected {dict(sorted(counts.items()))}",
next((r for r in self if r.status_code not in counts), self[0] if self else None))
return self
def where(self, status: int) -> list:
return [r for r in self if r.status_code == status]
# ── acting: personas at the doors ──────────────────────────────────────────────────────────
class Actor:
"""A persona at the composition's doors. Paths are the composition's (`/orders/orders/O-1`)."""
def __init__(self, run: "Run", persona: str):
self.run, self.persona = run, persona
def __repr__(self) -> str:
return f"<{self.persona}>"
def request(self, method: str, path: str, body: Any = None, params: Optional[dict] = None) -> Response:
return self.run._record(self.run.runner.http(self.persona, method.upper(), path, body, params,
self.run._request_id()))
def get(self, path: str, **params) -> Response:
return self.request("GET", path, None, params or None)
def post(self, path: str, body: Any = None, **params) -> Response:
return self.request("POST", path, body if body is not None else {}, params or None)
def put(self, path: str, body: Any = None) -> Response:
return self.request("PUT", path, body if body is not None else {})
def patch(self, path: str, body: Any = None) -> Response:
return self.request("PATCH", path, body if body is not None else {})
def delete(self, path: str, body: Any = None) -> Response:
return self.request("DELETE", path, body)
def tool(self, name: str, args: Optional[dict] = None, **kw) -> Response:
"""The same use case through MCP, the agent door — with a real MCP client session."""
return self.run._record(self.run.runner.mcp(self.persona, name, {**(args or {}), **kw},
self.run._request_id()))
def call(self, use_case: str, args: Optional[dict] = None, *, door: str = "http") -> Response:
"""A use case by NAME, through either door: the route is read off its declaration."""
brick, uc = self.run.use_case(use_case)
args = uc.to_wire(dict(args or {})) # 0.9.7: an aliased input (date_from) goes as its alias (from)
if door == "mcp":
return self.tool(use_case, args)
path = f"/{brick}" + re.sub(r"{(\w+)}", lambda m: str(args.pop(m.group(1))), uc.path)
if uc.method == "GET":
return self.request("GET", path, None, args or None)
return self.request(uc.method, path, args)
class PublicLink:
"""A link someone outside holds — reached through the public app with no login, by whoever holds it."""
def __init__(self, run: "Run", url: str, holder: str = "the recipient"):
self.run, self.url, self.holder = run, url, holder
def open(self) -> Response:
return self.run._record(self.run.runner.public("GET", self.url, None, self.holder))
def submit(self, form: Optional[dict] = None, files: Optional[dict] = None) -> Response:
"""POST the page's form; `files` = {field: (filename, bytes, content_type)} sends it multipart."""
return self.run._record(self.run.runner.public("POST", self.url, form or {}, self.holder, files))
class Clock:
"""The business clock (erp_platform.work.now), moved through what the runner can reach: in process
directly, live only through a deployment's demo clock route (erp_platform.democlock) — else Skip."""
def __init__(self, run: "Run"):
self.run = run
def _impl(self):
impl = self.run.runner.clock
why = impl.unavailable()
if why:
raise Skip(f"in-process only here: {why}")
return impl
def _instant(self, when: Any) -> datetime:
from erp_platform import work
if isinstance(when, datetime):
return work.aware(when).astimezone(timezone.utc)
text = str(when).strip()
if " " not in text and not re.search(r"(Z|[+-]\d\d:?\d\d)$", text):
text = f"{text} {self.run.catalogue.zone}" # a bare wall-clock time is the catalogue's zone
return work.parse_instant(text)
def set(self, when: Any) -> datetime:
"""Freeze the clock at a time: "2026-10-23T10:00 Europe/London", "2026-10-23T09:00Z", a datetime."""
at = self._instant(when)
self.run._step(f"the clock reads {at.isoformat().replace('+00:00', 'Z')}")
return self._impl().set(at)
to = set
def now(self) -> datetime:
return self._impl().now()
def advance(self, by: str, *, calendar: Any = None) -> datetime:
"""On by "+3d", "+2h30m", "-15m", or "+3 working days" (the same local wall-clock time, DST-correct,
counted in `calendar` or the catalogue's — Monday to Friday by default)."""
from erp_platform import work
impl = self._impl()
m = re.fullmatch(r"\s*([+-])\s*(\d+)\s+working\s+days?\s*", by)
if m:
cal = calendar or self.run.catalogue.calendar or work.WorkCalendar()
n = int(m.group(2)) * (-1 if m.group(1) == "-" else 1)
tz = work.zone(self.run.catalogue.zone)
local = impl.now().astimezone(tz)
day = cal.add_working_days(local.date(), n)
target = datetime.combine(day, local.time().replace(tzinfo=None)).replace(tzinfo=tz, fold=0)
self.run._step(f"the clock moves {by}")
return impl.set(target.astimezone(timezone.utc))
m = work._OFFSET.fullmatch(by.replace(" ", ""))
if not m or not any(m.groups()[1:]):
raise ValueError(f"advance by +3d, +2h30m or '+3 working days', not {by!r}")
sign = -1 if m.group(1) == "-" else 1
d, h, mins = (int(g or 0) for g in m.groups()[1:])
self.run._step(f"the clock moves {by}")
return impl.set(impl.now() + sign * timedelta(days=d, hours=h, minutes=mins))
class Run:
"""What a scenario acts through: `s` in `def scenario(s): ...`.
s.given("...") / s.when("...") / s.then("...") the phases, as a consultant reads them in the report
s.as_("clerk") / s.as_("clerk", 2) a persona (the 2nd clerk is another person, same role)
s.anonymous no identity at all
s.key("ORD") a unique key, prefixed with this run's id
s.clock set / advance / now
s.tick() every tick route the catalogue lists, as tick_as
s.eventually(probe, until=..., timeout=...) poll (ticking, if asked) until it holds
s.concurrently(n, lambda i: ...) n requests at once
s.link(url) a public link, held by someone with no login
s.fault.* failure injection — in process only (else Skip)
s.check(cond, "what should hold", got) any other expectation
"""
def __init__(self, runner: "Runner", scenario: Scenario, catalogue: Catalogue):
self.runner, self.scenario, self.catalogue = runner, scenario, catalogue
self.mode = runner.mode
self.steps: list = []
self.exchanges: list = []
self.phase = ""
self._n = 0
self._lock = threading.Lock()
self.clock = Clock(self)
self.notes: list = []
# phases
def given(self, text: str) -> "Run":
return self._phase("GIVEN", text)
def when(self, text: str) -> "Run":
return self._phase("WHEN", text)
def then(self, text: str) -> "Run":
return self._phase("THEN", text)
def _phase(self, kind: str, text: str) -> "Run":
self.phase = kind
self.steps.append(f"{kind} {text}")
return self
def _step(self, text: str) -> None:
self.steps.append(f" · {text}")
def note(self, text: str) -> None:
self.notes.append(text)
self._step(text)
# personas and keys
def as_(self, role: str, n: int = 1) -> Actor:
persona = role if n == 1 else f"{role}#{n}"
declared = self.scenario.personas
if declared and persona not in declared and role not in declared:
raise ValueError(f"{self.scenario.id} acts as {persona!r} but declares personas {list(declared)}")
if not self.catalogue.roles.known(role) and role != "anonymous":
raise ValueError(f"{persona!r}: no role {role!r} in the role catalogue {self.catalogue.roles.source}")
return Actor(self, persona)
@property
def anonymous(self) -> Actor:
return Actor(self, "anonymous")
def key(self, prefix: str = "K") -> str:
with self._lock:
self._n += 1
n = self._n
return f"{prefix}-{self.runner.run_id}-{self.scenario.id.split('-')[-1]}{uuid.uuid4().hex[:4]}-{n}"
def email(self, name: str = "someone") -> str:
"""A unique outside address — a customer, a signer — under the run's prefix."""
return f"{name}.{self.runner.run_id}.{uuid.uuid4().hex[:6]}@customer.example"
def link(self, url: str, holder: str = "the recipient") -> PublicLink:
return PublicLink(self, url, holder)
@property
def fault(self):
return self.runner.faults(self)
def use_case(self, name: str) -> tuple:
"""(brick name, UseCase) for a use case of the composition, by name."""
for m in self.runner.comp.members:
for uc in m.brick.use_cases:
if uc.name == name:
return m.name, uc
raise KeyError(f"no use case {name!r} in the composition")
# ticks, polling, concurrency
def tick(self) -> list:
if not self.catalogue.ticks:
raise ValueError(f"{self.catalogue.name}: the catalogue lists no tick routes")
out = []
for route in self.catalogue.ticks:
method, _, path = route.partition(" ")
out.append(Actor(self, self.catalogue.tick_as or "anonymous").request(method, path, {}).ok())
return out
def eventually(self, probe: Callable[[], Response], *, until: Callable[[Response], bool], timeout: float = 10.0,
every: float = 0.25, tick: bool = False, what: str = "") -> Response:
"""Ask `probe` until `until(response)` holds (ticking first each round if `tick`) or `timeout` passes."""
deadline = time.monotonic() + timeout
self._step(f"until {what or 'it holds'} (≤ {timeout:g}s)")
last = None
while True:
if tick:
self.tick()
last = probe()
try:
if until(last):
return last
except ScenarioFailed:
pass
if time.monotonic() >= deadline:
raise ScenarioFailed(f"{what or 'the condition'} did not hold within {timeout:g}s", last)
time.sleep(every)
def concurrently(self, n: int, fn: Callable[[int], Response]) -> Responses:
"""`n` requests released together (a barrier), each fn(i) on its own thread."""
cap = self.runner.max_concurrency
if n > cap:
self.note(f"concurrency {n} capped at {cap} in {self.mode} mode")
n = cap
self._step(f"{n} requests at once")
gate = threading.Barrier(n)
def one(i: int) -> Response:
gate.wait()
return fn(i)
with ThreadPoolExecutor(n) as pool:
return Responses(pool.map(one, range(n)))
def check(self, condition: Any, what: str, got: Any = None) -> None:
if not condition:
raise ScenarioFailed(f"{what}" + (f" — got {_show(got)}" if got is not None else ""))
# bookkeeping
def _request_id(self) -> str:
with self._lock:
self._n += 1
return f"scn-{self.runner.run_id}-{self.scenario.id}-{self._n}"
def _record(self, r: Response) -> Response:
r.exchange.phase = self.phase
with self._lock:
self.exchanges.append(r.exchange)
self._step(f"{r.exchange.persona} {r.exchange.method} {r.exchange.target} → {r.status_code}")
return r
# ── results ────────────────────────────────────────────────────────────────────────────────
@dataclass
class Result:
id: str
title: str
mode: str
outcome: str # passed | failed | skipped | error
duration: float
message: str = ""
stories: tuple = ()
tags: tuple = ()
steps: list = field(default_factory=list)
exchanges: list = field(default_factory=list)
failing: Optional[Exchange] = None
generated: bool = False
class Runner:
"""What a mode provides. Subclasses: InProcessRunner, LiveRunner."""
mode = ""
max_concurrency = 32
def __init__(self, catalogue: Catalogue):
self.catalogue = catalogue
self.run_id = "scn" + uuid.uuid4().hex[:6]
self.comp = None
self.clock = None
# the doors — implemented per mode
def http(self, persona, method, path, body, params, request_id) -> Response: raise NotImplementedError
def mcp(self, persona, name, args, request_id) -> Response: raise NotImplementedError
def public(self, method, url, form, holder, files=None) -> Response: raise NotImplementedError
def raw(self, method, path, headers, body=None) -> Response: raise NotImplementedError
def faults(self, run: Run): raise Skip(f"in-process only: fault injection is not possible in {self.mode} mode")
def start(self) -> None: ...
def stop(self) -> None: ...
def before(self, sc: Scenario) -> Optional[str]: return None # a reason to skip, or None
def after(self, sc: Scenario) -> None: ...
def __enter__(self):
self.start()
return self
def __exit__(self, *exc):
self.stop()
def scenarios(self, select: Optional[Callable[[Scenario], bool]] = None) -> list:
return [sc for sc in self.catalogue.expand(self.comp) if select is None or select(sc)]
def run_one(self, sc: Scenario) -> Result:
t0 = time.monotonic()
res = Result(sc.id, sc.title, self.mode, "passed", 0.0, stories=sc.stories, tags=sc.tags,
generated=sc.generated)
why = sc.runs_in(self.mode) or self.before(sc)
if why:
res.outcome, res.message = "skipped", why
return res
run = Run(self, sc, self.catalogue)
try:
sc.fn(run)
except Skip as e:
res.outcome, res.message = "skipped", str(e)
except ScenarioFailed as e:
res.outcome, res.message = "failed", str(e)
res.failing = e.response.exchange if e.response is not None else (run.exchanges[-1] if run.exchanges else None)
except Exception as e: # noqa: BLE001 — a broken scenario is an error, not a pass
res.outcome, res.message = "error", f"{type(e).__name__}: {e}"
res.failing = run.exchanges[-1] if run.exchanges else None
finally:
try:
self.after(sc)
except Exception as e: # noqa: BLE001
if res.outcome == "passed":
res.outcome, res.message = "error", f"cleanup failed: {type(e).__name__}: {e}"
if run.notes and res.outcome == "passed":
res.message = "; ".join(run.notes)
res.steps, res.exchanges = run.steps, run.exchanges
res.duration = time.monotonic() - t0
return res
def run_all(self, select: Optional[Callable[[Scenario], bool]] = None, shuffle: Optional[int] = None) -> list:
chosen = self.scenarios(select)
if shuffle is not None:
import random
random.Random(shuffle).shuffle(chosen) # independence: no scenario may lean on another's data
return [self.run_one(sc) for sc in chosen]
def mcp_response(persona: str, name: str, args: dict, result: Any, ms: int) -> Response:
"""An MCP tool result as a Response: a refusal "<status>: <reason>" becomes that status and {"detail": reason}."""
text = result.content[0].text if getattr(result, "content", None) else ""
if not result.is_error:
try:
body = json.loads(text)
except ValueError:
body = text
status = 200
else:
m = _MCP_REFUSAL.fullmatch(text.strip())
if m:
status, body = int(m.group(1)), {"detail": m.group(2)}
else:
status = 422 if "validation error" in text.lower() else 500
body = {"detail": text[:500]}
return Response(Exchange("mcp", persona, "TOOL", name, args, status, body, ms), text=text)
def http_response(door: str, persona: str, method: str, target: str, request: Any, r: Any, ms: int) -> Response:
try:
body = r.json()
except ValueError:
body = r.text[:2000]
return Response(Exchange(door, persona, method, target, request, r.status_code, body, ms),
headers=dict(r.headers), text=r.text)