dodil/erp-platform

tests/test_lint_card.py

"""erp-lint: each rule has a hit and a clean case; erp-card: one smoke test."""
import textwrap

import pytest
import yaml

from erp_platform import card, lint


def brick(tmp_path, files: dict, tests: dict = None, shared: dict = None):
    root = tmp_path / "b"
    for rel, text in files.items():
        p = root / "src" / "pkg" / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(textwrap.dedent(text))
    for rel, text in (tests or {}).items():
        p = root / "tests" / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(textwrap.dedent(text))
    md = None
    if shared:
        md = tmp_path / "method"
        (md / "modules").mkdir(parents=True)
        (md / "modules" / "shared.yaml").write_text(yaml.safe_dump({"keys": shared}))
    return str(root), (str(md) if md else None)


def rules(tmp_path, files, **kw):
    root, md = brick(tmp_path, files, **kw)
    return sorted(f.rule for f in lint.lint(root, md))


CASES = {
    "L1": ("import datetime\nx = datetime.datetime.utcnow()\n", "from x import business_today\nx = business_today()\n"),
    "L2": ("def approve_order(o, a):\n    if o.email == a.email:\n        raise ValueError\n",
           "def approve_order(o, a):\n    if o.email == a.email or o.sub == a.sub:\n        raise ValueError\n"),
    "L3": ("def f(repo):\n    with repo.transaction() as tx:\n        try:\n            tx.execute(1)\n        except Exception:\n            pass\n        tx.execute(2)\n",
           "def f(repo):\n    with repo.transaction() as tx:\n        try:\n            tx.execute(1)\n        except Exception:\n            raise\n        tx.execute(2)\n"),
    "L4": ("def f(r):\n    return int(r['bp_key'])\n", "def f(r):\n    return int(r['bp_id'])\n"),
    "L5": ("def f(price, qty):\n    return price * qty\n", "def f(price, qty):\n    validate(price)\n    return price * qty\n"),
    "L6": ("SQL = 'CREATE TABLE t (id SERIAL PRIMARY KEY)'\n", "SQL = 'CREATE TABLE t (id BIGINT PRIMARY KEY)'  # never SERIAL\n"),
}


@pytest.mark.parametrize("rule", sorted(CASES))
def test_rule_hit_and_clean(tmp_path, rule):
    bad, good = CASES[rule]
    shared = {"bp_key": {"type": "VARCHAR"}, "bp_id": {"type": "BIGINT"}}
    assert rule in rules(tmp_path / "bad", {"m.py": bad}, shared=shared)
    assert rule not in rules(tmp_path / "good", {"m.py": good}, shared=shared)


def test_suppression_needs_matching_rule(tmp_path):
    src = "import datetime\nx = datetime.datetime.utcnow()  # erp-lint: allow L1 audit timestamp\n"
    assert rules(tmp_path / "a", {"m.py": src}) == []
    assert rules(tmp_path / "b", {"m.py": src.replace("L1", "L2")}) == ["L1"]


def test_l7_fake_without_contract(tmp_path):
    fake = "class FakeGl:\n    pass\n"
    assert rules(tmp_path / "a", {"m.py": "x = 1\n"}, tests={"conftest.py": fake}) == ["L7"]
    ok = "from erp_platform.contract import checked_seams\n_c = checked_seams('x')\n" + fake
    assert rules(tmp_path / "b", {"m.py": "x = 1\n"}, tests={"conftest.py": ok}) == []
    assert rules(tmp_path / "c", {"m.py": "x = 1\n"}, tests={"conftest.py": "class FakeOdoo:\n    pass\n"}) == []


def test_cli_exit_code(tmp_path, capsys):
    root, _ = brick(tmp_path, {"m.py": "import datetime\nx = datetime.datetime.utcnow()\n"})
    assert lint.main([root]) == 1
    assert "m.py:2: L1 utc-today" in capsys.readouterr().out


