"""erp-compose scaffold: a module spec → a brick skeleton that installs, passes its own tests and composes.
The fixture is templates/brick-example written as a spec (tests/fixtures/scaffold/quote.module.yaml), so the
generated brick can be held against the starter brick's shape.
"""
import pathlib
import shutil
import subprocess
import sys
import pytest
import yaml
from erp_platform import bpmn, compose, scaffold
SPEC = pathlib.Path(__file__).parent / "fixtures" / "scaffold" / "quote.module.yaml"
TEMPLATE = pathlib.Path(__file__).resolve().parents[1] / "templates" / "brick-example"
def generate(**kw):
return scaffold.generate(yaml.safe_load(SPEC.read_text()), **kw)
def test_the_same_spec_gives_the_same_bytes(tmp_path):
(a, sa), (b, sb) = generate(), generate()
assert a == b and sa == sb
assert scaffold.main([str(SPEC), "--out", str(tmp_path / "one")]) == 0
assert scaffold.main([str(SPEC), "--out", str(tmp_path / "two")]) == 0
one = {p.relative_to(tmp_path / "one"): p.read_bytes() for p in (tmp_path / "one").rglob("*") if p.is_file()}
two = {p.relative_to(tmp_path / "two"): p.read_bytes() for p in (tmp_path / "two").rglob("*") if p.is_file()}
assert one == two and len(one) == sa["files"]
def test_an_edited_file_is_never_overwritten_without_force(tmp_path, capsys):
out = tmp_path / "brick"
assert scaffold.main([str(SPEC), "--out", str(out)]) == 0
assert scaffold.main([str(SPEC), "--out", str(out)]) == 0 # unchanged files: a re-run is a no-op
rules = out / "src" / "brick_example" / "domain" / "quote.py"
rules.write_text(rules.read_text() + "\n# the rule I wrote\n")
readme = out / "README.md"
readme.unlink()
assert scaffold.main([str(SPEC), "--out", str(out)]) == 1
assert "the rule I wrote" in rules.read_text() and not readme.exists() # refused: NOTHING written
assert "src/brick_example/domain/quote.py" in capsys.readouterr().err
assert scaffold.main([str(SPEC), "--out", str(out), "--force"]) == 0
assert "the rule I wrote" not in rules.read_text() and readme.exists()
def test_the_process_is_the_specs():
files, _ = generate()
process = yaml.safe_load(files["src/brick_example/application/process.yaml"])
# byte-identical but for the file name each says it was generated from
assert bpmn.to_bpmn(bpmn.from_module(process, SPEC.name)) == bpmn.generate(SPEC)
def test_the_shape_is_brick_examples():
files, s = generate()
template = {str(p.relative_to(TEMPLATE)) for p in TEMPLATE.rglob("*") if p.is_file()
and not {"__pycache__", ".pytest_cache"} & set(p.parts) and ".egg-info" not in str(p)}
assert template - set(files) == set(), "every file of the starter brick has its generated counterpart"
assert (s["entities"], s["aggregates"], s["stories"], s["xfail"]) == (2, 1, 3, 3)
schema = files["src/brick_example/schema.sql"]
for t in ("example_quotes", "example_lines", "example_quotes_versions", "example_settings"):
assert f"CREATE TABLE IF NOT EXISTS {t} (" in schema
ddl = "\n".join(ln for ln in schema.splitlines() if not ln.startswith("--"))
assert "unit_price NUMERIC(18,2)" in ddl and "SERIAL" not in ddl
assert "example_schema_migrations" in files["src/brick_example/migrations/00001_init/up.sql"]
uc = files["src/brick_example/application/use_cases.py"]
assert 'route="POST /quotes/{quote_id}/send-quote", permission="example:quote:send"' in uc
assert 'if_match="version"' in uc and 'EXAMPLE.declare_process("process.yaml")' in uc
# 0.9.7: the commit-time unique race → 409 comes with every brick (erp-brick-ar's _on_engine), and its test.
assert "def _on_engine(engine):" in uc and "with _on_engine(db.engine) as repo:" in uc
assert "def test_a_unique_race_lost_at_commit_is_409" in files["tests/test_doors.py"]
def test_an_ai_step_is_a_capability_seam_in_suggest_mode():
files, s = generate()
ai = files["src/brick_example/infrastructure/ai_steps.py"]
assert 'CapabilitySeam("mail.send")' in ai and '"mode": "suggest"' in ai and '"safe_for_auto": False' in ai
assert "no step-mode API yet" in ai and s["todos"]["ai"] == 1
def test_an_unknown_entity_is_refused():
with pytest.raises(scaffold.ScaffoldError, match="not an entity"):
generate(entities=["invoice"])
def test_the_scaffolded_brick_passes_its_own_tests_and_composes(tmp_path):
out = tmp_path / "brick-example"
assert scaffold.main([str(SPEC), "--out", str(out), "--dist", "erp-brick-example"]) == 0
r = subprocess.run([sys.executable, "-m", "pytest", "-q", "--tb=short", "-p", "no:cacheprovider",
"-W", "ignore::DeprecationWarning", "--rootdir", ".", "-c", "pyproject.toml", "-rx"],
cwd=out, capture_output=True, text=True, timeout=300)
assert r.returncode == 0, r.stdout[-3000:] + r.stderr[-2000:]
assert "failed" not in r.stdout.replace("xfailed", "")
assert r.stdout.count("XFAIL tests/test_rules.py") == 3 # one per story; the probes' xfails besides
sys.path.insert(0, str(out / "src"))
try:
report = compose.check(compose.load(out / "tests" / "tier1.compose.yaml"))
finally:
sys.path.remove(str(out / "src"))
for name in [n for n in sys.modules if n == "brick_example" or n.startswith("brick_example.")]:
del sys.modules[name]
assert report["problems"] == [] and [b["name"] for b in report["bricks"]] == ["example", "business-partner"]
if shutil.which("ruff"):
lint = subprocess.run(["ruff", "check", "--no-cache", "src", "tests"], cwd=out, capture_output=True, text=True)
assert lint.returncode == 0, lint.stdout[-2000:]