"""The live runner: the same scenarios against a real deployment, as real people.
LiveRunner(CAT, base_url="https://crm-demo-dodil-8000.ignite.dodil.cloud",
public_base_url="https://crm-demo-public-dodil-8000.ignite.dodil.cloud",
bucket="crm-demo", org="dodil")
* PERSONAS are real tokens (erp_platform.scenarios.personas.LiveCredentials, from ERP_SCENARIO_SECRETS): the
appid password grant for people, client_credentials for service clients, or a local issuer for a laptop
deployment behind scenarios.gateway.LocalGateway.
* IT REFUSES TO RUN unless the bucket the deployment writes is stamped `demo` in its own platform_meta (the
stamp reset_demo.py trusts — a URL or a credential can point at the wrong place; the stamp is the bucket's
own claim about itself), or `allow=True` is passed explicitly. Scenarios WRITE: they make records.
* EVERY KEY a scenario makes (s.key, s.email) carries this run's id, so a run's data can be told apart, found
and swept.
* NO FAULT INJECTION: `s.fault` and scenarios tagged `failure` are SKIPPED with the reason. Concurrency is
capped at 10. The CLOCK moves only through a demo clock route the deployment mounts
(erp_platform.democlock, permission <brick>:clock:set, refused under DODIL_ENV=prod|live): with none,
clock scenarios are SKIPPED and the report says so — never silently passed.
"""
import json
import os
import shutil
import subprocess
import time
from typing import Optional
from urllib.parse import urlparse
import httpx
from erp_platform import compose
from erp_platform.scenarios.core import (CLOCK_TAG, LIVE_MAX_CONCURRENCY, Response, Runner, Scenario, http_response,
mcp_response)
from erp_platform.scenarios.personas import ANONYMOUS, LiveCredentials
ALLOWED_STAMPS = ("demo",)
class LiveRefused(RuntimeError):
"""The live runner will not write to this target."""
def bucket_stamp(bucket: str, org: str = "") -> str:
"""The bucket's own environment stamp (platform_meta), read with the CLI login — '' when unstamped."""
dodil = shutil.which("dodil") or "dodil"
cmd = [dodil, "data", "pg", "-b", bucket, "SELECT value FROM platform_meta WHERE key = 'environment'",
"-o", "json"] + (["--org", org] if org else [])
env = {k: v for k, v in os.environ.items() if k != "DODIL_TOKEN"} # see livecheck._cli_env
r = subprocess.run(cmd, capture_output=True, text=True, env=env)
if r.returncode:
raise LiveRefused(f"could not read bucket {bucket!r}'s stamp (is the CLI logged in, with --org?)")
rows = (json.loads(r.stdout or "{}") or {}).get("rows") or []
return str(rows[0]["value"]).strip().lower() if rows else ""
class LiveClock:
def __init__(self, runner: "LiveRunner"):
self.runner = runner
self.why: Optional[str] = "not probed yet"
self.moved = False
def probe(self) -> None:
cat = self.runner.catalogue
if not cat.clock:
self.why = "the catalogue names no brick mounting the demo clock"
return
who = self.persona()
if who is None:
self.why = f"no role holds {cat.clock}:clock:set"
return
r = self.runner.http(who, "GET", f"/{cat.clock}/demo-clock", None, None, "")
self.why = None if r.status_code == 200 else (
f"the deployment exposes no demo clock (GET /{cat.clock}/demo-clock → {r.status_code})")
def persona(self) -> Optional[str]:
cat = self.runner.catalogue
if cat.clock_as:
return cat.clock_as
perm = f"{cat.clock}:clock:set"
return next((r for r, perms in cat.roles.roles.items() if perm in perms), None)
def unavailable(self) -> Optional[str]:
return self.why
def _post(self, body: dict) -> dict:
r = self.runner.http(self.persona(), "POST", f"/{self.runner.catalogue.clock}/demo-clock", body, None, "")
r.ok()
return r.body
def set(self, at):
from erp_platform import work
self.moved = True
return work.parse_instant(self._post({"at": at.isoformat(), "freeze": True})["now"])
def now(self):
from erp_platform import work
r = self.runner.http(self.persona(), "GET", f"/{self.runner.catalogue.clock}/demo-clock", None, None, "")
return work.parse_instant(r.ok().body["now"])
def reset(self) -> None:
if self.moved:
self._post({"reset": True})
self.moved = False
class LiveRunner(Runner):
mode = "live"
max_concurrency = LIVE_MAX_CONCURRENCY
def __init__(self, catalogue, *, base_url: str, public_base_url: str = "", bucket: str = "", org: str = "",
allow: bool = False, secrets: Optional[str] = None, timeout: float = 120.0):
super().__init__(catalogue)
self.base = base_url.rstrip("/")
self.public_base = public_base_url.rstrip("/")
self.bucket, self.org, self.allow, self.secrets, self.timeout = bucket, org, allow, secrets, timeout
self.clock = LiveClock(self)
def start(self) -> None:
if not self.allow:
if not self.bucket:
raise LiveRefused("name the bucket the deployment writes (--bucket), so its stamp can be checked — "
"or pass --allow to run against it anyway")
stamp = bucket_stamp(self.bucket, self.org)
if stamp not in ALLOWED_STAMPS:
raise LiveRefused(f"refusing: bucket {self.bucket} is stamped {stamp or 'nothing'!r}, not 'demo'. "
"Scenarios write records; run them against a demo deployment, or pass --allow.")
cat = self.catalogue
self.comp = cat.compose if isinstance(cat.compose, compose.Composition) else compose.load(cat.compose)
self.creds = LiveCredentials.from_env(cat.roles, self.secrets)
self.client = httpx.Client(base_url=self.base, timeout=self.timeout, follow_redirects=False)
self.clock.probe()
def stop(self) -> None:
client = getattr(self, "client", None)
if client is not None:
client.close()
def before(self, sc: Scenario) -> Optional[str]:
if CLOCK_TAG in sc.tags and self.clock.unavailable():
return f"in-process only here: {self.clock.unavailable()}"
missing = self.creds.missing(sc.personas)
if missing:
return f"no live credentials for {', '.join(missing)} in the secrets directory"
return None
def after(self, sc: Scenario) -> None:
self.clock.reset()
def headers(self, persona: str, request_id: str = "") -> dict:
h = {} if persona == ANONYMOUS else {"Authorization": f"Bearer {self.creds.token(persona)}"}
if request_id:
h["X-Request-Id"] = request_id
return h
def http(self, persona, method, path, body, params, request_id) -> Response:
headers = self.headers(persona, request_id)
t0 = time.monotonic()
r = self.client.request(method, path, json=body if method != "GET" else None, params=params, headers=headers)
return http_response("http", persona, method, path, {"params": params, "body": body, "headers": headers},
r, int((time.monotonic() - t0) * 1000))
def raw(self, method, path, headers, body=None) -> Response:
t0 = time.monotonic()
r = self.client.request(method, path, headers=headers, content=body)
return http_response("http", "raw", method, path, {"headers": headers}, r, int((time.monotonic() - t0) * 1000))
def mcp(self, persona, name, args, request_id) -> Response:
if persona == ANONYMOUS:
r = self.raw("POST", "/mcp", {"content-type": "application/json", "accept": "application/json"},
b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}')
r.exchange.door, r.exchange.persona, r.exchange.method, r.exchange.target = "mcp", persona, "TOOL", name
r.door = "mcp"
return r
import anyio
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
headers = self.headers(persona, request_id)
async def go():
async with httpx2.AsyncClient(headers=headers, timeout=self.timeout) as http:
async with streamable_http_client(f"{self.base}/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return await s.call_tool(name, args)
t0 = time.monotonic()
result = anyio.run(go)
return mcp_response(persona, name, args, result, int((time.monotonic() - t0) * 1000))
def public(self, method, url, form, holder, files=None) -> Response:
u = urlparse(url)
target = f"{self.public_base}{u.path}" if self.public_base else url
t0 = time.monotonic()
r = httpx.request(method, target, data=form if method == "POST" else None, files=files or None, timeout=self.timeout,
headers={"user-agent": "erp-scenarios"}, follow_redirects=False)
return http_response("public", holder, method, u.path, {"form": form}, r, int((time.monotonic() - t0) * 1000))