"""erp-scenarios — run a scenario catalogue in process or against a deployment.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py # in process (the default)
erp-scenarios run tests/scenarios/catalogue.py -k SC-OR-03 -k clock # by id, or by tag
erp-scenarios run crm/scenarios/catalogue.py --live \\
--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 --junit out/scenarios.xml --markdown out/scenarios.md
A CATALOGUE is a Python file or a module ("pkg.scenarios:CAT") holding one erp_platform.scenarios.Catalogue.
Live needs ERP_SCENARIO_SECRETS (see erp_platform.scenarios.personas). Exit 0 when nothing failed or errored
(skips are allowed and listed with their reason), 1 otherwise, 2 for a usage problem.
"""
import argparse
import importlib
import importlib.util
import pathlib
import sys
from erp_platform.scenarios import report
from erp_platform.scenarios.core import Catalogue
def load_catalogue(target: str) -> Catalogue:
name, _, attr = target.partition(":")
if name.endswith(".py") or pathlib.Path(name).is_file():
path = pathlib.Path(name).resolve()
spec = importlib.util.spec_from_file_location(f"_erp_scenarios_{path.stem}", path)
module = importlib.util.module_from_spec(spec)
sys.path.insert(0, str(path.parent))
spec.loader.exec_module(module)
else:
module = importlib.import_module(name)
if attr:
return getattr(module, attr)
found = [v for v in vars(module).values() if isinstance(v, Catalogue)]
if len(found) != 1:
raise SystemExit(f"{target}: expected one Catalogue, found {len(found)} — name it as {target}:NAME")
return found[0]
def quiet(verbose: bool = False) -> None:
"""The apps under test log every request (requestlog) and every moved-clock read (work): in a report run
those lines bury the table. Errors still show; -v shows everything."""
import logging
if not logging.getLogger().handlers:
logging.basicConfig(stream=sys.stderr, format="%(name)s %(levelname)s %(message)s")
level = logging.INFO if verbose else logging.ERROR
for name in ("erp_platform", "httpx", "mcp", "uvicorn"):
logging.getLogger(name).setLevel(level if name == "erp_platform" else max(level, logging.WARNING))
def _selector(keys: list):
if not keys:
return None
return lambda sc: any(k == sc.id or k in sc.tags or sc.id.startswith(k) for k in keys)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="erp-scenarios", description="Run SAP-grade scenarios in process or live.")
sub = ap.add_subparsers(dest="verb", required=True)
ls = sub.add_parser("list", help="the catalogue: ids, stories, tags, modes")
ls.add_argument("catalogue")
run = sub.add_parser("run", help="run scenarios")
run.add_argument("catalogue")
run.add_argument("-k", dest="keys", action="append", default=[], help="an id, an id prefix or a tag (repeatable)")
run.add_argument("--live", action="store_true", help="against a deployment instead of in process")
run.add_argument("--base-url", default="")
run.add_argument("--public-base-url", default="")
run.add_argument("--bucket", default="", help="the bucket the deployment writes: it must be stamped demo")
run.add_argument("--org", default="")
run.add_argument("--allow", action="store_true", help="run live although the bucket is not stamped demo")
run.add_argument("--secrets", default=None, help="the secrets directory (default $ERP_SCENARIO_SECRETS)")
run.add_argument("--junit", default="")
run.add_argument("--markdown", default="")
run.add_argument("--shuffle", type=int, default=None, help="run in a random order (this seed) — no order dependence")
run.add_argument("-v", "--verbose", action="store_true", help="show the apps' request lines and warnings")
a = ap.parse_args(argv)
quiet(getattr(a, "verbose", False))
cat = load_catalogue(a.catalogue)
if a.verb == "list":
from erp_platform import compose
comp = cat.compose if isinstance(cat.compose, compose.Composition) else compose.load(cat.compose)
for sc in cat.expand(comp):
live = sc.runs_in("live")
print(f"{sc.id:<28} {sc.title}")
print(f"{'':<28} stories {', '.join(sc.stories) or '-'} · tags {', '.join(sc.tags) or '-'}"
f" · {live or 'both modes'}")
return 0
if a.live:
if not a.base_url:
print("--live needs --base-url", file=sys.stderr)
return 2
from erp_platform.scenarios.live import LiveRefused, LiveRunner
runner = LiveRunner(cat, base_url=a.base_url, public_base_url=a.public_base_url, bucket=a.bucket,
org=a.org, allow=a.allow, secrets=a.secrets)
try:
runner.start()
except LiveRefused as e:
print(e, file=sys.stderr)
return 1
else:
from erp_platform.scenarios.inprocess import InProcessRunner
runner = InProcessRunner(cat)
runner.start()
try:
results = runner.run_all(_selector(a.keys), shuffle=a.shuffle)
finally:
runner.stop()
print(report.console(results))
if a.junit:
pathlib.Path(a.junit).parent.mkdir(parents=True, exist_ok=True)
pathlib.Path(a.junit).write_text(report.junit(results, cat.name))
if a.markdown:
pathlib.Path(a.markdown).parent.mkdir(parents=True, exist_ok=True)
pathlib.Path(a.markdown).write_text(report.markdown(results, f"{cat.name} — {runner.mode}"))
c = report.summary(results)
return 1 if c["failed"] or c["error"] else 0
if __name__ == "__main__":
raise SystemExit(main())