dodil/erp-platform

tests/test_local_tier.py

"""0.9.9 — papercuts from the 2026-10-25 small-ERP dry run: every erp-compose verb answers --help, init writes the
`modules:` the gate reads, and the local (SQLite) dev tier of migrate/serve."""
import json
import pathlib
import shutil
import sys
import types

import pytest

from erp_platform import compose, gate, local, work
from erp_platform.bricks import Brick

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE / "scenario_bricks"))


@pytest.fixture
def shop(tmp_path, monkeypatch):
    monkeypatch.delenv("BUCKET", raising=False)
    monkeypatch.delenv("DODIL_ENV", raising=False)
    shutil.copy(HERE / "scenarios" / "compose.yaml", tmp_path / "compose.yaml")
    yield tmp_path
    work.reset_clock()


# ── 1. --help ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("verb", ["check", "plan", "migrate", "serve"])
@pytest.mark.parametrize("flag", ["--help", "-h"])
def test_every_verb_answers_help_with_its_usage(verb, flag, capsys):
    assert compose.main([verb, flag]) == 0
    assert capsys.readouterr().out.startswith(f"usage: erp-compose {verb}")


def test_bare_help_prints_the_whole_usage(capsys):
    assert compose.main(["--help"]) == 0
    out = capsys.readouterr().out
    assert "erp-compose init" in out and "--local" in out


# ── 2. init writes modules: ───────────────────────────────────────────────────────────────

def test_init_writes_the_modules_it_composed_and_the_gate_reads_them(tmp_path):
    from erp_platform import compose_init
    plan = compose_init.Plan(["invoicing", "projects"], "v0.6.0", [], [])
    (tmp_path / "compose.yaml").write_text(compose_init.compose_yaml(plan, "acme"))
    assert gate.modules(tmp_path) == ["invoicing", "projects"]
    rows = gate.check(tmp_path, None, gate.modules(tmp_path))
    assert {r["module"] for r in rows} == {"invoicing", "projects"}        # not "(none)"


# ── 3. the local tier ─────────────────────────────────────────────────────────────────────

def test_migrate_local_creates_sqlite_files_and_warns_instead_of_refusing(shop, capsys):
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 0
    data = shop / local.DEFAULT_DIR
    assert {p.name for p in data.glob("*.db")} >= {"orders.db", "stock.db", "mailer.db", "_platform.db"}
    assert json.loads((data / local.MARKER).read_text())["tier"] == "local (dev)"
    assert (data / ".gitignore").read_text().splitlines()[-1] == "*"       # never committed
    cap = capsys.readouterr()
    assert "LOCAL DEV TIER" in cap.out and "WARNING" in cap.err
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 0   # re-runnable


def test_migrate_local_takes_a_directory(shop):
    assert compose.main(["migrate", "--local", str(shop / "here"), str(shop / "compose.yaml")]) == 0
    assert (shop / "here" / local.MARKER).exists()


def test_migrate_without_local_still_refuses_an_unsigned_design(shop, capsys):
    assert compose.main(["migrate", str(shop / "compose.yaml")]) == 1
    assert "REFUSED" in capsys.readouterr().err


@pytest.mark.parametrize("env", [{"BUCKET": "acme-prod"}, {"DODIL_ENV": "prod"}])
def test_local_is_refused_when_a_bucket_or_a_live_env_is_named(shop, monkeypatch, capsys, env):
    for k, v in env.items():
        monkeypatch.setenv(k, v)
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 1
    assert "--local refused" in capsys.readouterr().err
    assert not (shop / local.DEFAULT_DIR).exists()


def test_serve_local_refuses_an_unmigrated_dir(shop, capsys):
    assert compose.main(["serve", "--local", str(shop / "compose.yaml")]) == 1
    assert "migrate --local" in capsys.readouterr().err


def test_serve_local_binds_loopback_mounts_the_clock_and_starts_it(shop, monkeypatch, capsys):
    import uvicorn
    from fastapi.testclient import TestClient
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 0
    seen = {}

    def fake_run(app, host, port, **kw):
        seen.update(host=host, now=work.now())
        with TestClient(app) as c:
            seen["clock"] = c.get("/orders/demo-clock", headers={"X-Dodil-User": json.dumps({"sub": "u"})})
    monkeypatch.setattr(uvicorn, "run", fake_run)
    before = work.now()
    assert compose.main(["serve", "--local", "--clock", "+1d", str(shop / "compose.yaml")]) == 0
    assert seen["host"] == "127.0.0.1"
    assert (seen["now"] - before).total_seconds() > 86000                 # the business day is tomorrow
    assert seen["clock"].status_code == 200
    assert "/orders/demo-clock" in capsys.readouterr().out


