dodil/erp-brick-tax
DODIL ERP brick: tax
tests/live_check.py
"""ONE live check: the brick on a real DataK3 bucket (scratch `w2taxlab`, org dodil), with only a CLI login.
dodil data bucket create w2taxlab # once
BUCKET=w2taxlab DODIL_ORG=dodil python tests/live_check.py
It applies migrations 00001, 00002 and 00003 itself (ADD COLUMN included), then runs over the pg wire: the UK pack, a frozen sale, a
return PREPARED (an HMRC fake), REFRESHED in place (an UPDATE, no savepoint, no delete-then-insert), a CLAIM COLLISION between two real
transactions (one commits, the other must fail), a CIS deduction and its replay, a capital item and its replay, and the MCP tool count.
HMRC is never contacted. Not collected by pytest. The token is the CLI login's and is never printed."""
import pathlib
import sys
from datetime import date, timedelta
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "src"))
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from conftest import ADMIN, BILLER, FULL, OWNER, SYSTEM, FakeHmrc, FakeSecrets, api, hmrc_ob, line, tool, tools # noqa: E402
from erp_platform import livecheck as lc # noqa: E402
from brick_tax.application import use_cases # noqa: E402
from brick_tax.infrastructure import seams # noqa: E402
from brick_tax.infrastructure.tables import Repository, md # noqa: E402
SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_tax"
TABLES = [t.name for t in md.sorted_tables]
engine = lc.engine()
def apply_migrations():
lc.drop_all(TABLES)
with engine.begin() as c:
for m in ("00001_init", "00002_audit1", "00003_v4"):
sql = "\n".join(ln for ln in (SRC / "migrations" / m / "up.sql").read_text().splitlines() if not ln.strip().startswith("--"))
for stmt in [s.strip() for s in sql.split(";") if s.strip()]:
c.exec_driver_sql(stmt)
print(f"migrations 00001..00003 applied: {len(TABLES)} tables expected")
def repo():
from contextlib import contextmanager
@contextmanager
def r():
with engine.begin() as conn:
yield Repository(conn)
return r()
use_cases.repository = repo
apply_migrations()
today = date.today()
q0 = date(today.year, 3 * ((today.month - 1) // 3) + 1, 1)
q1 = date(q0.year + (q0.month == 10), (q0.month + 2) % 12 + 1, 1) - timedelta(days=1)
FULL = {**FULL, "issue_date": today.isoformat()}
def ok(name, result, expect=200):
status, body = result
lc.check(name, status == expect, (status, str(body)[:300]))
return body if status == expect else {}
try:
h = FakeHmrc()
h.obs = [hmrc_ob("LIVEQ", q0.isoformat(), q1.isoformat(), (q1 + timedelta(days=37)).isoformat())]
seams.hmrc, seams.secrets = h, FakeSecrets()
ok("load the UK pack", api("POST", "/codes/load-pack", SYSTEM, {}))
for code in ("S", "R", "Z", "E", "OS", "RC", "NV"):
ok(f"activate {code}", api("POST", f"/codes/{code}/activate-code", ADMIN, {}))
ok("register", api("POST", "/registration/record-registered", ADMIN, {"vat_number": "GB123456789", "effective_from": "2011-01-04"}))
ok("a rule", api("POST", "/rules", ADMIN, {"category_code": "standard", "treatment_code": "uk_business", "tax_code": "S"}))
ok("a customer", api("POST", "/parties/7311/set-party-tax", BILLER, {"treatment_code": "uk_business"}))
ok("compute", api("POST", "/documents/compute", SYSTEM, {"lines": [line(1, "1000.00")], "tax_point": today.isoformat(), "bp_id": "7311", "document_ref": "LIVE-1"}))
ok("freeze", api("POST", "/documents/LIVE-1/freeze", SYSTEM, {"contents": FULL}))
ok("connect HMRC (fake)", api("POST", "/hmrc/connection/connect", OWNER, {"code": "c", "redirect_uri": "https://app.example/cb"}))
ok("sync obligations", api("POST", "/obligations/sync", SYSTEM, {"from_date": q0.isoformat(), "to_date": q1.isoformat()}))
ret = ok("prepare the return", api("POST", "/vat-returns/LIVEQ/prepare", ADMIN, {}))
lc.check("box 1 is 200.00", ret.get("boxes", {}).get("1") == "200.00", ret.get("boxes"))
ref = ok("refresh in place (UPDATE)", api("POST", "/vat-returns/LIVEQ/refresh", ADMIN, {}))
lc.check("the refresh keeps the boxes", ref.get("boxes") == ret.get("boxes"))
# a claim collision between two REAL transactions: exactly one wins
row = {"document_ref": "RACE-1", "tax_code": "S", "entry_kind": "document", "vrn": "123456789", "period_key": "LIVEQ", "direction": "sale",
"tax_point": today, "net_amount": 1, "vat_amount": 0, "box1_amount": 0, "box4_amount": 0, "box_value": "6", "source": "document"}
a, b = engine.connect(), engine.connect()
ta, tb = a.begin(), b.begin()
Repository(a).entries.claim(row)
failed = False
try:
Repository(b).entries.claim({**row, "period_key": "OTHER"})
tb.commit()
except Exception as exc: # a Conflict, or the commit's 23505
failed = True
tb.rollback()
print("loser:", type(exc).__name__)
try: # the OTHER racer may be the loser: DataK3 checks the key at commit
ta.commit()
except Exception as exc:
failed = True
ta.rollback()
print("loser:", type(exc).__name__)
a.close()
b.close()
with engine.begin() as c:
held = Repository(c).entries.where(document_ref="RACE-1")
lc.check("exactly one claim won", len(held) == 1 and failed, (len(held), failed))
ok("add a CIS subcontractor", api("POST", "/cis/subcontractors", ADMIN, {"bp_id": 8801, "name": "Bright Sparks Ltd", "utr": "1234567890", "company_number": "01234567"}))
ok("verify (person)", api("POST", "/cis/subcontractors/8801/verify", ADMIN, {"status": "net", "verification_number": "V1234567890"}))
d = {"payment_ref": "LIVE-P1", "bp_id": 8801, "paid_on": today.isoformat(), "bill_ids": ["B1"], "gross_ex_vat": "5000.00", "materials": "1000.00"}
first = ok("CIS deduction", api("POST", "/cis/deductions", SYSTEM, d))
again = ok("CIS deduction replay", api("POST", "/cis/deductions", SYSTEM, d))
lc.check("20% of the labour, replayed", first.get("deduction") == "800.00" and again.get("replay") is True, again)
item = {"item_ref": "LIVE-BLD", "kind": "land_building", "bought_on": "2026-06-01", "net_value": "500000.00", "input_vat": "100000.00", "first_pct": "80"}
ok("capital item", api("POST", "/capital-goods", ADMIN, item))
lc.check("capital item replay", ok("capital item again", api("POST", "/capital-goods", ADMIN, item)).get("replay") is True)
names = tools(SYSTEM)
print(f"mcp: {len(names)} tools")
lc.check("mcp lists the v4 tools", {"compute_cis_deduction", "prepare_vat_return", "change_vat_scheme"} <= set(names), names[:5])
kind, got = tool(SYSTEM, "get_cis_subcontractor", {"bp_id": 8801})
lc.check("read through MCP", kind == "ok", got)
finally:
raise SystemExit(lc.finish(TABLES))