dodil/erp-platform
tests/test_seed_starter.py
"""0.9.10 — a fresh small-ERP build in fewer steps (the 2026-10-25 timed lab run: 74 turns, most of them hand-written
setup): serve --local's dev owner holds every declared permission, `erp-compose seed` replays configuration calls
re-runnably, and `init --starter` fetches a starter's calls from the method."""
import json
import pathlib
import shutil
import sys
import pytest
import yaml
from erp_platform import auth, compose, compose_init, local, work
HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE / "scenario_bricks"))
@pytest.fixture
def shop(tmp_path, monkeypatch):
for k in ("BUCKET", "DODIL_ENV", "DEV_ALLOW_ANON", "DEV_USER_SUB", "DEV_USER_EMAIL", "DEV_USER_ROLES",
"DEV_USER_PERMISSIONS"):
monkeypatch.delenv(k, raising=False)
shutil.copy(HERE / "scenarios" / "compose.yaml", tmp_path / "compose.yaml")
assert compose.main(["migrate", "--local", str(tmp_path / "compose.yaml")]) == 0
yield tmp_path
work.reset_clock()
def _serve(monkeypatch, shop, *requests):
"""serve --local with uvicorn replaced by a TestClient: the answers of header-less requests, and auth after."""
import uvicorn
from fastapi.testclient import TestClient
seen = []
def fake_run(app, host, port, **kw):
with TestClient(app) as c:
seen.extend(c.request(m, p, json=b) for m, p, b in requests)
monkeypatch.setattr(uvicorn, "run", fake_run)
assert compose.main(["serve", "--local", str(shop / "compose.yaml")]) == 0
return seen
# ── 1. the dev owner ──────────────────────────────────────────────────────────────────────
def test_serve_local_makes_header_less_requests_the_owner_with_machine_permissions(shop, monkeypatch, capsys):
before = (auth.DEV_ALLOW_ANON, list(auth.DEV_USER_PERMISSIONS))
tick, place = _serve(monkeypatch, shop, ("POST", "/orders/tick", {}),
("POST", "/orders/orders", {"order_id": "O-1", "customer": "ACME",
"lines": [{"sku": "W", "quantity": 1, "unit_price": "1.00"}]}))
assert tick.status_code == 200, tick.text # orders:tick:run: machine, exact grant only
assert place.status_code in (200, 201), place.text
out = capsys.readouterr().out
assert "dev owner owner" in out and "orders:tick:run" in out and "machine:" in out
assert (auth.DEV_ALLOW_ANON, auth.DEV_USER_PERMISSIONS) == before # undone when serve ends
def test_a_set_dev_user_permissions_narrows_the_owner(shop, monkeypatch, capsys):
monkeypatch.setenv("DEV_USER_PERMISSIONS", "*:*:*")
monkeypatch.setattr(auth, "DEV_USER_PERMISSIONS", ["*:*:*"])
(tick,) = _serve(monkeypatch, shop, ("POST", "/orders/tick", {}))
assert tick.status_code == 403 # a wildcard never reaches a machine permission
assert "DEV_USER_PERMISSIONS as set" in capsys.readouterr().out
def test_the_owner_holds_every_declared_permission_but_act_for(shop):
comp = compose.load(str(shop / "compose.yaml"))
person, machine = local.owner_permissions(comp)
assert "orders:tick:run" in machine and "orders:order:write" in person and "stock:reservation:write" in person
assert auth.ACT_FOR not in person + machine
# ── 2. seed ───────────────────────────────────────────────────────────────────────────────
CALLS = [
{"use_case": "place_order", "as": "o1", "args": {"order_id": "O-1", "customer": "ACME",
"lines": [{"sku": "W", "quantity": 2, "unit_price": "5.00"}]}},
{"use_case": "request_customer_approval", "args": {"order_id": "$o1.order_id", "customer_email": "a@acme.example"}},
{"use_case": "approve_order", "args": {"order_id": "$o1.order_id"}},
{"use_case": "tick", "args": {}}, # a machine permission: the owner holds it
]
def _seed(shop, rows, capsys):
(shop / "calls.yaml").write_text(yaml.safe_dump({"calls": rows}))
code = compose.main(["seed", "--local", str(shop / "compose.yaml"), str(shop / "calls.yaml")])
return code, capsys.readouterr()
def test_seed_replays_the_calls_and_a_re_run_skips_what_is_there(shop, capsys):
code, cap = _seed(shop, CALLS, capsys)
assert code == 0, cap.err
assert "seeded 4 call(s): 1 applied, 3 keyed upsert(s), 0 already there" in cap.out
code, cap = _seed(shop, CALLS, capsys)
assert code == 0, cap.err
assert "already there (the first answer, replayed)" in cap.out # the link was not issued twice
assert "1 replayed" in cap.out
def test_seed_stops_at_the_first_refusal_naming_it_and_keeps_what_came_before(shop, capsys):
rows = CALLS[:1] + [{"use_case": "approve_order", "args": {"order_id": "NOPE"}}] + CALLS[3:]
code, cap = _seed(shop, rows, capsys)
assert code == 1
assert "call 2/3 (approve_order): refused 404" in cap.err and "re-run" in cap.err
assert "tick" not in cap.out # nothing after the refusal ran
code, cap = _seed(shop, CALLS[:1], capsys)
assert code == 0 and "place_order" in cap.out # the first call is there
def test_seed_names_an_unknown_use_case_and_a_missing_reference(shop, capsys):
assert _seed(shop, [{"use_case": "nope", "args": {}}], capsys)[1].err.count("no use case 'nope'") == 1
code, cap = _seed(shop, [{"use_case": "approve_order", "args": {"order_id": "$o9.order_id"}}], capsys)
assert code == 1 and "no earlier call saved its answer `as: o9`" in cap.err
def test_seed_without_local_needs_a_bucket(shop, capsys):
(shop / "calls.yaml").write_text("[]\n")
assert compose.main(["seed", str(shop / "compose.yaml"), str(shop / "calls.yaml")]) == 2
assert "BUCKET" in capsys.readouterr().err
def test_seed_local_refuses_an_unmigrated_dir(shop, capsys):
(shop / "calls.yaml").write_text("[]\n")
assert compose.main(["seed", "--local", str(shop / "elsewhere"), str(shop / "compose.yaml"),
str(shop / "calls.yaml")]) == 1
assert "migrate --local" in capsys.readouterr().err
def test_seed_resolves_today_and_tomorrow_on_the_business_clock():
from erp_platform.process_export import resolve
work.set_clock(at=work.parse_instant("2026-11-20T09:00Z"))
try:
assert resolve({"a": ["$today", "$tomorrow", "$5"]}, {}) == {"a": ["2026-11-20", "2026-11-21", "$5"]}
finally:
work.reset_clock()
# ── 3. init --starter ─────────────────────────────────────────────────────────────────────
def _method_with_starter(tmp_path, text):
m = tmp_path / "method"
(m / "starters").mkdir(parents=True)
(m / "index.json").write_text(json.dumps({"latest_release": "v1", "modules": []}))
(m / "starters" / "shop.calls.yaml").write_text(text)
return compose_init.Method(local=m)
def test_read_starter_returns_its_text_and_modules(tmp_path):
text = "modules: [orders, stock]\ncalls:\n - {use_case: tick, args: {}}\n"
assert compose_init.read_starter(_method_with_starter(tmp_path, text), "shop") == (text, ["orders", "stock"])
@pytest.mark.parametrize("name, text, says", [
("../x", "", "lower-case letters"),
("other", "modules: [a]\ncalls: []\n", "has no starters/other.calls.yaml"),
("shop", "- {use_case: tick}\n", "expected a mapping with modules"),
])
def test_read_starter_refuses(tmp_path, name, text, says):
with pytest.raises(compose_init.InitFailed, match=says):
compose_init.read_starter(_method_with_starter(tmp_path, text), name)
def test_init_starter_writes_the_calls_and_prints_the_next_commands(tmp_path, monkeypatch):
from test_compose_init import world as _w # noqa: F401 (the fixture's helpers)
import test_compose_init as tci
src = tmp_path / "src"
src.mkdir()
plat = src / "erp-platform"
plat.mkdir()
(plat / "README.md").write_text("p\n")
plat_url, demo_url = tci.git_repo(plat, "v0.6.0"), tci.git_repo(tci.demo_brick(src), "v0.2.0")
method = tci.fake_method(tmp_path, {"demo": [tci.repo("brick-demo", "brick", "v0.2.0", demo_url),
tci.repo("erp-platform", "platform", "v0.6.0", plat_url)]}, tci.ENTRY)
(method / "starters").mkdir()
starter = "# PLACEHOLDER names\nmodules: [demo]\ncalls:\n - {use_case: items, args: {}}\n"
(method / "starters" / "demo-shop.calls.yaml").write_text(starter)
monkeypatch.setenv("ERP_METHOD_DIR", str(method))
said = []
out = tmp_path / "acme"
assert compose_init.init([], root=out, no_install=True, starter="demo-shop", say=said.append) == 0
assert (out / "config" / "demo-shop.calls.yaml").read_text() == starter # 0.9.11: config/<name>.calls.yaml
assert yaml.safe_load((out / "compose.yaml").read_text())["modules"] == ["demo"] # the starter's modules
text = "\n".join(said)
assert "seed --local compose.yaml config/demo-shop.calls.yaml" in text and "migrate --local" in text
assert "serve --local" in text
with pytest.raises(compose_init.InitFailed, match="configures demo too"):
compose_init.init(["other"], root=tmp_path / "b", no_install=True, starter="demo-shop", say=said.append)
# 0.9.11: a second starter (sample data) beside the first, seeded after it; neither overwrites the other
sample = "modules: [demo]\ncalls:\n - {use_case: items, args: {sample: true}}\n"
(method / "starters" / "demo-shop-sample.calls.yaml").write_text(sample)
said.clear()
monkeypatch.setattr(compose_init, "init", lambda mods, **kw: said.append(kw["starter"]) or 0)
assert compose_init.main(["--starter", "demo-shop", "--starter", "demo-shop-sample", "--no-install"]) == 0
assert said == [["demo-shop", "demo-shop-sample"]] # the CLI keeps both, in order
monkeypatch.undo()
monkeypatch.setenv("ERP_METHOD_DIR", str(method))
said.clear()
two = tmp_path / "two"
assert compose_init.init([], root=two, no_install=True, starter=["demo-shop", "demo-shop-sample"],
say=said.append) == 0
assert (two / "config" / "demo-shop.calls.yaml").read_text() == starter
assert (two / "config" / "demo-shop-sample.calls.yaml").read_text() == sample
text = "\n".join(said)
assert text.index("seed --local compose.yaml config/demo-shop.calls.yaml") < \
text.index("seed --local compose.yaml config/demo-shop-sample.calls.yaml")
def test_serve_after_a_seed_today_starts_the_clock_tomorrow_and_a_set_clock_wins(shop, monkeypatch, capsys):
import uvicorn
assert _seed(shop, CALLS[:1], capsys)[0] == 0
seen = []
monkeypatch.setattr(uvicorn, "run", lambda app, host, port, **kw: seen.append(work.now()))
before = work.now()
assert compose.main(["serve", "--local", str(shop / "compose.yaml")]) == 0
assert (seen[-1] - before).total_seconds() > 86000 and "seeded today" in capsys.readouterr().out
assert compose.main(["serve", "--local", "--clock", "+2h", str(shop / "compose.yaml")]) == 0
assert (seen[-1] - before).total_seconds() < 86000
def test_a_module_named_brick_mounts_a_clock_permission_someone_can_hold():
from erp_platform import democlock
from erp_platform.bricks import Brick
b = Brick("lt-sales-order", namespace="lt_so")
democlock.mount(b)
(u,) = [u for u in b.use_cases if u.name.startswith("set_")]
assert u.permission == "lt_so:clock:set" and auth.is_concrete(u.permission)