"""erp-compose scaffold 0.9.1, against the first real scaffolded brick's spec (erp-method projects/spec,
modules/projects/module.yaml, copied as tests/fixtures/scaffold/projects.module.yaml). Each test is a defect
erp-brick-projects found in 0.9.0."""
import pathlib
import shutil
import subprocess
import sys
import pytest
import yaml
from erp_platform import scaffold
SPEC = pathlib.Path(__file__).parent / "fixtures" / "scaffold" / "projects.module.yaml"
TEXT = SPEC.read_text()
SRC = "src/brick_projects"
def generate(text=TEXT):
return scaffold.generate(yaml.safe_load(text), text=text)
def test_settings_are_one_typed_row_per_setting_in_the_specs_table():
files, s = generate()
kinds = dict(s["settings"])
assert kinds["deemed_acceptance"] == "bool" and kinds["chase_days_blocked"] == "int"
assert kinds["review_window_warning"] == "int" and kinds["single_payment_line"] == "bool"
assert [k for k, v in kinds.items() if v == "mode"] == [
"ai.ai_draft_plan.mode", "ai.ai_chase_brief.mode", "ai.ai_flag_shortfall.mode"]
assert "state" not in kinds and "setting" not in kinds # pj_state is a config table, not a setting
schema = files[f"{SRC}/schema.sql"]
assert "CREATE TABLE IF NOT EXISTS pj_setting (" in schema and "pj_settings " not in schema
assert files[f"{SRC}/migrations/00001_init/up.sql"].count("INSERT INTO pj_setting ") == len(kinds)
def test_the_optimistic_version_is_the_specs_version_column():
files, _ = generate()
schema = files[f"{SRC}/schema.sql"]
assert "row_version" not in schema and "PRIMARY KEY (project_id, version)" in schema
assert "PRIMARY KEY (item_id, version)" in schema
item = schema.split("CREATE TABLE IF NOT EXISTS pj_item (")[1].split(");")[0]
assert item.count(" version ") == 1 and item.count(" revision ") == 1 # revision stays the business field
def test_config_and_history_are_no_writable_aggregates_and_no_permission_is_invented():
files, s = generate()
text = "".join(files.values())
assert 'permission="TODO(permission)' not in text
assert "pj:state:write" not in text and "pj:event:write" not in text
assert "keep_event" not in text and "keep_state" not in text and "pj_event_versions" not in text
uc = files[f"{SRC}/application/use_cases.py"]
assert 'route="POST /states/{state_id}", permission="pj:settings:write"' in uc
assert (s["configs"], s["histories"]) == (["pj_state"], ["pj_event"])
tables = files[f"{SRC}/infrastructure/tables.py"]
assert "_append_history(self.conn, \"item\"" in tables and "conn.execute(insert(pj_event), rows)" in tables
def test_a_permission_the_catalog_lacks_is_a_todo_no_one_holds():
text = TEXT.replace(" - pj:settings:write\n", "", 1)
files, _ = generate(text)
uc = files[f"{SRC}/application/use_cases.py"]
assert 'permission="TODO(permission):pj:settings:write"' in uc
def test_an_unquoted_off_is_off_and_an_unquoted_on_is_refused_with_its_line():
files, _ = generate(TEXT.replace('modes_allowed: ["off", suggest]', "modes_allowed: [off, suggest]"))
ai = files[f"{SRC}/infrastructure/ai_steps.py"]
assert '"modes_allowed": ["off", "suggest"]' in ai and '"False"' not in ai
bad = TEXT.replace('modes_allowed: ["off", suggest]', "modes_allowed: [on, suggest]", 1)
line = bad.splitlines().index(" modes_allowed: [on, suggest]") + 1
with pytest.raises(scaffold.ScaffoldError, match=rf"ai_draft_plan: modes_allowed True is not a mode .*line {line}\)"):
generate(bad)
def test_ai_steps_are_wired_to_the_ai_capabilities():
files, s = generate()
ai = files[f"{SRC}/infrastructure/ai_steps.py"]
for step, cap in (("ai_draft_plan", "ai.suggest"), ("ai_chase_brief", "ai.draft_message"),
("ai_flag_shortfall", "ai.flag_risk")):
assert f'"{step}": CapabilitySeam("{cap}")' in ai
assert "AUDIT = (" in ai and "def propose(" in ai and s["todos"]["ai"] == 1
def test_the_scaffolded_brick_runs_its_probes_and_its_scenario_catalogue(tmp_path):
pytest.importorskip("brick_bp", reason="brick-business-partner is not installed")
out = tmp_path / "brick-projects"
assert scaffold.main([str(SPEC), "--out", str(out)]) == 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:]
xfail = r.stdout
assert "test_the_catalogue_runs_in_process PASSED" in xfail or "failed" not in xfail.replace("xfailed", "")
for gap in ("test_a_command_under_a_terminal_parent_is_refused", "test_the_requester_may_not_approve",
"test_the_permission_matrix", "test_the_transition_that_names_a_seam_calls_it"):
assert f"XFAIL tests/test_probes.py::{gap}" in xfail, gap # the gaps CI lists
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:]