def test_card_smoke(tmp_path):
    spec = {"id": "demo", "version": 1, "maturity": "planned", "summary": "WHY PROSE", "permissions": {"namespace": "d", "catalog": ["d:x:read"]},
            "actors": [{"id": "clerk"}],
            "stories": [{"id": "D-01", "actor": "clerk", "story": "As a clerk I want", "given": "a", "when": "b", "then": "c"}],
            "workflows": [{"entity": "doc", "states": ["a", "b"], "terminal": ["b"],
                           "transitions": [{"key": "go", "from": "a", "to": "b", "by": "clerk", "gate": "g"}]}],
            "seams": [{"story": "D-01", "with": "gl", "owner": "gl", "direction": "outbound", "route": "POST /x",
                       "request": "{a: 1}", "response": "{b}", "rule": "r"}],
            "entities": [{"id": "doc", "natural_key": "doc_id", "reference_table": "d_doc", "master_data": "owned"}],
            "sap_parity": [{"sap": "S1", "ours": "covered"}, {"sap": "S2", "ours": "different", "why": "w"}]}
    p = tmp_path / "module.yaml"
    p.write_text(yaml.safe_dump(spec))
    out = tmp_path / "c.md"
    assert card.main([str(p), "-o", str(out)]) == 0
    t = out.read_text()
    assert "D-01 (clerk): GIVEN a WHEN b THEN c" in t and "request: {a: 1}" in t and "go: a > b by clerk; gate: g" in t
    assert "S2: different" in t and "S1" not in t and "WHY PROSE" not in t and "erp-lint" in t


def test_card_stories_slice(tmp_path):
    spec = {"id": "m", "version": 1, "maturity": "planned", "permissions": {"namespace": "m", "catalog": ["m:a:read", "m:b:write"]},
            "actors": [{"id": "u", "permissions": ["m:a:read"]}],
            "stories": [{"id": f"M-0{i}", "actor": "u", "given": "g", "when": f"touches widget{i}", "then": "t"} for i in range(1, 5)],
            "workflows": [], "jobs": [{"story": "M-02", "trigger": "t", "runs": "r", "idempotent": "yes"},
                                      {"story": "M-04", "trigger": "t", "runs": "r", "idempotent": "yes"}],
            "seams": [{"story": s, "with": "x", "owner": "x", "direction": "out", "route": f"GET /{s}", "request": "q",
                       "response": "r", "rule": "r"} for s in ("M-02", "M-03", "M-04")],
            "entities": [{"id": "widget2", "natural_key": "k", "reference_table": "m_w2", "relationships": ["gadget"]},
                         {"id": "gadget", "natural_key": "k", "reference_table": "m_g"},
                         {"id": "widget4", "natural_key": "k", "reference_table": "m_w4"}]}
    p = tmp_path / "module.yaml"
    p.write_text(yaml.safe_dump(spec))
    full = card.build(spec)
    cut = card.build(spec, None, "M-01..M-02,M-03")
    assert "M-04" in full and "M-04" not in cut and "M-02" in cut and "M-03" in cut
    assert "GET /M-02" in cut and "GET /M-04" not in cut and "widget4" not in cut
    assert "key k; table m_g" in cut                      # one hop of relationships
    assert "SLICE:" in cut and len(cut) < len(full)
    with pytest.raises(SystemExit):
        card.build(spec, None, "M-09")
    assert card.parse_stories("M-02..M-04", ["M-01", "M-02", "M-03", "M-04"]) == ["M-02", "M-03", "M-04"]


def test_every_module_spec_in_an_erp_method_checkout_renders_a_card():
    # older specs omit `direction`, `key` and `idempotent` and mix strings with mappings in lists; a card must still render
    import os
    from pathlib import Path

    import yaml

    from erp_platform.card import build
    root = Path(os.environ.get("ERP_METHOD_DIR", Path(__file__).resolve().parents[2] / "erp-method")) / "modules"
    specs = sorted(root.glob("*/module.yaml"))
    if not specs:
        import pytest
        pytest.skip("no erp-method checkout next to erp-platform")
    for f in specs:
        assert build(yaml.safe_load(f.read_text())).startswith("#"), f


def test_a_story_range_is_by_number_not_by_file_position():
    from erp_platform.card import parse_stories
    known = ["SO-02", "SO-29", "SO-30", "SO-12", "SO-27", "SO-01", "SO-25", "SO-26"]   # out of order, as a grown spec is
    assert parse_stories("SO-25..SO-30", known) == ["SO-29", "SO-30", "SO-27", "SO-25", "SO-26"]
    assert parse_stories("SO-01,SO-12", known) == ["SO-12", "SO-01"]