dodil/erp-brick-tax
DODIL ERP brick: tax
tests/conftest.py
"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the
permissions given, and the two doors — `api(...)` and `tool(...)`. Business 'today' is 2026-11-20 in every test."""
import base64
import json
import os
import pathlib
from contextlib import contextmanager
from datetime import datetime, timezone
from decimal import Decimal
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from erp_platform.contract import Recorder, checked_seams
from erp_platform.seam import derived_id
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_tax.app import app
from brick_tax.application import use_cases
from brick_tax.infrastructure import seams
from brick_tax.infrastructure.tables import Repository, md
# Every outgoing seam call is checked against the CALLEE's signed spec (erp-method). A missing checkout is an error, never a skip.
HERE = pathlib.Path(__file__).resolve().parent
METHOD_DIR = os.environ.get("ERP_METHOD_DIR") or str(HERE.parents[1] / "erp-method")
if not (pathlib.Path(METHOD_DIR) / "modules" / "shared.yaml").exists():
raise RuntimeError(f"seam contract checks need an erp-method checkout: set ERP_METHOD_DIR (tried {METHOD_DIR})")
checked_seams = checked_seams(caller="tax", method_dir=METHOD_DIR) # autouse: any real Seam call
_CONTRACT = Recorder()
@pytest.fixture(autouse=True)
def contract():
"""The fakes below replace whole ports (they never reach a Seam), so each checks its call against the callee's spec."""
_CONTRACT.violations.clear()
yield _CONTRACT
assert not _CONTRACT.violations, "\n".join(_CONTRACT.violations)
def spec_check(callee, method, path, body=None):
from erp_platform.contract import check
bad = check(callee, method, path, body, caller="tax", method_dir=METHOD_DIR)
if bad:
msg = f"tax -> {callee} {method} {path} breaks the callee's spec:\n " + "\n ".join(bad)
_CONTRACT.violations.append(msg)
raise AssertionError(msg)
@pytest.fixture(autouse=True)
def engine(tmp_path, monkeypatch):
e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False})
md.create_all(e)
with e.begin() as conn:
Repository(conn).seed()
@contextmanager
def repo():
with e.begin() as conn:
yield Repository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
yield e
e.dispose()
def person(email, permissions, roles=()):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": list(roles)}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
# The brick's DEFAULT ROLES (src/brick_tax/roles.yaml), as people: the permissions and no role name.
_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_tax" / "roles.yaml").read_text())["roles"]
def as_role(role, email=None):
return person(email or f"{role}@shop.example", _ROLES[role])
ADMIN, OWNER, BILLER, SYSTEM = (as_role(r) for r in ("tax_admin", "owner", "biller", "system"))
ADMIN2 = as_role("tax_admin", "second.admin@shop.example") # a second holder of tax:code:approve
BILLER2 = as_role("biller", "second.biller@shop.example")
NOBODY = person("nobody@shop.example", [])
async def _call(method, path, headers, body=None):
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://brick") as c:
if method == "GET":
return await c.get(path, headers=headers, params=body)
return await c.request(method, path, headers=headers, json=body)
def api(method, path, headers, body=None):
r = anyio.run(_call, method, path, headers, body)
return r.status_code, r.json()
async def _mcp(headers, fn):
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=headers, base_url="http://brick") as http:
async with streamable_http_client("http://brick/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return await fn(s)
def tool(headers, name, args):
res = anyio.run(_mcp, headers, lambda s: s.call_tool(name, args))
text = res.content[0].text
return ("refused", text.split(": ", 1)[1]) if res.is_error else ("ok", json.loads(text))
def tools(headers):
listed = anyio.run(_mcp, headers, lambda s: s.list_tools())
return sorted(t.name for t in listed.tools)
# ── a frozen business clock, and the neighbours' fakes ────────────────────────────────────────
TODAY = "2026-11-20"
@pytest.fixture(autouse=True)
def clock():
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@contextmanager
def on_day(when):
"""Run a block with the business clock at `when` (an ISO instant), and put it back."""
from erp_platform import work
before = work.now()
work.set_clock(at=datetime.fromisoformat(when), freeze=True)
try:
yield
finally:
work.set_clock(at=before, freeze=True)
class FakePartners:
"""business-partner's contract (NOT contract-checked: business-partner's signed spec lists no seam for tax, so there is no
entry to check GET /partners/{bp_id} or /migration-map against — reported as a spec gap): a customer the register knows, with the name and address it holds (or none)."""
def __init__(self):
self.known = {7311: {"legal_name": "Acme Builders Ltd", "address": "1 High Street, London"}}
def customer(self, bp_id, caller):
return self.known.get(int(bp_id))
def resolve_odoo(self, odoo_db, odoo_partner_id, caller):
"""business-partner's Odoo migration map: Odoo partner id -> bp_id."""
return {1: 101, 2: 102, 3: 103, 4: 104}.get(int(odoo_partner_id))
class FakeLedger:
"""gl's contract: sterling per unit of a currency on a date (or none)."""
def __init__(self):
self.rates = {"EUR": Decimal("0.85")}
self.posted, self.calls = {}, []
def fx_rate(self, currency, on):
spec_check("gl", "GET", f"/fx-rates/{currency}", {"on": on.isoformat()}) # 200 {rate: string}; 404 -> none
return self.rates.get(currency)
def post_journal(self, body):
"""GL-30: 201 {journal_id, entry_no, period_id, status: posted}; the same (source_document, event) answers 200 and the same id."""
spec_check("gl", "POST", "/journal-entry/journals", body)
key = (body["source_document"], body["event"])
self.posted.setdefault(key, {"journal_id": str(derived_id(*key)), "entry_no": f"J-{len(self.posted) + 1}",
"period_id": "2026-11", "status": "posted", "body": body})
self.calls.append(key)
return {k: v for k, v in self.posted[key].items() if k != "body"}
class FakeAr:
"""ar's AR-09: the approved write-offs from a date, each with the invoice's dates and total."""
def __init__(self):
self.rows, self.invoices = [], {}
def invoice(self, invoice_id):
"""AR-17: GET /invoices/{invoice_id} -> the frozen invoice with its lines (goods lines carry product_id and unit_code)."""
spec_check("ar", "GET", f"/invoices/{invoice_id}")
return self.invoices.get(invoice_id)
def writeoffs(self, approved_from):
spec_check("ar", "GET", "/writeoffs", {"status": "approved", "approved_from": approved_from.isoformat()})
return [r for r in self.rows if r["approved_at"][:10] >= approved_from.isoformat()]
class FakePayments:
"""payments' PAY-40 settlement and PAY-46 allocations, shaped as payments' spec answers them."""
def __init__(self):
self.outstanding, self.allocs, self.cash = {}, [], []
def cash_basis(self, frm, to):
"""PAY-46: GET /cash-basis?from=&to= -> {allocations: [...]}: receipts dated by received_on within the VAT period."""
spec_check("payments", "GET", "/cash-basis", {"from": frm.isoformat(), "to": to.isoformat()})
return [a for a in self.cash if frm.isoformat() <= a["received_on"] <= to.isoformat()]
def settlement(self, invoice_ids, as_at):
spec_check("payments", "GET", "/settlement", {"invoice_ids": ",".join(invoice_ids), "as_at": as_at.isoformat()})
known = [i for i in invoice_ids if i in self.outstanding]
return {"as_at": as_at.isoformat(), "invoices": [{"invoice_id": i, "currency": "GBP", "total": "1200.00",
"outstanding": self.outstanding[i], "state": "part_paid"} for i in known],
"unknown": [i for i in invoice_ids if i not in self.outstanding]}
def allocations(self, invoice_ids, frm):
spec_check("payments", "GET", "/allocations", {"invoice_ids": ",".join(invoice_ids), **({"from": frm.isoformat()} if frm else {})})
return [a for a in self.allocs if a["invoice_id"] in invoice_ids and (frm is None or a["allocated_on"] > frm.isoformat())]
@pytest.fixture(autouse=True)
def ar_and_payments(monkeypatch):
a, p = FakeAr(), FakePayments()
monkeypatch.setattr(seams, "receivables", a)
monkeypatch.setattr(seams, "payments", p)
return a, p
@pytest.fixture(autouse=True)
def neighbours(monkeypatch):
p, g = FakePartners(), FakeLedger()
monkeypatch.setattr(seams, "ledger_post", g)
monkeypatch.setattr(seams, "partners", p)
monkeypatch.setattr(seams, "ledger", g)
return p, g
def must(result):
"""A call that has to succeed: (status, body) -> body."""
status, body = result
assert status == 200, (status, body)
return body
def gates_of(body):
"""The unmet gates a refusal carries, parsed from its detail."""
detail = body["detail"] if isinstance(body, dict) else body
return json.loads(detail.split(" — unmet gates: ", 1)[1]) if " — unmet gates: " in detail else []
# ── builders the tests share ─────────────────────────────────────────────────────────────────
def load_uk(activate=True):
"""The UK pack, with every code activated by the tax admin."""
must(api("POST", "/codes/load-pack", SYSTEM, {}))
if activate:
for code in ("S", "R", "Z", "E", "OS", "RC", "NV"):
must(api("POST", f"/codes/{code}/activate-code", ADMIN, {}))
def registered(vat="GB123456789", since="2011-01-04"):
return must(api("POST", "/registration/record-registered", ADMIN, {"vat_number": vat, "effective_from": since}))
def uk_rules():
"""Four rules a UK service business needs, plus reverse charge for CIS contractors; first rows start today."""
for body in ({"category_code": "standard", "treatment_code": "uk_business", "tax_code": "S"},
{"category_code": "standard", "treatment_code": "overseas_business", "supply": "services", "tax_code": "OS"},
{"category_code": "construction_service", "treatment_code": "uk_cis_contractor", "tax_code": "RC"},
{"category_code": "reduced", "treatment_code": "any", "tax_code": "R"},
{"category_code": "zero", "treatment_code": "any", "tax_code": "Z"},
{"category_code": "exempt", "treatment_code": "any", "tax_code": "E"}):
must(api("POST", "/rules", ADMIN, body))
def customer(bp_id=7311, treatment="uk_business", vat=None):
return must(api("POST", f"/parties/{bp_id}/set-party-tax", BILLER, {"treatment_code": treatment, **({"vat_number": vat} if vat else {})}))
def setup(register=True):
"""Pack, registration, rules, and customer 7311 as a UK business: ready to tax a document."""
load_uk()
if register:
registered()
uk_rules()
customer()
def line(n, amount, category="standard", **kw):
return {"line_no": n, "description": kw.pop("description", f"item {n}"), "category_code": category, "amount": amount,
"quantity": kw.pop("quantity", "1"), "unit_price": kw.pop("unit_price", amount), **kw}
def compute(lines, tax_point=TODAY, who=SYSTEM, **kw):
return api("POST", "/documents/compute", who, {"lines": lines, "tax_point": tax_point, "bp_id": 7311, **kw})
FULL = {"invoice_number": "INV-2026-0001", "issue_date": TODAY, "supplier_name": "Dodil Ltd",
"supplier_address": "2 Low Road, Leeds", "customer_name": "Acme Builders Ltd", "customer_address": "1 High Street, London"}
# ── v3: the purchase side ────────────────────────────────────────────────────────────────────
def purchase_setup():
"""Sale side as `setup()`, plus the purchase codes PS (input 20%), PE (blocked: business entertainment, VAT a cost) and PRC
(purchase reverse charge: due 20% charged 0, output box 1, input box 4), their purchase rules, and two suppliers."""
setup()
must(api("POST", "/categories/entertainment", ADMIN, {"description": "business entertainment"}))
for body in ({"tax_code": "PS", "treatment": "standard", "rate_pct": "20", "valid_from": "2011-01-04", "reporting_box": "4",
"value_box": "7", "reclaimable_pct": "100"},
{"tax_code": "PE", "treatment": "standard", "rate_pct": "20", "valid_from": "2011-01-04", "reporting_box": "",
"value_box": "7", "reclaimable_pct": "0"},
{"tax_code": "PRC", "treatment": "reverse_charge", "rate_pct": "20", "charged_pct": "0", "valid_from": "2011-01-04",
"reporting_box": "1", "input_reporting_box": "4", "value_box": "7", "reclaimable_pct": "100",
"document_wording": "Reverse charge: we account for the VAT"}):
must(api("POST", "/codes", ADMIN, {**body, "direction": "purchase"}))
must(api("POST", f"/codes/{body['tax_code']}/activate-code", ADMIN, {}))
for body in ({"category_code": "standard", "treatment_code": "uk_business", "tax_code": "PS"},
{"category_code": "entertainment", "treatment_code": "uk_business", "tax_code": "PE"},
{"category_code": "standard", "treatment_code": "overseas_business", "supply": "services", "tax_code": "PRC"}):
must(api("POST", "/rules", ADMIN, {**body, "direction": "purchase"}))
customer(8800, "uk_business", "GB987654321")
customer(8801, "overseas_business", "DE811234567")
def compute_purchase(lines, ref="PB-1", bp_id=8800, tax_point=TODAY, who=SYSTEM, **kw):
return api("POST", "/documents/compute", who, {"lines": lines, "tax_point": tax_point, "bp_id": bp_id, "direction": "purchase",
"document_ref": ref, **kw})
# ── v4: HMRC, the VAT return, bad debt relief ────────────────────────────────────────────────
from erp_platform.seam import derived_id as _derived # noqa: E402,F401
from brick_tax.infrastructure.hmrc import HmrcError, HmrcTimeout # noqa: E402
OWNER2 = as_role("owner", "second.owner@shop.example")
SERVICE = {**person("svc@shop.example", ["tax:vat_return:submit"]), "X-Dodil-Auth-Source": "platform"} # a service client holding submit
BROWSER = {"Gov-Client-Browser-JS-User-Agent": "Mozilla/5.0", "Gov-Client-Device-ID": "dev-0001", "Gov-Client-Multi-Factor": "type=TOTP",
"Gov-Client-Public-IP": "203.0.113.7", "Gov-Client-Public-IP-Timestamp": "2027-01-12T10:00:00.000Z",
"Gov-Client-Public-Port": "50123", "Gov-Client-Screens": "width=1920&height=1080", "Gov-Client-Timezone": "UTC+00:00",
"Gov-Client-Window-Size": "width=1280&height=800"}
DECLARATION = ("When you submit this VAT information you are making a legal declaration that the information is true and complete. "
"A false declaration can result in prosecution.")
AFTER_26AD = "2027-01-12T09:00:00+00:00"
def hmrc_ob(key, start, end, due, status="O", received=None):
return {"periodKey": key, "start": start, "end": end, "due": due, "status": status, **({"received": received} if received else {})}
class FakeHmrc:
"""HMRC's VAT API as the sandbox answers it: obligations, a return per periodKey (DUPLICATE_SUBMISSION when sent twice),
liabilities and payments. A timeout can leave the return filed (HMRC got it) or not."""
def __init__(self):
self.obs = [hmrc_ob("26AC", "2026-07-01", "2026-09-30", "2026-11-07"), hmrc_ob("26AD", "2026-10-01", "2026-12-31", "2027-02-07")]
self.returns, self.submitted, self.log = {}, [], []
self.timeout_next, self.timeout_files, self.error_next, self.obligation_error = False, False, None, None
self.liabs, self.pays = [], []
def authorize_url(self, environment, state, redirect_uri):
return f"https://hmrc.test/{environment}/oauth/authorize?state={state}"
def exchange_code(self, environment, code, redirect_uri):
self.log.append(("exchange", environment))
return {"access_token": "AT-secret", "refresh_token": "RT-secret", "expires_in": 14400}
def obligations(self, environment, vrn, frm, to, tokens, headers):
self.log.append(("obligations", environment, vrn, frm.isoformat(), to.isoformat()))
if self.obligation_error:
raise HmrcError(self.obligation_error, "HMRC says no")
return [o for o in self.obs if frm.isoformat() <= o["end"] and o["start"] <= to.isoformat()]
def get_return(self, environment, vrn, period_key, tokens, headers):
self.log.append(("get_return", period_key))
return self.returns.get(period_key)
def submit_return(self, environment, vrn, body, tokens, headers):
self.log.append(("submit", body["periodKey"]))
key = body["periodKey"]
if self.timeout_next:
self.timeout_next = False
if self.timeout_files:
self.returns[key] = body
raise HmrcTimeout()
if self.error_next:
code, self.error_next = self.error_next, None
raise HmrcError(code, "HMRC refused")
if key in self.returns:
raise HmrcError("DUPLICATE_SUBMISSION", "already filed")
self.returns[key] = body
self.submitted.append((vrn, body, dict(headers)))
for o in self.obs:
if o["periodKey"] == key:
o["status"], o["received"] = "F", "2027-01-12"
return {"processingDate": "2027-01-12T10:00:00.000Z", "formBundleNumber": "256660290587", "paymentIndicator": "DD",
"chargeRefNumber": "aaaaaaaaaaaaaaaa"}
def liabilities(self, environment, vrn, frm, to, tokens, headers):
return list(self.liabs)
def payments(self, environment, vrn, frm, to, tokens, headers):
return list(self.pays)
class FakeSecrets:
def __init__(self):
self.store = {}
def put(self, ref, value):
self.store[ref] = value
def get(self, ref):
return self.store.get(ref)
@pytest.fixture(autouse=True)
def hmrc_world(monkeypatch):
h, sec = FakeHmrc(), FakeSecrets()
monkeypatch.setattr(seams, "hmrc", h)
monkeypatch.setattr(seams, "secrets", sec)
monkeypatch.setenv("VENDOR_PUBLIC_IP", "198.51.100.9")
monkeypatch.setenv("VENDOR_LICENSE_IDS", "dodil=0123456789abcdef")
return h, sec
def connect_hmrc(who=OWNER):
return must(api("POST", "/hmrc/connection/connect", who, {"code": "auth-code", "redirect_uri": "https://app.example/hmrc/callback"}))
def sync(frm="2026-01-01", to="2027-12-31"):
return must(api("POST", "/obligations/sync", SYSTEM, {"from_date": frm, "to_date": to}))
def filing_ready():
"""UK pack, registered GB123456789, the HMRC connection, and HMRC's obligations copied: ready to prepare 26AD."""
with on_day("2026-01-01T09:00:00+00:00"): # the rules' first rows start on the day they are made: earlier tax points need an earlier set-up
purchase_setup()
connect_hmrc()
sync()
def frozen_sale(ref, lines, bp_id=7311, tax_point=TODAY):
must(api("POST", "/documents/compute", SYSTEM, {"lines": lines, "tax_point": tax_point, "bp_id": bp_id, "document_ref": ref}))
return must(api("POST", f"/documents/{ref}/freeze", SYSTEM, {"contents": FULL}))
def frozen_purchase(ref, lines, bp_id=8800, tax_point=TODAY):
must(compute_purchase(lines, ref=ref, bp_id=bp_id, tax_point=tax_point))
bill = {"invoice_number": f"SUP-{ref}", "issue_date": tax_point, "supplier_name": "Acme Supplies Ltd", "supplier_address": "9 Dock Road, Hull",
"supplier_vat_number": "GB987654321", "customer_name": "Dodil Ltd"}
return must(api("POST", f"/documents/{ref}/freeze", SYSTEM, {"contents": bill}))
def prepare(period="26AD", who=ADMIN):
return api("POST", f"/vat-returns/{period}/prepare", who, {})
# ── v4 slice B: purchasing's supplier payments, product's commodity data, HMRC's VAT-number check and CIS channel ──────
class FakePurchasing:
"""PUR-66: GET /supplier-payments?status=&from=&to= -> {payments: [...]}: confirmed payments by paid_on."""
def __init__(self):
self.payments = []
def supplier_payments(self, frm, to):
spec_check("purchasing", "GET", "/supplier-payments", {"status": "confirmed", "from": frm.isoformat(), "to": to.isoformat()})
return [p for p in self.payments if p["status"] == "confirmed" and frm.isoformat() <= p["paid_on"] <= to.isoformat()]
class FakeProducts:
"""PRD-28: GET /products/{product_id}?on= -> commodity_code, country_of_origin and units[].measures."""
def __init__(self):
self.items = {}
def product(self, product_id, on):
spec_check("product", "GET", f"/products/{product_id}", {"on": on.isoformat()})
return self.items.get(product_id)
@pytest.fixture(autouse=True)
def purchasing_and_products(monkeypatch):
pur, prod = FakePurchasing(), FakeProducts()
monkeypatch.setattr(seams, "purchasing", pur)
monkeypatch.setattr(seams, "products", prod)
return pur, prod
def _hmrc_extras(h):
"""HMRC's VAT-number lookup and CIS channel on the fake: numbers it knows, verification results, CIS300 receipts."""
h.vat_numbers = {"555555555": {"valid": True, "name": "Brick Wall Contractors Ltd", "address": "5 Mortar Lane, Leeds",
"consultation_number": "CN-0001"}}
h.vat_down, h.cis_results, h.cis_submitted = False, {}, []
def check_vat_number(environment, vat_number):
if h.vat_down:
raise HmrcError("UNREACHABLE", "HMRC is down")
return h.vat_numbers.get(vat_number)
def verify_subcontractor(environment, body, tokens, headers):
return h.cis_results.get(body["utr"], {"status": "net", "verification_number": "V" + body["utr"]})
def submit_cis300(environment, body, tokens, headers):
h.cis_submitted.append(body)
return {"correlation_id": "CIS-" + body["tax_month_end"], "accepted_at": "2026-12-10T10:00:00Z"}
h.check_vat_number, h.verify_subcontractor, h.submit_cis300 = check_vat_number, verify_subcontractor, submit_cis300
@pytest.fixture(autouse=True)
def hmrc_v4_extras(hmrc_world):
_hmrc_extras(hmrc_world[0])