dodil/erp-brick-tax
DODIL ERP brick: tax
tests/test_slice_b.py
"""v4 slice B: CIS (TAX-38 .. TAX-41), trade declarations (TAX-42, 43), import VAT and the PVA statement (TAX-44, 45), the construction
reverse charge (TAX-46), pinned lines and the place of supply (TAX-47), the VAT-number check (TAX-13 v4), the AI review of a return, the
new settings and who may do what."""
import json
from decimal import Decimal
from types import SimpleNamespace
from conftest import (
ADMIN,
BILLER,
FULL,
OWNER,
SERVICE,
SYSTEM,
api,
compute,
compute_purchase,
connect_hmrc,
customer,
filing_ready,
frozen_purchase,
frozen_sale,
gates_of,
hmrc_ob,
line,
load_uk,
must,
on_day,
prepare,
purchase_setup,
registered,
sync,
uk_rules,
)
from brick_tax.infrastructure import ai_steps
D = Decimal
DEC10 = "2026-12-10T09:00:00+00:00"
def box(out, n):
return D(out["boxes"][str(n)])
# ═══ CIS ═════════════════════════════════════════════════════════════════════════════════════
def cis_world():
load_uk()
registered()
connect_hmrc()
must(api("POST", "/settings/cis_contractor_utr", ADMIN, {"value": "9876543210"}))
must(api("POST", "/cis/subcontractors", ADMIN, {"bp_id": 8801, "name": "Bright Sparks Ltd", "utr": "1234567890", "company_number": "01234567"}))
def deduct(ref="PAY-1", bp=8801, paid="2026-11-20", gross="5000.00", materials="1000.00"):
return api("POST", "/cis/deductions", SYSTEM, {"payment_ref": ref, "bp_id": bp, "paid_on": paid, "bill_ids": ["B-1"],
"gross_ex_vat": gross, "materials": materials})
def test_tax_38_a_payment_to_an_unverified_subcontractor_answers_the_cis_verified_gate_then_hmrc_verifies(hmrc_world):
cis_world()
status, body = deduct()
assert status == 422 and gates_of(body)[0] == {"gate": "cis_verified", "met": False, "fix": "CIS > Subcontractors > Verify",
"reason": "Bright Sparks Ltd has not been verified with HMRC"}
got = must(api("GET", "/cis/subcontractors/8801", SYSTEM))
assert got["readiness"][0]["met"] is False and got["status"] is None
v = must(api("POST", "/cis/subcontractors/8801/verify", ADMIN, {}))
assert (v["status"], v["rate_pct"], v["verification_number"]) == ("net", "20.0000", "V1234567890") or v["status"] == "net"
assert v["verified_on"] == "2026-11-20" and v["readiness"][0]["met"] is True
def test_tax_38_a_person_may_enter_what_they_read_from_hmrc_and_a_reverification_is_a_new_dated_row(hmrc_world):
cis_world()
status, _ = api("POST", "/cis/subcontractors/8801/verify", ADMIN, {"status": "unmatched"})
assert status == 409 # no verification number
v = must(api("POST", "/cis/subcontractors/8801/verify", ADMIN, {"status": "unmatched", "verification_number": "V1234567890A"}))
assert v["status"] == "unmatched" and D(str(v["rate_pct"])) == 30
assert api("POST", "/cis/subcontractors/8801/verify", ADMIN, {"status": "net", "verification_number": "V1"})[0] == 409 # not edited the same day
with on_day("2026-11-21T09:00:00+00:00"):
r = must(api("POST", "/cis/subcontractors/8801/reverify", ADMIN, {"status": "net", "verification_number": "V1234567890B"}))
assert D(str(r["rate_pct"])) == 20 and r["verified_on"] == "2026-11-21"
assert must(api("GET", "/cis/subcontractors/8801", SYSTEM, {"on": "2026-11-20"}))["status"] == "unmatched" # the old row still answers its dates
assert api("GET", "/cis/subcontractors/9999", SYSTEM)[0] == 404
def test_tax_38_one_already_paid_in_the_last_two_tax_years_needs_no_verification(hmrc_world, engine):
from brick_tax.infrastructure.tables import Repository
cis_world()
with engine.begin() as conn:
Repository(conn).cis_deductions.put({"payment_ref": "OLD", "bp_id": 8801, "tax_month": "2025-06-05", "paid_on": __import__("datetime").date(2025, 6, 1),
"bill_ids": "B", "gross_ex_vat": D(100), "materials": D(0), "labour": D(100), "status": "net",
"rate_pct": D(20), "deduction": D(20), "state": "computed"})
out = must(deduct())
assert out["status"] == "unmatched" and D(str(out["rate_pct"])) == 30 # no HMRC verification on file: the 30% rate
def test_tax_39_the_deduction_is_hmrcs_rate_on_the_labour_only_and_a_retry_adds_nothing(hmrc_world):
cis_world()
must(api("POST", "/cis/subcontractors/8801/verify", ADMIN, {}))
out = must(deduct())
assert D(str(out["deduction"])) == D("800.00") and D(str(out["labour"])) == D("4000.00") # 20% x (5,000.00 - 1,000.00 materials)
assert out["tax_month"] == "2026-12-05" and out["status"] == "net" and out["replay"] is False
again = must(deduct())
assert again["replay"] is True and D(str(again["deduction"])) == D("800.00")
assert api("POST", "/cis/deductions", SYSTEM, {"payment_ref": "PAY-2", "bp_id": 8801, "paid_on": "2026-11-20", "bill_ids": ["B"],
"gross_ex_vat": "100.00", "materials": "500.00"})[0] == 422 # materials above the bill's own figure
def test_tax_39_a_gross_subcontractor_is_deducted_at_zero_and_still_recorded(hmrc_world):
cis_world()
hmrc_world[0].cis_results["1234567890"] = {"status": "gross", "verification_number": "V1234567890G"}
must(api("POST", "/cis/subcontractors/8801/verify", ADMIN, {}))
out = must(deduct("PAY-G"))
assert D(str(out["deduction"])) == 0 and out["status"] == "gross"
def test_tax_39_a_reversal_is_one_row_and_refused_once_the_tax_month_is_submitted(hmrc_world):
cis_world()
must(api("POST", "/cis/subcontractors/8801/verify", ADMIN, {}))
must(deduct("PAY-A"))
must(deduct("PAY-B", paid="2026-11-22"))
body = {"bp_id": 8801, "reason": "payment rejected by the bank", "reversed_on": "2026-11-15"}
r = must(api("POST", "/cis/deductions/PAY-A/reverse", SYSTEM, body))
assert r["status"] == "reversed" and must(api("POST", "/cis/deductions/PAY-A/reverse", SYSTEM, body))["status"] == "reversed"
with on_day(DEC10):
out = must(api("POST", "/cis/returns/2026-12-05/prepare", ADMIN, {}))
assert D(str(out["total_deduction"])) == D("800.00") and len(out["lines"]) == 1 # the reversed one no longer counts
must(api("POST", "/cis/returns/2026-12-05/submit", OWNER, {"employment_status_declared": True, "verification_declared": True}))
status, refused = api("POST", "/cis/deductions/PAY-B/reverse", SYSTEM, {**body, "reversed_on": "2026-12-11"})
assert status == 409 and gates_of(refused)[0]["gate"] == "cis_return_open"
def test_tax_40_the_monthly_return_lists_each_subcontractor_a_person_submits_and_statements_go_out(hmrc_world):
cis_world()
must(api("POST", "/cis/subcontractors", ADMIN, {"bp_id": 8802, "name": "Volt Ltd", "utr": "5555555555", "ni_number": "QQ123456C"}))
for bp in (8801, 8802):
must(api("POST", f"/cis/subcontractors/{bp}/verify", ADMIN, {}))
must(deduct("P1"))
must(deduct("P2", paid="2026-11-25", gross="2000.00", materials="0.00"))
must(deduct("P3", bp=8802, paid="2026-12-01", gross="1000.00", materials="200.00"))
status, body = api("POST", "/cis/returns/2026-12-05/prepare", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "tax_month_ended" # the month has not ended
with on_day(DEC10):
out = must(api("POST", "/cis/returns/2026-12-05/prepare", ADMIN, {}))
by = {str(x["bp_id"]): x for x in out["lines"]}
assert D(str(by["8801"]["deduction"])) == D("1200.00") and D(str(by["8801"]["gross_ex_vat"])) == D("7000.00")
assert by["8802"]["utr"] == "5555555555" and by["8802"]["verification_number"] == "V5555555555" and D(str(out["total_deduction"])) == D("1360.00")
assert out["due_date"] == "2026-12-19" and out["pay_hmrc_by"] == "2026-12-22" and out["gates"] == []
assert api("POST", "/cis/returns/2026-12-05/submit", OWNER, {"employment_status_declared": True, "verification_declared": False})[0] == 422
assert api("POST", "/cis/returns/2026-12-05/submit", SERVICE, {"employment_status_declared": True, "verification_declared": True})[0] == 403
done = must(api("POST", "/cis/returns/2026-12-05/submit", OWNER, {"employment_status_declared": True, "verification_declared": True}))
assert done["lifecycle"] == "submitted" and done["receipt"]["correlation_id"] == "CIS-2026-12-05"
assert len(hmrc_world[0].cis_submitted) == 1
sent = must(api("POST", "/cis/returns/2026-12-05/send-statements", OWNER, {}))
assert len(sent["statements"]) == 2 and D(sent["statements"][0]["labour"]) == D("6000.00")
assert api("POST", "/cis/returns/2026-12-05/prepare", ADMIN, {})[0] == 409 # a submitted return never changes
def test_tax_40_a_month_with_no_deductions_is_a_nil_return_and_an_overdue_one_answers_the_penalty_gate(hmrc_world):
cis_world()
with on_day("2026-12-25T09:00:00+00:00"):
out = must(api("POST", "/cis/returns/2026-12-05/prepare", ADMIN, {}))
assert out["nil_return"] is True and out["lines"] == [] and out["gates"][0]["gate"] == "cis300_on_time" and out["gates"][0]["met"] is False
must(api("POST", "/settings/cis_contractor_utr", ADMIN, {"value": None})) # not a contractor any more
with on_day("2027-01-10T09:00:00+00:00"):
status, body = api("POST", "/cis/returns/2027-01-05/prepare", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "cis_contractor_utr"
SUFFERED = {"contractor_bp_id": 7311, "statement_ref": "CIS-ST-9", "tax_month": "2026-12-05", "invoice_id": "INV-77", "payment_id": "PAY-77",
"gross": "2000.00", "materials": "0.00", "deduction": "400.00"}
def test_tax_41_what_a_contractor_deducted_from_us_is_kept_once_per_statement_and_summed_per_month_and_year(hmrc_world):
first = must(api("POST", "/cis/suffered", SYSTEM, SUFFERED))
assert first == {"recorded": True, "replay": False}
assert must(api("POST", "/cis/suffered", SYSTEM, SUFFERED)) == {"recorded": True, "replay": True}
must(api("POST", "/cis/suffered", SYSTEM, {**SUFFERED, "statement_ref": "CIS-ST-10", "tax_month": "2027-01-05", "deduction": "100.00"}))
got = must(api("GET", "/cis/suffered", ADMIN))
assert D(str(got["per_tax_month"]["2026-12-05"])) == 400 and D(str(got["per_tax_year"]["2026/2027"])) == 500
def test_tax_41_a_deduction_above_thirty_percent_of_the_labour_is_refused():
status, body = api("POST", "/cis/suffered", SYSTEM, {**SUFFERED, "statement_ref": "BAD", "deduction": "700.00"})
assert status == 422 and gates_of(body)[0]["gate"] == "within_cis_rate"
assert api("POST", "/cis/suffered", ADMIN, SUFFERED)[0] == 403 # a machine permission
# ═══ trade declarations ══════════════════════════════════════════════════════════════════════
def eu_world(hmrc_world, ni=True):
filing_ready()
hmrc_world[0].obs += [hmrc_ob("27AA", "2027-01-01", "2027-03-31", "2027-05-07"), hmrc_ob("27AD", "2027-10-01", "2027-12-31", "2028-02-07")]
sync(to="2028-12-31")
must(api("POST", "/categories/eu_goods", ADMIN, {"description": "goods sold to the EU"}))
must(api("POST", "/treatments/eu_business", ADMIN, {"description": "an EU business", "needs_vat_number": True}))
must(api("POST", "/codes", ADMIN, {"tax_code": "EUG", "treatment": "zero", "rate_pct": "0", "valid_from": "2011-01-04", "reporting_box": "6",
"value_box": "8", "eu_goods": True}))
must(api("POST", "/codes/EUG/activate-code", ADMIN, {}))
must(api("POST", "/rules", ADMIN, {"category_code": "eu_goods", "treatment_code": "eu_business", "tax_code": "EUG"}))
customer(7501, "eu_business", "IE1234567T")
customer(7502, "eu_business", "FR12345678901")
must(api("POST", "/registration/change-scheme", ADMIN, {"scheme": "standard", "effective_from": "2027-01-01", "ni_goods": ni}))
def eu_sale(ref, amount, bp, tax_point="2027-11-05"):
price = amount.lstrip("-")
return frozen_sale(ref, [line(1, amount, category="eu_goods", supply="goods", unit_price=price)], bp_id=bp, tax_point=tax_point)
def test_tax_42_the_ec_sales_list_has_a_line_per_customer_vat_number_and_ties_to_box_8(hmrc_world):
eu_world(hmrc_world)
eu_sale("INV-IE", "12000.00", 7501)
eu_sale("INV-FR", "3000.00", 7502)
eu_sale("CN-FR", "-500.00", 7502, "2027-11-20")
with on_day("2027-12-10T09:00:00+00:00"):
out = must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "ec_sales_list", "period": "2027-11"}))
assert [(x["partner_vat_number"], D(str(x["value"])), x["indicator"]) for x in out["lines"]] == [
("FR12345678901", 2500, "0"), ("IE1234567T", 12000, "0")]
assert D(str(out["total"])) == 14500 and out["file"].startswith("VAT number,Value,Indicator\nFR12345678901,2500,0\n")
tie = {g["gate"]: g for g in out["gates"]}["ec_sales_ties_box8"]
assert tie["met"] is False # no return holds these months yet: a gate, never silent
status, body = api("POST", "/trade-declarations/file", ADMIN, {"kind": "ec_sales_list", "period": "2027-11", "submission_ref": "ECSL-1"})
assert status == 409 and gates_of(body)[0]["gate"] == "ec_sales_ties_box8"
ret = must(prepare("27AD"))
assert box(ret, 8) == 14500 and box(ret, 6) == 14500 # box 8 is part of box 6
out = must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "ec_sales_list", "period": "2027-11"}))
assert {g["gate"]: g["met"] for g in out["gates"]}["ec_sales_ties_box8"] is True
filed = must(api("POST", "/trade-declarations/file", ADMIN, {"kind": "ec_sales_list", "period": "2027-11", "submission_ref": "ECSL-1"}))
assert filed["lifecycle"] == "filed" and filed["submission_ref"] == "ECSL-1" and filed["filed_on"] == "2027-12-10"
eu_sale("INV-IE2", "100.00", 7501, "2027-11-25")
amended = must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "ec_sales_list", "period": "2027-11"}))
assert amended["version"] == 2 and amended["lifecycle"] == "draft" # an amendment is a new version
assert must(api("GET", "/trade-declarations/ec_sales_list/2027-11", ADMIN))["version"] == 2
def test_tax_42_it_needs_an_ni_goods_registration_and_a_quarterly_list_needs_goods_below_the_limit(hmrc_world):
eu_world(hmrc_world, ni=False)
status, body = api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "ec_sales_list", "period": "2027-11"})
assert status == 409 and gates_of(body)[0]["gate"] == "ni_goods"
must(api("POST", "/registration/change-scheme", ADMIN, {"scheme": "standard", "effective_from": "2027-10-01", "ni_goods": True}))
eu_sale("INV-BIG", "40000.00", 7501, "2027-10-15") # above 35,000 in the quarter
with on_day("2028-01-10T09:00:00+00:00"):
out = must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "ec_sales_list", "period": "2027-Q4"}))
assert {g["gate"]: g["met"] for g in out["gates"]}["quarterly_allowed"] is False
def intrastat_world(hmrc_world, ar_and_payments, purchasing_and_products, big="300000.00"):
eu_world(hmrc_world)
eu_sale("INV-D1", big, 7501)
ar_and_payments[0].invoices["INV-D1"] = {"invoice_id": "INV-D1", "bp_id": 7501, "company_code": "UK01", "currency": "GBP", "issue_date": "2027-11-05",
"tax_point_date": "2027-11-05", "status": "issued", "tax": {}, "totals": {},
"lines": [{"line_no": 1, "product_id": "P-1", "description": "chargers", "quantity": "100", "unit_code": "EA",
"net_amount": "200000.00", "tax_code": "EUG"},
{"line_no": 2, "product_id": "P-2", "description": "cables", "quantity": "50", "unit_code": "EA",
"net_amount": "100000.00", "tax_code": "EUG"}]}
purchasing_and_products[1].items.update({
"P-1": {"product_id": "P-1", "commodity_code": "85044090", "country_of_origin": "CN",
"units": [{"unit_code": "EA", "measures": {"net_weight": "0.5", "weight_unit_code": "kg"}}]},
"P-2": {"product_id": "P-2", "commodity_code": None, "country_of_origin": "CN",
"units": [{"unit_code": "EA", "measures": {"net_weight": "200", "weight_unit_code": "g"}}]}})
def test_tax_43_intrastat_lines_come_from_ar_and_products_and_a_missing_commodity_code_is_a_gate(hmrc_world, ar_and_payments, purchasing_and_products):
intrastat_world(hmrc_world, ar_and_payments, purchasing_and_products)
with on_day("2027-12-10T09:00:00+00:00"):
out = must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "intrastat_dispatches", "period": "2027-11"}))
by = {x["commodity_code"]: x for x in out["lines"]}
p1 = by["85044090"]
assert (p1["partner_country"], p1["nature"], p1["origin_country"]) == ("IE", "11", "CN") and D(str(p1["value"])) == 200000
assert D(str(p1["net_mass_kg"])) == 50 and D(str(p1["supplementary_units"])) == 100 and p1["partner_vat_number"] == "IE1234567T"
assert D(str(by[""]["net_mass_kg"])) == 10 # 50 x 200 g
assert out["gates"] == [{"gate": "commodity_code", "met": False, "reason": "product P-2 has no commodity code", "fix": "Product > the product > Trade"}]
status, body = api("POST", "/trade-declarations/file", ADMIN, {"kind": "intrastat_dispatches", "period": "2027-11", "submission_ref": "IS-1"})
assert status == 409 and gates_of(body)[0]["gate"] == "commodity_code"
purchasing_and_products[1].items["P-2"]["commodity_code"] = "85444290"
must(api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "intrastat_dispatches", "period": "2027-11"}))
# the line count is unchanged but the code is now known: the draft is rebuilt in place
got = must(api("GET", "/trade-declarations/intrastat_dispatches/2027-11", ADMIN))
assert sorted(x["commodity_code"] for x in got["lines"]) == ["85044090", "85444290"] and got["gates"] == []
assert must(api("POST", "/trade-declarations/file", ADMIN, {"kind": "intrastat_dispatches", "period": "2027-11", "submission_ref": "IS-1"}))["lifecycle"] == "filed"
def test_tax_43_below_the_dispatches_threshold_no_declaration_is_due(hmrc_world, ar_and_payments, purchasing_and_products):
intrastat_world(hmrc_world, ar_and_payments, purchasing_and_products, big="100000.00")
with on_day("2027-12-10T09:00:00+00:00"):
status, body = api("POST", "/trade-declarations/prepare", ADMIN, {"kind": "intrastat_dispatches", "period": "2027-11"})
assert status == 409 and gates_of(body)[0]["gate"] == "intrastat_threshold"
# ═══ import VAT ══════════════════════════════════════════════════════════════════════════════
def imp(mrn, vat="2000.00", value="10000.00", method="postponed", entry="2026-11-05"):
return api("POST", "/imports", SYSTEM, {"mrn": mrn, "entry_date": entry, "bill_ref": "PB-IMP", "bp_id": 8800, "customs_value": value,
"import_vat": vat, "method": method})
def test_tax_44_postponed_vat_is_output_and_input_and_nets_to_zero_with_its_value_in_box_7_not_6(hmrc_world):
filing_ready()
out = must(imp("26GB12345678901234"))
assert (out["tax_code"], D(str(out["box_vat_output"])), D(str(out["box_vat_input"])), D(str(out["box_value"]))) == ("PVA", 2000, 2000, 10000)
assert must(imp("26GB12345678901234"))["replay"] is True # keyed by MRN: recorded once
ret = must(prepare())
assert box(ret, 1) == 2000 and box(ret, 4) == 2000 and box(ret, 5) == 0 and box(ret, 7) == 10000 and box(ret, 6) == 0
def test_tax_44_import_vat_paid_is_reclaimed_only_once_the_certificate_lists_the_entry(hmrc_world):
filing_ready()
paid = must(imp("26GB99900000000001", method="paid"))
assert paid["tax_code"] == "IMPORT_PAID" and paid["certificate_listed"] is False
assert box(must(prepare()), 4) == 0
must(api("POST", "/imports/26GB99900000000001/certificate-listed", ADMIN, {}))
out = must(api("POST", "/vat-returns/26AD/refresh", ADMIN, {}))
assert box(out, 4) == 2000 and box(out, 1) == 0 and box(out, 7) == 10000
assert api("POST", "/imports", ADMIN, {"mrn": "X" * 8, "entry_date": "2026-11-05", "bill_ref": "B", "bp_id": 1, "customs_value": "1.00",
"import_vat": "0.20", "method": "paid"})[0] == 403
def statement(lines=None):
return api("POST", "/pva-statements", ADMIN, {"month": "2026-11", "file_text": "MRN,VAT\nA,2000\n", "lines": lines or [
{"mrn": "MRN-AAAAA", "import_vat": "2000.00"}, {"mrn": "MRN-BBBBB", "import_vat": "1200.00"}, {"mrn": "MRN-CCCCC", "import_vat": "500.00"}]})
def test_tax_45_each_statement_line_is_matched_by_mrn_and_the_return_uses_hmrcs_figures(hmrc_world):
filing_ready()
must(imp("MRN-AAAAA"))
must(imp("MRN-BBBBB", vat="1000.00", value="5000.00"))
st = must(statement())
got = {x["mrn"]: x for x in st["lines"]}
assert (got["MRN-AAAAA"]["status"], got["MRN-BBBBB"]["status"], got["MRN-CCCCC"]["status"]) == ("matched", "differs", "missing")
assert D(str(got["MRN-BBBBB"]["difference"])) == 200 and D(str(got["MRN-BBBBB"]["our_vat"])) == 1000 and st["file"].startswith("MRN,VAT")
assert statement()[0] == 409 # one statement per month
status, body = api("POST", "/pva-statements/2026-11/reconcile", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "pva_lines_resolved"
for mrn in ("MRN-BBBBB", "MRN-CCCCC"):
must(api("POST", "/pva-statements/2026-11/explain", ADMIN, {"mrn": mrn, "reason": "HMRC's figure is right: the agent mis-keyed it"}))
assert must(api("POST", "/pva-statements/2026-11/reconcile", ADMIN, {}))["lifecycle"] == "reconciled"
ret = must(prepare())
assert box(ret, 1) == 3700 and box(ret, 4) == 3700 and box(ret, 5) == 0 # 2,000 + 1,000 + 200 + 500, both sides
def test_tax_45_the_return_cannot_lock_until_the_periods_statements_are_reconciled(hmrc_world):
from conftest import AFTER_26AD
filing_ready()
must(imp("MRN-AAAAA"))
must(prepare())
with on_day(AFTER_26AD):
status, body = api("POST", "/vat-returns/26AD/lock", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "pva_reconciled" and "not imported" in gates_of(body)[0]["reason"]
must(statement([{"mrn": "MRN-AAAAA", "import_vat": "2000.00"}]))
assert "not reconciled" in gates_of(api("POST", "/vat-returns/26AD/lock", ADMIN, {})[1])[0]["reason"]
must(api("POST", "/pva-statements/2026-11/reconcile", ADMIN, {}))
assert api("POST", "/vat-returns/26AD/lock", ADMIN, {})[0] == 200
# ═══ the construction reverse charge ═════════════════════════════════════════════════════════
def test_tax_46_an_end_user_notice_takes_the_construction_lines_out_of_the_reverse_charge_from_its_date(hmrc_world):
filing_ready()
must(api("POST", "/rules", ADMIN, {"category_code": "construction_service", "treatment_code": "uk_business", "tax_code": "S"}))
customer(7400, "uk_cis_contractor", "GB111222333")
status, body = api("POST", "/parties/7400/set-party-tax", ADMIN, {"treatment_code": "uk_business", "valid_from": "2026-11-21", "notice_kind": "end_user"})
assert status == 409 and gates_of(body)[0]["gate"] == "notice_evidence"
row = must(api("POST", "/parties/7400/set-party-tax", ADMIN, {"treatment_code": "uk_business", "valid_from": "2026-11-21", "notice_kind": "end_user",
"notice_reference": "EU-77", "notice_date": "2026-11-18"}))
assert row["notice"] == {"kind": "end_user", "reference": "EU-77", "date": "2026-11-18"}
build = [line(1, "1000.00", category="construction_service")]
assert must(compute(build, bp_id=7400) if False else api("POST", "/documents/compute", SYSTEM, {"lines": build, "tax_point": "2026-11-20", "bp_id": 7400}))["lines"][0]["tax_code"] == "RC"
after = must(api("POST", "/documents/compute", SYSTEM, {"lines": build, "tax_point": "2026-11-21", "bp_id": 7400}))["lines"][0]
assert after["tax_code"] == "S" and D(str(after["rate_pct"])) == 20 and D(str(after["vat"])) == 200
def test_tax_46_a_purchase_of_construction_services_is_self_accounted_output_and_input_with_net_in_box_7(hmrc_world):
filing_ready()
must(api("POST", "/codes", ADMIN, {"tax_code": "RC_PURCHASE", "treatment": "reverse_charge", "direction": "purchase", "rate_pct": "20",
"charged_pct": "0", "valid_from": "2021-03-01", "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/RC_PURCHASE/activate-code", ADMIN, {}))
must(api("POST", "/rules", ADMIN, {"category_code": "construction_service", "treatment_code": "uk_cis_contractor", "tax_code": "RC_PURCHASE",
"direction": "purchase"}))
customer(8802, "uk_cis_contractor", "GB333444555")
frozen_purchase("PB-CIS", [line(1, "4000.00", category="construction_service")], bp_id=8802)
ret = must(prepare())
assert box(ret, 1) == 800 and box(ret, 4) == 800 and box(ret, 7) == 4000 and box(ret, 6) == 0 and box(ret, 5) == 0
# ═══ pinned lines, place of supply, self-billing ═════════════════════════════════════════════
def test_tax_47_a_pinned_line_takes_the_earlier_documents_code_and_rate_not_todays(hmrc_world):
filing_ready()
frozen_sale("DEP-2026-0003", [line(1, "5000.00")]) # S 20%
must(api("POST", "/codes/S/schedule-rate", ADMIN, {"rate_pct": "21", "valid_from": "2027-01-01"}))
pinned = line(4, "-2000.00", category=None, unit_price="2000.00", description="deposit deducted", pinned_from="DEP-2026-0003", tax_code="S")
out = must(compute([line(1, "10000.00"), pinned], tax_point="2027-04-15", document_ref="FINAL-1"))
by = {x["line_no"]: x for x in out["lines"]}
assert D(str(by[1]["rate_pct"])) == 21 and D(str(by[1]["vat"])) == 2100
assert D(str(by[4]["rate_pct"])) == 20 and D(str(by[4]["vat"])) == D("-400.00") and by[4]["rule"] == "pinned to DEP-2026-0003"
assert by[4]["pinned_from"] == "DEP-2026-0003" and by[4]["source"] == "pinned"
def test_tax_47_a_code_the_earlier_document_does_not_carry_and_a_pin_beyond_its_net_are_refused(hmrc_world):
filing_ready()
frozen_sale("DEP-1", [line(1, "5000.00")])
status, body = compute([line(1, "-100.00", category=None, unit_price="100.00", pinned_from="DEP-1", tax_code="Z")], document_ref="F0")
assert status == 409 and gates_of(body)[0]["gate"] == "pinned_code"
status, body = compute([line(1, "-100.00", category=None, unit_price="100.00", pinned_from="NOPE", tax_code="S")], document_ref="F0")
assert status == 409 and gates_of(body)[0]["gate"] == "pinned_document"
must(compute([line(1, "-4000.00", category=None, unit_price="4000.00", pinned_from="DEP-1", tax_code="S")], document_ref="F1"))
must(api("POST", "/documents/F1/freeze", SYSTEM, {"contents": {**FULL, "invoice_number": "INV-F1"}}))
status, body = compute([line(1, "-2000.00", category=None, unit_price="2000.00", pinned_from="DEP-1", tax_code="S")], document_ref="F2")
assert status == 409 and gates_of(body)[0]["gate"] == "pinned_within_original" # 4,000 + 2,000 is more than 5,000
assert must(compute([line(1, "-1000.00", category=None, unit_price="1000.00", pinned_from="DEP-1", tax_code="S")], document_ref="F3"))["lines"][0]["vat"] == "-200.00"
def test_tax_47_ship_to_country_decides_the_place_of_supply_of_goods_lines(hmrc_world):
filing_ready()
must(api("POST", "/rules", ADMIN, {"category_code": "standard", "treatment_code": "overseas_business", "supply": "goods", "tax_code": "Z"}))
goods, services = line(1, "100.00", supply="goods"), line(2, "100.00", supply="services")
plain = must(compute([goods, services]))
assert [x["tax_code"] for x in plain["lines"]] == ["S", "S"]
abroad = must(api("POST", "/documents/compute", SYSTEM, {"lines": [goods, services], "tax_point": TODAY_, "bp_id": 7311, "ship_to_country": "fr"}))
assert [x["tax_code"] for x in abroad["lines"]] == ["Z", "S"] # only the goods follow where they go
home = must(api("POST", "/documents/compute", SYSTEM, {"lines": [goods], "tax_point": TODAY_, "bp_id": 7311, "ship_to_country": "GB"}))
assert home["lines"][0]["tax_code"] == "S"
TODAY_ = "2026-11-20"
def test_tax_47_a_self_billed_purchase_is_checked_against_our_self_billing_series(hmrc_world):
purchase_setup()
must(compute_purchase([line(1, "100.00")], ref="SB-1", self_billed=True))
out = must(api("POST", "/documents/SB-1/check-contents", SYSTEM, {"direction": "purchase", "contents": {}}))
g = {x["gate"]: x for x in out["gates"]}
assert g["invoice_number"]["met"] is False and "self-billing series" in g["invoice_number"]["reason"]
must(compute_purchase([line(1, "100.00")], ref="PB-N"))
plain = must(api("POST", "/documents/PB-N/check-contents", SYSTEM, {"direction": "purchase", "contents": {}}))
assert "self-billing" not in {x["gate"]: x for x in plain["gates"]}["invoice_number"]["reason"]
# ═══ the VAT-number check (TAX-13 v4) ════════════════════════════════════════════════════════
def vat_world():
load_uk()
registered()
uk_rules()
customer(7312, "uk_cis_contractor", "GB555555555")
class FakeAI:
def __init__(self, suggestion, rationale="", model="kimi-k2.6"):
self.suggestion, self.rationale, self.model, self.inputs = suggestion, rationale, model, []
def __call__(self, step, key, input, mode):
self.inputs.append((step, input))
return None if mode == "off" else SimpleNamespace(status="done", message="", output={
"suggestion": self.suggestion, "rationale": self.rationale, "confidence": 80, "model": self.model})
def rc_sale(ref, bp, tax_point=TODAY_):
must(api("POST", "/documents/compute", SYSTEM, {"lines": [line(1, "1000.00", category="construction_service")], "tax_point": tax_point,
"bp_id": bp, "document_ref": ref}))
return api("POST", f"/documents/{ref}/freeze", SYSTEM, {"contents": FULL})
def test_tax_13_hmrcs_name_and_consultation_number_are_kept_and_the_ai_flags_a_mismatch_without_changing_the_treatment(hmrc_world, monkeypatch):
vat_world()
ai = FakeAI({"name_mismatch": True}, "HMRC says Brick Wall Contractors Ltd; the partner is Acme Builders Ltd")
monkeypatch.setattr(ai_steps, "propose", ai)
out = must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
assert out["check_result"] == "valid" and out["check_name"] == "Brick Wall Contractors Ltd" and out["consultation_number"] == "CN-0001"
assert out["checked_on"] == "2026-11-20" and "differs" not in (out["check_flag"] or "") and "Acme" in out["check_flag"]
assert out["treatment_code"] == "uk_cis_contractor" # nothing changed
step, sent = ai.inputs[0]
assert step == "check_vat_number" and sent["context"]["hmrc_name"] == "Brick Wall Contractors Ltd"
assert api("POST", "/parties/7312/check-vat-number", BILLER, {})[0] == 403 # a machine permission
def test_tax_13_the_ai_comparison_is_off_when_the_setting_says_so(hmrc_world, monkeypatch):
vat_world()
ai = FakeAI({"name_mismatch": True}, "x")
monkeypatch.setattr(ai_steps, "propose", ai)
must(api("POST", "/settings/ai.check_vat_number.mode", ADMIN, {"value": "off"}))
out = must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
assert out["check_result"] == "valid" and out["ai"] is None and ai.inputs == [] and out["check_flag"] is None
def test_tax_13_an_invalid_number_blocks_a_reverse_charge_invoice_until_corrected(hmrc_world):
vat_world()
customer(7313, "uk_cis_contractor", "GB999999999")
assert must(api("POST", "/parties/7313/check-vat-number", SYSTEM, {}))["check_result"] == "invalid"
status, body = rc_sale("INV-RC", 7313)
assert status == 409 and "vat_number_checked" in {g["gate"] for g in gates_of(body)}
assert rc_sale("INV-OK", 7312)[0] == 200 # never checked: not blocked here
must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
assert rc_sale("INV-OK2", 7312)[0] == 200 # a current valid check
def test_tax_13_a_stale_check_is_rerun_before_the_next_reverse_charge_document_issues(hmrc_world):
vat_world()
must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
with on_day("2027-03-01T09:00:00+00:00"): # 101 days later, more than 90
status, body = rc_sale("INV-OLD", 7312, "2027-03-01")
assert status == 409 and "last checked 101 days ago" in [g for g in gates_of(body) if g["gate"] == "vat_number_checked"][0]["reason"]
must(api("POST", "/settings/vat_number_check_days", ADMIN, {"value": 200, "valid_from": "2027-03-02"}))
must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
assert rc_sale("INV-NEW", 7312, "2027-03-01")[0] == 200
def test_tax_13_a_number_hmrc_could_not_check_is_confirmed_by_a_person_with_a_reason(hmrc_world):
vat_world()
customer(7314, "overseas_business", "DE811234567")
out = must(api("POST", "/parties/7314/check-vat-number", SYSTEM, {}))
assert out["check_result"] == "could_not_check" # HMRC's lookup checks UK numbers only
assert api("POST", "/parties/7314/confirm-vat-number", BILLER, {"reason": "seen the certificate"})[0] == 403
assert api("POST", "/parties/7314/confirm-vat-number", ADMIN, {"reason": " "})[0] == 422
done = must(api("POST", "/parties/7314/confirm-vat-number", ADMIN, {"reason": "VIES shows the number valid on 2026-11-20"}))
assert done["confirmed_by"] == "tax_admin@shop.example" and "VIES" in done["confirm_reason"]
assert api("POST", "/parties/7312/confirm-vat-number", ADMIN, {"reason": "x"})[0] == 409 # nothing was checked on 7312
hmrc_world[0].vat_down = True
must(api("POST", "/parties/7312/check-vat-number", SYSTEM, {}))
assert must(api("GET", "/parties/7312", SYSTEM))["current"]["check_result"] == "could_not_check" # HMRC unavailable
# ═══ the AI review of a draft return ═════════════════════════════════════════════════════════
FINDING = {"findings": [{"evidence": "box 4 up 2.6x on the last four quarters; three bills from one supplier make 70%", "box": 4}]}
def test_the_ai_review_lists_findings_for_the_tax_admin_and_changes_nothing(hmrc_world, monkeypatch):
filing_ready()
frozen_sale("INV-1", [line(1, "1000.00")])
before = must(prepare())
ai = FakeAI(FINDING)
monkeypatch.setattr(ai_steps, "propose", ai)
out = must(api("POST", "/vat-returns/26AD/ai-review", SYSTEM, {}))
assert len(out["findings"]) == 1 and out["findings"][0]["decision"] == "open" and out["findings"][0]["model"] == "kimi-k2.6"
assert "box 4 up 2.6x" in out["findings"][0]["evidence"]
assert must(api("GET", "/vat-returns/period/26AD", ADMIN))["boxes"] == before["boxes"] # nothing changes
sent = json.dumps(ai.inputs[0][1]) # figures only: no names, no document ids
assert "Acme" not in sent and "INV-1" not in sent and '"box1"' in sent
assert must(api("POST", "/vat-returns/26AD/ai-review", SYSTEM, {}))["ai"]["kind"] == "replay"
decided = must(api("POST", "/vat-returns/26AD/findings/1/decide", ADMIN, {"decision": "dismiss", "note": "expected: a new supplier"}))
assert decided["decision"] == "dismiss" and decided["decided_by"] == "tax_admin@shop.example"
assert must(api("GET", "/vat-returns/26AD/findings", ADMIN))["findings"][0]["decision"] == "dismiss"
def test_locking_never_waits_on_a_finding_and_the_review_can_be_switched_off(hmrc_world, monkeypatch):
from conftest import AFTER_26AD
filing_ready()
must(prepare())
monkeypatch.setattr(ai_steps, "propose", FakeAI(FINDING))
must(api("POST", "/vat-returns/26AD/ai-review", SYSTEM, {}))
with on_day(AFTER_26AD):
assert api("POST", "/vat-returns/26AD/lock", ADMIN, {})[0] == 200 # the open finding does not stop the lock
must(api("POST", "/settings/ai.review_vat_return.mode", ADMIN, {"value": "off"}))
must(api("POST", "/vat-returns/26AD/reopen", ADMIN, {"reason": "check"}))
status, body = api("POST", "/vat-returns/26AD/ai-review", SYSTEM, {})
assert status == 409 and "switched off" in body["detail"]
def test_with_no_provider_the_review_returns_no_findings_and_a_person_reviews(hmrc_world):
filing_ready()
must(prepare())
out = must(api("POST", "/vat-returns/26AD/ai-review", SYSTEM, {}))
assert out["findings"] == [] and out["ai"]["kind"] in ("manual", "off")
assert api("POST", "/vat-returns/26AD/ai-review", ADMIN, {})[0] == 403 # the system's step
# ═══ settings and permissions ════════════════════════════════════════════════════════════════
def test_the_remaining_v4_settings_exist_with_their_defaults_and_validate():
got = {s["setting_key"]: s for s in must(api("GET", "/settings", ADMIN))["settings"]}
assert got["vat_number_check_days"]["value"] == 90 and got["cis_contractor_utr"]["value"] == ""
assert got["ai.review_vat_return.mode"]["value"] == "suggest" and got["ai.check_vat_number.mode"]["value"] == "suggest"
assert api("POST", "/settings/ai.review_vat_return.mode", ADMIN, {"value": "auto"})[0] == 422
assert must(api("POST", "/settings/vat_number_check_days", ADMIN, {"value": 30}))
def test_who_may_do_what_in_slice_b(hmrc_world):
cis_world()
assert api("POST", "/cis/subcontractors", BILLER, {"bp_id": 1, "name": "x", "utr": "1234567890", "ni_number": "AB"})[0] == 403
assert api("POST", "/cis/deductions", ADMIN, {"payment_ref": "p", "bp_id": 8801, "paid_on": "2026-11-10", "bill_ids": [], "gross_ex_vat": "1.00",
"materials": "0.00"})[0] == 403 # the deduction is a machine call
assert api("POST", "/trade-declarations/prepare", BILLER, {"kind": "ec_sales_list", "period": "2027-11"})[0] == 403
assert api("POST", "/partial-exemption/years", SYSTEM, {"year_end": "2027-03-31", "provisional_pct": "71"})[0] == 403
assert api("POST", "/capital-goods", BILLER, {"item_ref": "x", "kind": "computer", "bought_on": "2026-01-01", "net_value": "60000.00",
"input_vat": "1.00", "first_pct": "80"})[0] == 403
assert api("POST", "/cis/returns/2026-12-05/submit", ADMIN, {"employment_status_declared": True, "verification_declared": True})[0] == 403 # the owner's
assert api("POST", "/pva-statements", BILLER, {"month": "2026-11", "lines": []})[0] == 403