dodil/erp-platform
tests/test_compose_init.py
"""erp-compose init: resolve a module from the method, clone its bricks at their tags, write the files, check.
The method is a fake erp-method checkout ($ERP_METHOD_DIR) and the bricks are local git repos (file://), so
nothing here touches git.dodil.io. The last test runs the whole thing end to end — a real venv, a real pip
install of this erp-platform and a tiny brick, `erp-compose check` and the composition test."""
import json
import pathlib
import shutil
import subprocess
import textwrap
import pytest
import yaml
from erp_platform import compose_init
from erp_platform.compose_init import InitFailed
ROOT = pathlib.Path(__file__).resolve().parent.parent
GIT = ["git", "-c", "user.name=t", "-c", "user.email=t@t", "-c", "init.defaultBranch=main"]
# The composition test erp-method ships (modules/test_composition.py), abridged to what init needs: it boots
# compose.yaml from the folder above tests/ and calls one plain read on every brick.
TEST_COMPOSITION = textwrap.dedent('''\
import inspect, pathlib
from contextlib import ExitStack
import anyio, httpx
from sqlalchemy import create_engine
from erp_platform import compose, seam
from erp_platform.scenarios import Catalogue
from erp_platform.scenarios.inprocess import stage
COMPOSE = pathlib.Path(__file__).resolve().parent.parent / "compose.yaml"
def test_composition_boots_and_every_brick_answers(tmp_path):
comp = compose.load(COMPOSE)
assert compose.check(comp)["problems"] == []
engine = create_engine(f"sqlite:///{tmp_path / 'b.db'}", connect_args={"check_same_thread": False})
perms = sorted({u.permission for m in comp.members for u in m.brick.use_cases if u.permission})
me = {"sub": "t.1", "email": "t@t", "app_roles": ["t"], "permissions": perms}
async def get(app, path):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t") as c:
return await c.get(path, headers=seam.gateway_headers(me))
with ExitStack() as st:
app = stage(Catalogue("t", compose=comp, roles=None), st, lambda n: engine).app
for m in comp.members:
reads = [u for u in m.brick.use_cases if u.kind == "query" and u.method == "GET" and u.tool
and not u.path_params and all(d is not inspect.Parameter.empty for _, _, d in u.inputs)]
assert anyio.run(get, app, f"/{m.name}{reads[0].path}").status_code == 200
''')
def git_repo(path: pathlib.Path, tag: str) -> str:
"""Commit what is in `path`, tag it, and return a file:// URL to clone it from."""
subprocess.run([*GIT, "init", "-q", path], check=True)
subprocess.run([*GIT, "-C", path, "add", "-A"], check=True)
subprocess.run([*GIT, "-C", path, "commit", "-qm", "release"], check=True)
subprocess.run([*GIT, "-C", path, "tag", tag], check=True)
return path.as_uri()
def demo_brick(root: pathlib.Path, version: str = "0.2.0") -> pathlib.Path:
"""A minimal brick in the family's shape: one table, one repository, one plain read."""
d = root / "brick-demo"
pkg = d / "src" / "brick_demo"
for sub in ("", "application", "infrastructure"):
(pkg / sub).mkdir(parents=True, exist_ok=True)
(pkg / sub / "__init__.py").write_text("")
(d / "pyproject.toml").write_text(textwrap.dedent(f'''\
[project]
name = "brick-demo"
version = "{version}"
dependencies = ["erp-platform"]
[build-system]
requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
[tool.setuptools.package-data]
brick_demo = ["schema.sql"]
'''))
(pkg / "schema.sql").write_text("CREATE TABLE IF NOT EXISTS demo_item (item_id VARCHAR NOT NULL, "
"PRIMARY KEY (item_id));\n")
(pkg / "infrastructure" / "tables.py").write_text(textwrap.dedent('''\
from sqlalchemy import Column, MetaData, String, Table, select
md = MetaData()
item = Table("demo_item", md, Column("item_id", String, primary_key=True))
class DemoRepository:
def __init__(self, conn):
self.conn = conn
def all(self):
return [r.item_id for r in self.conn.execute(select(item))]
'''))
(pkg / "application" / "use_cases.py").write_text(textwrap.dedent(f'''\
from contextlib import contextmanager
from erp_platform.bricks import Brick
DEMO = Brick("demo", version="{version}")
@contextmanager
def repository():
raise RuntimeError("no database configured")
@DEMO.use_case(kind="query", route="GET /items", permission="demo:item:read")
def items(caller: dict) -> dict:
"""Every item."""
with repository() as repo:
return {{"items": repo.all()}}
'''))
return d
def fake_method(root: pathlib.Path, modules: dict, entries: dict, pending=()) -> pathlib.Path:
"""An erp-method checkout: index.json with `modules[].repos`, `brick_entries` and the composition test."""
m = root / "erp-method"
(m / "modules").mkdir(parents=True)
(m / "modules" / "test_composition.py").write_text(TEST_COMPOSITION)
(m / "index.json").write_text(json.dumps({
"version": "9.9.9", "latest_release": "v9.9.9",
"method": {"test_composition": "modules/test_composition.py"},
"modules": [{"id": k, "repos": v} for k, v in modules.items()],
"pending": [{"id": p} for p in pending],
"brick_entries": entries}))
return m
def repo(name, role, tag, url):
return {"name": name, "role": role, "tag": tag, "https": url}
ENTRY = {"brick-demo": {"name": "demo", "brick": "brick_demo.application.use_cases:DEMO"}}
@pytest.fixture
def world(tmp_path, monkeypatch):
"""A fake method with module `demo` (erp-platform + brick-demo, both local git repos)."""
src = tmp_path / "src"
src.mkdir()
plat = src / "erp-platform"
plat.mkdir()
(plat / "README.md").write_text("placeholder platform\n")
plat_url = git_repo(plat, "v0.6.0")
demo_url = git_repo(demo_brick(src), "v0.2.0")
method = fake_method(tmp_path, {
"demo": [repo("brick-demo", "brick", "v0.2.0", demo_url), repo("erp-platform", "platform", "v0.6.0", plat_url),
repo("erp-ui", "ui", "v0.1.0", "file:///nowhere")],
"other": [repo("brick-demo", "brick", "v0.1.0", demo_url), repo("erp-platform", "platform", "v0.6.0", plat_url)],
"empty": []}, ENTRY, pending=["gl"])
monkeypatch.setenv("ERP_METHOD_DIR", str(method))
monkeypatch.delenv("ERP_METHOD_REF", raising=False)
out = tmp_path / "acme"
return {"tmp": tmp_path, "src": src, "method": method, "out": out, "plat_url": plat_url}
def quiet(*_a, **_k):
pass
def test_resolve_puts_erp_platform_first_and_skips_what_is_not_a_brick(world):
plan = compose_init.resolve(compose_init.Method(local=world["method"]), ["demo"])
assert [(r.name, r.tag) for r in plan.repos] == [("erp-platform", "v0.6.0"), ("brick-demo", "v0.2.0")]
assert plan.skipped == ["erp-ui"] and plan.method_ref == "local"
@pytest.mark.parametrize("mods, says", [
(["nope"], "no module 'nope'"),
(["gl"], "planned, with no code yet"),
(["empty"], "has no bricks yet"),
(["demo", "other"], "one module pins v0.2.0, other pins v0.1.0"),
])
def test_resolve_refuses_what_cannot_be_composed(world, mods, says):
with pytest.raises(InitFailed, match=says):
compose_init.resolve(compose_init.Method(local=world["method"]), mods)
def test_a_brick_with_no_entry_is_refused(world, tmp_path):
m = fake_method(tmp_path / "m2", {"demo": [repo("brick-demo", "brick", "v0.2.0", "x"),
repo("erp-platform", "platform", "v0.6.0", "y")]}, {})
with pytest.raises(InitFailed, match="no brick_entries entry"):
compose_init.resolve(compose_init.Method(local=m), ["demo"])
def test_a_remote_method_only_clones_from_dodil_git(world):
method = compose_init.Method.__new__(compose_init.Method)
method.local, method.ref = None, "v9.9.9"
method.index = json.loads((world["method"] / "index.json").read_text())
with pytest.raises(InitFailed, match="not https://git.dodil.io/; refused"):
compose_init.resolve(method, ["demo"])
def test_files_are_written_once_and_a_second_run_changes_nothing(world):
out = world["out"]
assert compose_init.init(["demo"], root=out, no_install=True, say=quiet) == 0
spec = yaml.safe_load((out / "compose.yaml").read_text())
assert spec == {"name": "acme-demo", "version": "0.1.0", "tier": 1, "modules": ["demo"], "bricks": [
{"name": "demo", "brick": "brick_demo.application.use_cases:DEMO", "version": "0.2.0"}]}
reqs = [l for l in (out / "requirements.txt").read_text().splitlines() if not l.startswith("#")]
host = lambda u: u.replace("file:///", "file://localhost/", 1) # noqa: E731
demo_url = (world["src"] / "brick-demo").as_uri()
assert reqs == [f"erp-platform[test] @ git+{host(world['plat_url'])}@v0.6.0 # platform",
f"brick-demo @ git+{host(demo_url)}@v0.2.0 # demo"] # pinned, never copied in
assert (out / "tests" / "test_composition.py").read_text() == TEST_COMPOSITION
assert not (out / "bricks").exists() and not (out / ".gitignore").exists()
before = {p: p.stat().st_mtime_ns for p in out.rglob("*") if p.is_file() and ".git" not in p.parts}
said = []
assert compose_init.init(["demo"], root=out, no_install=True, say=said.append) == 0
assert {p: p.stat().st_mtime_ns for p in out.rglob("*") if p.is_file() and ".git" not in p.parts} == before
assert sum("unchanged" in s for s in said) == 3
def test_an_edited_file_is_kept_unless_forced(world):
out = world["out"]
compose_init.init(["demo"], root=out, no_install=True, say=quiet)
(out / "compose.yaml").write_text("name: mine\n")
with pytest.raises(InitFailed, match="differs from what init writes"):
compose_init.init(["demo"], root=out, no_install=True, say=quiet)
assert (out / "compose.yaml").read_text() == "name: mine\n"
compose_init.init(["demo"], root=out, no_install=True, force=True, say=quiet)
assert "acme-demo" in (out / "compose.yaml").read_text()
def test_read_copy_clones_to_read_and_git_ignores_it(world):
out = world["out"]
assert compose_init.init(["demo"], root=out, no_install=True, read_copy=True, say=quiet) == 0
assert (out / "bricks" / "brick-demo" / "pyproject.toml").exists() and not (out / "bricks" / "erp-ui").exists()
assert (out / ".gitignore").read_text() == "bricks/\n"
assert "bricks/" not in (out / "requirements.txt").read_text() # nothing installs from the copy
compose_init.init(["demo"], root=out, no_install=True, read_copy=True, say=quiet)
assert (out / ".gitignore").read_text() == "bricks/\n"
def test_own_copies_the_brick_into_the_repo_pins_the_path_and_records_why(world):
out = world["out"]
compose_init.init(["demo"], root=out, no_install=True, say=quiet)
with pytest.raises(InitFailed, match="--reason is required"):
compose_init.own("demo", " ", root=out, say=quiet)
with pytest.raises(InitFailed, match="no pinned brick 'nope'"):
compose_init.own("nope", "CR-1", root=out, say=quiet)
assert compose_init.own("demo", "CR-3: a director co-signs every contract", root=out, say=quiet) == 0
owned = out / "owned" / "brick-demo"
assert (owned / "pyproject.toml").exists() and not (owned / ".git").exists() # tracked by the customer
assert "-e ./owned/brick-demo # demo (owned: upstream brick-demo@v0.2.0)" in (out / "requirements.txt").read_text()
spec = yaml.safe_load((out / "compose.yaml").read_text())
assert spec["bricks"][0]["owned"] == {"path": "owned/brick-demo", "upstream": "brick-demo@v0.2.0",
"reason": "CR-3: a director co-signs every contract"}
with pytest.raises(InitFailed, match="already the customer's own"):
compose_init.own("brick-demo", "again", root=out, say=quiet)
# a re-init keeps the ownership: same files, nothing refused, the pin stays the path
said = []
assert compose_init.init(["demo"], root=out, no_install=True, say=said.append) == 0
assert sum("unchanged" in s for s in said) == 3
from erp_platform import compose
assert compose.main(["own", "demo", "--reason", "x", "--dir", str(out)]) == 1
def test_a_clone_at_another_tag_is_refused(world):
out = world["out"]
subprocess.run(["git", "clone", "-q", world["plat_url"], out / "bricks" / "erp-platform"], check=True)
plat = out / "bricks" / "erp-platform"
(plat / "README.md").write_text("moved on\n")
subprocess.run([*GIT, "-C", plat, "commit", "-qam", "later"], check=True)
with pytest.raises(InitFailed, match="is not erp-platform at v0.6.0"):
compose_init.init(["demo"], root=out, no_install=True, read_copy=True, say=quiet)
def test_the_cli_exits_non_zero_with_the_reason(world, capsys):
from erp_platform import compose
assert compose.main(["init", "nope", "--dir", str(world["out"]), "--no-install"]) == 1
assert "no module 'nope'" in capsys.readouterr().err
assert compose.main(["init"]) == 2
def test_end_to_end_init_installs_checks_and_passes_the_composition_test(tmp_path, monkeypatch):
"""A real venv: this erp-platform (as a tagged repo) + brick-demo, then check and pytest — exit 0."""
src = tmp_path / "src"
plat = src / "erp-platform"
plat.mkdir(parents=True)
for f in ("pyproject.toml", "LICENSE", "NOTICE"):
shutil.copy(ROOT / f, plat / f)
shutil.copytree(ROOT / "src", plat / "src", ignore=shutil.ignore_patterns("__pycache__", "*.egg-info"))
plat_url = git_repo(plat, "v0.6.0")
demo_url = git_repo(demo_brick(src), "v0.2.0")
method = fake_method(tmp_path, {"demo": [repo("brick-demo", "brick", "v0.2.0", demo_url),
repo("erp-platform", "platform", "v0.6.0", plat_url)]}, ENTRY)
monkeypatch.setenv("ERP_METHOD_DIR", str(method))
out = tmp_path / "acme"
said = []
assert compose_init.init(["demo"], root=out, say=said.append) == 0, "\n".join(said)
assert any("Ready: demo 0.2.0" in s for s in said)
said.clear()
assert compose_init.init(["demo"], root=out, say=said.append) == 0 # idempotent: no reinstall
assert any("unchanged" in s and ".venv" in s for s in said)
def test_a_master_data_brick_carries_its_process_view_waiver_into_compose_yaml():
"""0.8.2: bp declares no process view; its brick_entries waiver must reach compose.yaml or check refuses."""
plan = compose_init.Plan(["hr"], "local", [
compose_init.Repo("erp-platform", "platform", "v0.8.2", "https://git.dodil.io/dodil/erp-platform.git"),
compose_init.Repo("brick-business-partner", "brick", "v0.3.1", "https://x/b.git",
{"name": "bp", "brick": "brick_bp.application.use_cases:BP",
"process_view": "none: master data only"})], [])
text = compose_init.compose_yaml(plan, "acme")
assert 'version: 0.3.1, process_view: "none: master data only" }' in text
from erp_platform import compose
assert compose._waivers(text) == {"bp": "master data only"}
def test_init_writes_each_modules_spec_and_the_methods_receipts_as_a_baseline_never_a_signature(world):
"""0.9.11: processes/<module>/module.yaml + processes/<module>/baseline/ (history too), byte for byte;
signoff/<module>/ stays empty — the customer's signature never comes from erp-method — so the gate refuses."""
from erp_platform import gate
m = world["method"]
d = m / "modules" / "demo"
(d / "signoff" / "demo" / "history").mkdir(parents=True)
(d / "module.yaml").write_text("id: demo\n")
(d / "signoff" / "demo" / "demo_item.json").write_text('{"bpmn_sha256": "abc"}\n')
(d / "signoff" / "demo" / "history" / "demo_item.v1.json").write_text('{"v": 1}\n')
idx = json.loads((m / "index.json").read_text())
idx["modules"][0]["spec"] = "modules/demo/module.yaml"
idx["files"] = ["modules/demo/module.yaml", "modules/demo/signoff/demo/demo_item.json",
"modules/demo/signoff/demo/history/demo_item.v1.json", "modules/other/x.json"]
(m / "index.json").write_text(json.dumps(idx))
out, said = world["out"], []
assert compose_init.init(["demo"], root=out, no_install=True, say=said.append) == 0
base = out / "processes" / "demo" / "baseline"
assert (out / "processes" / "demo" / "module.yaml").read_text() == "id: demo\n"
assert (base / "demo_item.json").read_bytes() == (d / "signoff" / "demo" / "demo_item.json").read_bytes()
assert (base / "history" / "demo_item.v1.json").exists()
assert not (out / "signoff").exists()
assert any("NOT a signature" in s and "signoff record --all" in s for s in said)
assert {r["state"] for r in gate.check(out, mods=["demo"])} == {"unsigned"}
assert not gate.enforce("erp-compose migrate", out, mods=["demo"], say=said.append) # a bucket refuses
assert gate.enforce("erp-compose migrate --local", out, mods=["demo"], refuse=False, say=said.append)