def test_local_flags_only_where_they_mean_something(shop):
    y = str(shop / "compose.yaml")
    assert compose.main(["check", "--local", y]) == 2
    assert compose.main(["serve", "--clock", "+1d", y]) == 2
    assert compose.main(["serve", "--local", "--public", y]) == 2


def test_the_clock_goes_on_a_brick_that_already_mounts_it_else_the_first():
    a, b = Brick("lt_a"), Brick("lt_b")
    comp = types.SimpleNamespace(members=[types.SimpleNamespace(brick=a), types.SimpleNamespace(brick=b)])
    assert local.clock_brick(comp) == (a, True)
    from erp_platform import democlock
    democlock.mount(b)
    assert local.clock_brick(comp) == (b, False)


# ── 0.9.11: serve --local runs the bricks' scheduled passes ───────────────────────────────

def test_the_passes_are_what_a_brick_declares_else_what_looks_like_one():
    from erp_platform import local as lt
    b = Brick("lt_ticks", machine_permissions=("lt_ticks:job:run",))

    @b.use_case(kind="command", route="POST /post-pending", permission="lt_ticks:job:run", idempotent=True)
    def post_pending(caller: dict, limit: int = 10) -> dict:
        """Post what is pending."""
        return {}

    @b.use_case(kind="command", route="POST /import", permission="lt_ticks:job:run", idempotent=True)
    def import_catalogue(caller: dict) -> dict:        # a machine command with no input, but no pass
        """Import."""
        return {}

    @b.use_case(kind="command", route="POST /x/{x_id}/finish", permission="lt_ticks:job:run", idempotent=True)
    def finish_x(caller: dict, x_id: str) -> dict:     # needs an id: never a pass
        """Finish one."""
        return {}
    comp = types.SimpleNamespace(members=[types.SimpleNamespace(name="lt", brick=b)])
    assert [u.name for _m, u in lt.tick_use_cases(comp)] == ["post_pending"]
    b.ticks = ("import_catalogue",)                    # Brick(ticks=…) is exactly what runs
    assert [u.name for _m, u in lt.tick_use_cases(comp)] == ["import_catalogue"]
    b.ticks = ()
    assert lt.tick_use_cases(comp) == []


def test_did_counts_work_and_not_what_was_already_done():
    assert local.did({"posted": 0, "results": [{"document_id": "a", "already": True}]}) == ""
    assert local.did({"as_of": "2026-10-25", "raised": ["rec:1:1"], "already_raised": 0, "failed": []}) == "raised=1"
    assert local.did({"posted": 2, "results": [{"posted": True}, {"posted": True}]}) == "posted=2 results=2"


def test_a_tick_runs_the_pass_as_the_clock_and_logs_once_per_change(shop):
    from erp_platform.process_export import send, use_cases
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 0
    root = shop
    comp = compose.load(str(shop / "compose.yaml"))
    app, st = local.prepare_serve(comp, local.data_dir(root, None), say=lambda *_: None)
    with st:
        r = send(app, use_cases(comp), local.owner_identity(comp), "place_order",
                 {"order_id": "T1", "customer": "acme", "lines": [{"sku": "S", "quantity": 1, "unit_price": "5"}]})
        assert r.status_code == 200, r.text
        lines = []
        t = local.Ticker(comp, app, 5, say=lines.append)
        assert [u.name for _m, u in t.passes] == ["tick"]
        assert t.once() == []                                      # nothing due yet: no line
        work.set_clock(offset=__import__("datetime").timedelta(days=30))
        assert len(t.once()) == 1 and "orders tick: overdue=1" in lines[0]
        assert t.once() == []                                      # done: no second line
        o = send(app, use_cases(comp), local.owner_identity(comp), "get_order", {"order_id": "T1"}).json()
        assert o["status"] == "overdue"


def test_serve_local_starts_the_ticker_unless_no_ticks(shop, monkeypatch, capsys):
    import uvicorn
    assert compose.main(["migrate", "--local", str(shop / "compose.yaml")]) == 0
    seen = []
    monkeypatch.setattr(uvicorn, "run", lambda app, **kw: seen.append(
        [t.name for t in __import__("threading").enumerate() if t.name == local.THREAD]))
    assert compose.main(["serve", "--local", "--ticks-every", "0.2", str(shop / "compose.yaml")]) == 0
    assert "scheduled work: every 0.2s as erp-compose.clock — orders tick" in capsys.readouterr().out
    assert compose.main(["serve", "--local", "--no-ticks", str(shop / "compose.yaml")]) == 0
    assert "scheduled work: off (--no-ticks)" in capsys.readouterr().out
    assert seen == [[local.THREAD], []]
    assert not [t for t in __import__("threading").enumerate() if t.name == local.THREAD]   # stopped with serve
    y = str(shop / "compose.yaml")
    assert compose.main(["serve", "--no-ticks", y]) == 2 and compose.main(["serve", "--local", "--ticks-every", "0", y]) == 2