#!/usr/bin/env python
"""The scenario harness's LIVE runner, against a real DataK3 bucket.
The proof composition (tests/scenarios: orders + stock + mailer) is served on this machine as a tier-1 app —
staff app behind the stand-in gateway, public app beside it, real sockets — with every brick's rows, the
capability claims and the public-link tables on the bucket, through a real engine (livecheck.engine(): your
CLI login, never printed). The LIVE runner then runs the SAME catalogue the unit suite runs in process, as
personas with real bearer tokens (a local issuer: erp-platform-test has no user pool).
1. the runner refuses the bucket, which is not stamped demo — then runs with --allow, as it says to;
2. every scenario that can run live passes on DataK3; the fault scenarios are skipped with their reason;
3. with DODIL_ENV=live the deployment hides its demo clock, and the clock scenarios are SKIPPED, not passed.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_scenario_check.py [--keep] [--report DIR]
It creates its tables (brick tables, claim and version tables, erp_public_link*) and drops them at the end.
"""
import os
import pathlib
import sys
import tempfile
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "tests" / "scenarios"), str(ROOT / "tests" / "scenario_bricks")]
os.environ.setdefault("DODIL_ENV", "test")
from erp_platform import compose, livecheck as lc # noqa: E402
from erp_platform.scenarios import cli, local, report # noqa: E402
from erp_platform.scenarios.gateway import LocalIssuer # noqa: E402
from erp_platform.scenarios.live import LiveRefused, LiveRunner # noqa: E402
import catalogue as shop # noqa: E402
CAT = shop.CAT
PLATFORM_SQL = ROOT / "src" / "erp_platform" / "platform.sql"
PLATFORM_TABLES = ["erp_public_link", "erp_public_link_event"]
def tables_of(comp) -> list:
out = []
for m in comp.members:
out.append((m.schema(), sorted(compose._tables(m.schema()))))
out.append((PLATFORM_SQL, PLATFORM_TABLES))
return out
def main() -> int:
if lc.BUCKET != "erp-platform-test":
raise SystemExit("this check writes tables: run it on BUCKET=erp-platform-test (DODIL_ORG=dodil)")
cli.quiet()
comp = compose.load(CAT.compose)
schema = tables_of(comp)
every = [t for _, ts in schema for t in ts]
for path, ts in schema:
lc.create_schema(path, ts)
engine = lc.engine(pool_size=10)
out = pathlib.Path(sys.argv[sys.argv.index("--report") + 1]) if "--report" in sys.argv else None
code = 1
try:
with tempfile.TemporaryDirectory() as tmp:
key = LocalIssuer.new_key()
issuer = LocalIssuer(key, issuer="https://appid.local/dodil/erp-platform-test", audience="pool:scenario-shop")
local.write_secrets(pathlib.Path(tmp) / "secrets", key, issuer=issuer.issuer, audience=issuer.audience)
secrets = str(pathlib.Path(tmp) / "secrets")
with local.serve(CAT, lambda name: engine, issuer, create=False) as (base, public_base):
print(f"composition served at {base} (public {public_base}) on bucket {lc.BUCKET}\n")
try:
LiveRunner(CAT, base_url=base, public_base_url=public_base, bucket=lc.BUCKET, org=lc.ORG).start()
print("FAIL the live runner accepted an unstamped bucket")
return 1
except LiveRefused as e:
print(f"refused, as it must be: {e}\n")
with LiveRunner(CAT, base_url=base, public_base_url=public_base, bucket=lc.BUCKET, org=lc.ORG,
allow=True, secrets=secrets) as r:
results = r.run_all()
print(report.console(results))
os.environ["DODIL_ENV"] = "live"
with LiveRunner(CAT, base_url=base, public_base_url=public_base, allow=True, secrets=secrets) as r:
clocked = r.run_all(lambda sc: "clock" in sc.tags)
os.environ["DODIL_ENV"] = "test"
print("\nwith DODIL_ENV=live on the deployment:")
print(report.console(clocked))
if out:
out.mkdir(parents=True, exist_ok=True)
(out / "live.md").write_text(report.markdown(results + clocked, f"{CAT.name} — live on {lc.BUCKET}"))
(out / "live.xml").write_text(report.junit(results + clocked, CAT.name))
c = report.summary(results)
clock_skipped = all(x.outcome == "skipped" and "no demo clock" in x.message for x in clocked)
print(f"\nclock scenarios skipped (not passed) on a live environment: {clock_skipped}")
code = 0 if not c["failed"] and not c["error"] and clock_skipped and c["passed"] else 1
finally:
engine.dispose()
if "--keep" not in sys.argv:
left = lc.drop_all(every)
print("tables dropped" + (f" — COULD NOT DROP {left}" if left else ""))
return code
if __name__ == "__main__":
raise SystemExit(main())