dodil/erp-brick-tax
DODIL ERP brick: tax
tests/test_schemes.py
"""v4 slice B, the schemes: cash accounting (TAX-36), the flat rate (TAX-37), partial exemption (TAX-48), capital goods (TAX-49), and
the DataK3-safe claim of a return's entries."""
from decimal import Decimal
from conftest import (
ADMIN,
BILLER,
FULL,
SYSTEM,
api,
filing_ready,
frozen_purchase,
frozen_sale,
gates_of,
hmrc_ob,
line,
must,
on_day,
prepare,
sync,
)
D = Decimal
LATE = "2027-04-10T09:00:00+00:00" # after 27AA ended, before 27AB ends
AFTER_27AB = "2027-07-10T09:00:00+00:00"
def periods_2027(h, extra=()):
h.obs += [hmrc_ob("27AA", "2027-01-01", "2027-03-31", "2027-05-07"), hmrc_ob("27AB", "2027-04-01", "2027-06-30", "2027-08-07"),
hmrc_ob("27AC", "2027-07-01", "2027-09-30", "2027-11-07"), *extra]
def scheme(body):
return api("POST", "/registration/change-scheme", ADMIN, body)
def ready_2027(hmrc_world, extra=()):
filing_ready()
periods_2027(hmrc_world[0], extra)
sync(to="2029-12-31")
def box(out, n):
return D(out["boxes"][str(n)])
def pay(p, payment_id, invoice, amount, received, seq=1):
p.cash.append({"payment_id": payment_id, "invoice_id": invoice, "seq": seq, "amount": amount, "currency": "GBP", "received_on": received,
"allocated_on": received, "source_kind": "receipt"})
# ── TAX-36 cash accounting ─────────────────────────────────────────────────────────────────────
def cash_world(hmrc_world):
ready_2027(hmrc_world)
must(scheme({"scheme": "cash_accounting", "effective_from": "2027-01-01", "expected_turnover": "900000.00"}))
def test_tax_36_the_vat_follows_the_money_the_receipt_date_is_the_sales_tax_point(hmrc_world, ar_and_payments, purchasing_and_products):
_, pays = ar_and_payments
pur, _ = purchasing_and_products
cash_world(hmrc_world)
frozen_sale("INV-2027-0012", [line(1, "1000.00")], tax_point="2027-01-15") # VAT 200.00
frozen_purchase("PB-1", [line(1, "250.00")], tax_point="2027-03-10") # input VAT 50.00
pay(pays, "PAY-1", "INV-2027-0012", "1200.00", "2027-04-03")
pur.payments.append({"payment_id": "SP-1", "run_id": "R1", "bp_id": 8800, "bank_account_id": "BA1", "amount": "300.00", "currency": "GBP",
"reference": "PB-1", "status": "confirmed", "paid_on": "2027-03-20", "bills": [{"bill_id": "PB-1", "amount": "300.00"}]})
with on_day(LATE):
a = must(prepare("27AA"))
assert a["accounting_basis"] == "cash" and box(a, 1) == 0 and box(a, 4) == D("50.00") and box(a, 7) == 250
b = must(prepare("27AB"))
assert box(b, 1) == D("200.00") and box(b, 6) == 1000 and box(b, 4) == 0
def test_tax_36_a_part_payment_taxes_the_vat_fraction_and_the_periods_sum_to_the_penny(hmrc_world, ar_and_payments):
_, pays = ar_and_payments
cash_world(hmrc_world)
frozen_sale("INV-P", [line(1, "333.33")], tax_point="2027-01-15") # VAT 66.67, gross 400.00
pay(pays, "P1", "INV-P", "100.00", "2027-02-01")
pay(pays, "P2", "INV-P", "100.00", "2027-04-02")
pay(pays, "P3", "INV-P", "200.00", "2027-05-02")
with on_day(AFTER_27AB):
got = [box(must(prepare(k)), 1) for k in ("27AA", "27AB")]
assert got[0] == D("16.67") and got[1] == D("50.00") # 25% then 75% of 66.67
assert sum(got) == D("66.67") # exact: the sum IS the invoice's VAT
def test_tax_36_an_invoice_payable_over_six_months_ahead_stays_on_the_invoice_basis(hmrc_world):
cash_world(hmrc_world)
must(api("POST", "/documents/compute", SYSTEM, {"lines": [line(1, "1000.00")], "tax_point": "2027-02-01", "bp_id": 7311,
"document_ref": "INV-LONG", "due_date": "2027-12-01"}))
must(api("POST", "/documents/INV-LONG/freeze", SYSTEM, {"contents": FULL}))
with on_day(LATE):
assert box(must(prepare("27AA")), 1) == D("200.00") # unpaid, yet in its tax point's return
def test_tax_36_leaving_the_scheme_brings_every_outstanding_vat_into_the_leaving_period(hmrc_world):
cash_world(hmrc_world)
frozen_sale("INV-HUGE", [line(1, "1700000.00")], tax_point="2027-02-10") # turnover above 1,600,000
frozen_sale("INV-U", [line(1, "1000.00")], tax_point="2027-01-20")
with on_day(LATE):
out = must(prepare("27AA"))
assert box(out, 1) == D("340200.00") # 20% of both, still unpaid
assert {g["gate"]: g["met"] for g in out["scheme_gates"]}["cash_accounting_limit"] is False
assert out["cash_basis"]["leaving"] is True
def test_tax_36_the_basis_changes_only_from_the_start_of_a_vat_period_after_today(hmrc_world):
ready_2027(hmrc_world)
status, body = scheme({"scheme": "cash_accounting", "effective_from": "2027-01-15", "expected_turnover": "1000.00"})
assert status == 409 and gates_of(body)[0]["gate"] == "vat_period_start"
status, body = scheme({"scheme": "cash_accounting", "effective_from": "2026-10-01", "expected_turnover": "1000.00"})
assert status == 409 and gates_of(body)[0]["gate"] == "after_today"
status, body = scheme({"scheme": "cash_accounting", "effective_from": "2027-01-01", "expected_turnover": "1400000.00"})
assert status == 409 and gates_of(body)[0]["gate"] == "scheme_eligible" # above the joining limit
assert api("POST", "/registration/change-scheme", BILLER, {"scheme": "standard", "effective_from": "2027-01-01"})[0] == 403
# ── TAX-37 flat rate ───────────────────────────────────────────────────────────────────────────
def flat_world(hmrc_world):
ready_2027(hmrc_world)
must(scheme({"scheme": "flat_rate", "effective_from": "2027-01-01", "expected_turnover": "100000.00",
"flat_rate_sector": "computer and IT consultancy", "first_year_until": "2027-12-31"}))
def test_tax_37_the_limited_cost_trader_gate_comes_first_and_box_4_holds_only_capital_goods(hmrc_world):
flat_world(hmrc_world)
frozen_sale("INV-F", [line(1, "50000.00")], tax_point="2027-02-01") # 60,000.00 including VAT
frozen_purchase("PB-G2", [line(1, "250.00", supply="goods")], tax_point="2027-02-05")
frozen_purchase("PB-LAPTOP", [line(1, "2000.00", supply="goods")], tax_point="2027-02-09") # 2,400.00 incl. VAT: capital goods
with on_day(LATE):
out = must(prepare("27AA"))
g = {x["gate"]: x for x in out["scheme_gates"]}
assert g["limited_cost_trader"]["met"] is True and "1,200.00" in g["limited_cost_trader"]["reason"]
assert out["flat_rate"]["rate_pct"] == "15.5" or D(str(out["flat_rate"]["rate_pct"])) == D("15.5") # 16.5% less the first-year 1%
assert box(out, 1) == D("9300.00") and box(out, 6) == 60000 and box(out, 4) == D("400.00")
assert D(str(out["flat_rate"]["goods_cost"])) == D("300.00")
def test_tax_37_above_two_percent_goods_the_sector_rate_applies_less_the_first_year_discount(hmrc_world):
flat_world(hmrc_world)
frozen_sale("INV-F", [line(1, "50000.00")], tax_point="2027-02-01")
frozen_purchase("PB-GOODS", [line(1, "4000.00", supply="goods")], tax_point="2027-02-05") # 4,800.00 goods, not capital (a line of 2,000 or more is: 4,800 is capital)
frozen_purchase("PB-SMALL", [line(1, "1000.00", supply="goods"), line(2, "1000.00", supply="goods")], tax_point="2027-02-06")
with on_day(LATE):
out = must(prepare("27AA"))
assert {g["gate"]: g["met"] for g in out["scheme_gates"]}["limited_cost_trader"] is False
assert D(str(out["flat_rate"]["rate_pct"])) == D("13.5") and box(out, 1) == D("8100.00") # 14.5% less 1%
def test_tax_37_turnover_above_the_limit_answers_flat_rate_eligible_false_and_the_lock_waits_for_the_way_out(hmrc_world):
flat_world(hmrc_world)
frozen_sale("INV-BIG", [line(1, "200000.00")], tax_point="2027-02-01") # 240,000.00 incl. VAT, above 230,000
with on_day(LATE):
out = must(prepare("27AA"))
assert {g["gate"]: g["met"] for g in out["scheme_gates"]}["flat_rate_eligible"] is False
status, body = api("POST", "/vat-returns/27AA/lock", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "flat_rate_eligible"
must(scheme({"scheme": "standard", "effective_from": "2027-07-01"})) # recorded: leaving from the next period
assert api("POST", "/vat-returns/27AA/lock", ADMIN, {})[0] == 200
# ── TAX-48 partial exemption ───────────────────────────────────────────────────────────────────
def pe_world(hmrc_world, pct="71"):
ready_2027(hmrc_world)
must(api("POST", "/categories/pe_taxable", ADMIN, {"description": "for taxable supplies"}))
must(api("POST", "/categories/pe_exempt", ADMIN, {"description": "for exempt supplies"}))
must(api("POST", "/categories/pe_residual", ADMIN, {"description": "for both"}))
for code, attr in (("PT", "taxable"), ("PX", "exempt"), ("PR", "residual")):
must(api("POST", "/codes", ADMIN, {"tax_code": code, "treatment": "standard", "direction": "purchase", "rate_pct": "20",
"valid_from": "2011-01-04", "reporting_box": "4", "value_box": "7", "reclaimable_pct": "100",
"attribution": attr}))
must(api("POST", f"/codes/{code}/activate-code", ADMIN, {}))
must(api("POST", "/rules", ADMIN, {"category_code": f"pe_{attr}", "treatment_code": "uk_business", "tax_code": code, "direction": "purchase"}))
must(scheme({"scheme": "standard", "effective_from": "2027-01-01", "partial_exemption_year_end": "03-31"}))
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2027-03-31", "provisional_pct": pct}))
def pe_bills(taxable="30000.00", exempt="4500.00", residual="15000.00"):
frozen_purchase("PB-T", [line(1, taxable, category="pe_taxable")], tax_point="2027-01-20")
frozen_purchase("PB-X", [line(1, exempt, category="pe_exempt")], tax_point="2027-02-01")
frozen_purchase("PB-R", [line(1, residual, category="pe_residual")], tax_point="2027-02-15")
def test_tax_48_each_return_reclaims_taxable_plus_the_residual_percentage_and_exempt_in_full_while_de_minimis_holds(hmrc_world):
pe_world(hmrc_world)
pe_bills() # input VAT 6,000.00 / 900.00 / 3,000.00
with on_day(LATE):
out = must(prepare("27AA"))
pe = out["partial_exemption"]
assert D(str(pe["taxable_and_residual_reclaimed"])) == D("8130.00") # 6,000.00 + 71% of 3,000.00
assert D(str(pe["exempt_reclaimed"])) == D("900.00") and pe["de_minimis_holds"] is True
assert box(out, 4) == D("9030.00")
assert {g["gate"]: g["met"] for g in out["scheme_gates"]}["de_minimis_1"] is True
def test_tax_48_when_de_minimis_fails_the_exempt_input_vat_is_not_reclaimed(hmrc_world):
pe_world(hmrc_world)
pe_bills(exempt="40000.00") # exempt input VAT 8,000.00 in a quarter
with on_day(LATE):
out = must(prepare("27AA"))
assert out["partial_exemption"]["de_minimis_holds"] is False
assert D(str(out["partial_exemption"]["exempt_reclaimed"])) == 0 and box(out, 4) == D("8130.00")
def test_tax_48_a_year_with_no_percentage_refuses_the_return_with_its_gate(hmrc_world):
pe_world(hmrc_world)
pe_bills()
with on_day("2027-07-10T09:00:00+00:00"):
status, body = prepare("27AC") # the year ending 2028-03-31 has no percentage
assert status == 409 and gates_of(body)[0]["gate"] == "pe_percentage"
def test_tax_48_the_annual_adjustment_recomputes_the_percentage_and_lands_on_the_first_return_after_the_year_end(hmrc_world):
pe_world(hmrc_world)
pe_bills()
frozen_sale("INV-T", [line(1, "70000.00")], tax_point="2027-02-10") # taxable supplies
frozen_sale("INV-E", [line(1, "30000.00", category="exempt")], tax_point="2027-02-11") # exempt supplies
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2028-03-31", "provisional_pct": "70"}))
with on_day("2027-03-20T09:00:00+00:00"):
status, body = api("POST", "/partial-exemption/years/2027-03-31/adjust", ADMIN, {})
assert status == 409 and gates_of(body)[0]["gate"] == "year_ended"
with on_day(LATE):
must(prepare("27AA"))
y = must(api("POST", "/partial-exemption/years/2027-03-31/adjust", ADMIN, {}))
assert D(str(y["final_pct"])) == 70 and D(str(y["adjustment_amount"])) == D("-30.00") # 3,000.00 x (70% - 71%)
again = must(api("POST", "/partial-exemption/years/2027-03-31/adjust", ADMIN, {}))
assert again["replay"] is True and D(str(again["adjustment_amount"])) == D("-30.00")
assert D(str(must(prepare("27AA"))["boxes"]["4"])) == D("9030.00") # not on a return of the year itself
out = must(prepare("27AB"))
assert box(out, 4) == D("-30.00") # the first return after the year end
assert must(api("GET", "/partial-exemption/years/2027-03-31", ADMIN))["adjusted_in_period_key"] == "27AB"
def test_tax_48_the_percentage_rounds_up_to_a_whole_percent_while_residual_input_is_small(hmrc_world):
pe_world(hmrc_world)
with on_day(LATE):
y = must(api("POST", "/partial-exemption/years/2027-03-31/adjust", ADMIN, {"taxable_supplies": "7041.00", "exempt_supplies": "2959.00"}))
assert D(str(y["final_pct"])) == 71 # 70.41% rounded UP
# ── TAX-49 capital goods ───────────────────────────────────────────────────────────────────────
BUILDING = {"item_ref": "BLD-1", "kind": "land_building", "bought_on": "2026-06-01", "net_value": "500000.00", "input_vat": "100000.00",
"first_pct": "80"}
def test_tax_49_an_item_is_recorded_once_and_year_two_adjusts_one_interval_onto_the_first_return_after_the_year_end(hmrc_world):
pe_world(hmrc_world)
first = must(api("POST", "/capital-goods", ADMIN, BUILDING))
assert first["intervals"] == 10 and first["replay"] is False
assert must(api("POST", "/capital-goods", ADMIN, BUILDING))["replay"] is True
status, body = api("POST", "/capital-goods", ADMIN, {**BUILDING, "item_ref": "SHED", "net_value": "100000.00"})
assert status == 409 and gates_of(body)[0]["gate"] == "capital_item_threshold"
assert api("POST", "/capital-goods", ADMIN, {**BUILDING, "item_ref": "SH2", "kind": "computer", "net_value": "40000.00"})[0] == 409
assert must(api("POST", "/capital-goods", ADMIN, {**BUILDING, "item_ref": "PC-1", "kind": "computer", "net_value": "60000.00",
"input_vat": "12000.00"}))["intervals"] == 5
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2028-03-31", "provisional_pct": "80"}))
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2029-03-31", "provisional_pct": "70"}))
with on_day(LATE): # year one is the interval the item was bought in: no adjustment
year_one = must(api("POST", "/partial-exemption/years/2027-03-31/adjust", ADMIN, {"taxable_supplies": "60", "exempt_supplies": "40"}))
assert year_one["capital_goods"] == []
with on_day("2028-04-10T09:00:00+00:00"):
one = must(api("POST", "/partial-exemption/years/2028-03-31/adjust", ADMIN, {"taxable_supplies": "70", "exempt_supplies": "30"}))
got = {c["item_ref"]: c for c in one["capital_goods"]}
assert D(got["BLD-1"]["amount"]) == D("-1000.00") and got["BLD-1"]["interval"] == 2 # (100,000.00 / 10) x (70% - 80%)
again = must(api("POST", "/partial-exemption/years/2028-03-31/adjust", ADMIN, {}))
assert again["replay"] is True and [c for c in again["capital_goods"] if c["item_ref"] == "BLD-1"][0]["amount"] == got["BLD-1"]["amount"]
item = must(api("GET", "/capital-goods/BLD-1", ADMIN))
assert len(item["adjustments"]) == 1 # each interval is adjusted once
sync(to="2029-12-31")
def test_tax_49_the_adjustment_is_an_entry_naming_the_item_and_interval(hmrc_world, engine):
from brick_tax.infrastructure.tables import Repository
pe_world(hmrc_world)
hmrc_world[0].obs.append(hmrc_ob("28AA", "2028-04-01", "2028-06-30", "2028-08-07"))
must(api("POST", "/capital-goods", ADMIN, BUILDING))
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2028-03-31", "provisional_pct": "80"}))
must(api("POST", "/partial-exemption/years", ADMIN, {"year_end": "2029-03-31", "provisional_pct": "70"}))
with on_day("2028-04-10T09:00:00+00:00"):
sync(to="2029-12-31")
must(api("POST", "/partial-exemption/years/2028-03-31/adjust", ADMIN, {"taxable_supplies": "70", "exempt_supplies": "30"}))
out = must(prepare("28AA"))
assert box(out, 4) == D("-1000.00") or box(out, 4) < 0
with engine.begin() as conn:
e = [r for r in Repository(conn).entries.where(period_key="28AA") if r["entry_kind"] == "capital_goods"][0]
assert e["document_ref"] == "BLD-1" and "interval 2" in e["detail"] and D(str(e["box4_amount"])) == D("-1000.00")
def test_tax_49_a_sale_inside_the_adjustment_period_answers_the_disposal_gate(hmrc_world):
pe_world(hmrc_world)
must(api("POST", "/capital-goods", ADMIN, BUILDING))
status, body = api("POST", "/capital-goods/BLD-1/dispose", ADMIN, {"disposed_on": "2028-09-01"})
assert status == 409 and gates_of(body)[0]["gate"] == "disposal_adjustment"
out = must(api("POST", "/capital-goods/BLD-1/dispose", ADMIN, {"disposed_on": "2028-09-01", "disposal_pct": "0"}))
adj = out["adjustments"][0]
assert adj["interval"] == "3D" and D(adj["amount"]) == D("-64000.00") # sold in interval 3: 8 intervals x 10,000.00 x (0% - 80%)
# ── the claim of a return's entries is safe on the DataK3 pg wire ───────────────────────────────
def test_a_claim_is_a_check_then_a_plain_insert_and_a_racing_one_surfaces_as_a_conflict(hmrc_world, engine):
import pytest
from erp_platform.repository import Conflict
from brick_tax.infrastructure.tables import Repository
flat = {"document_ref": "INV-1", "tax_code": "S", "entry_kind": "document", "vrn": "123456789", "period_key": "26AD", "direction": "sale",
"net_amount": D(1), "vat_amount": D(0), "box1_amount": D(0), "box4_amount": D(0), "box_value": "6", "source": "document"}
with engine.begin() as conn:
entries = Repository(conn).entries
assert entries.claim(flat) is True
with engine.begin() as conn:
entries = Repository(conn).entries
assert entries.claim({**flat, "period_key": "26AE"}) is False # held by another return: skipped, not swallowed
real_get = entries.get
entries.get = lambda **k: None # a racer that passed the check before the other committed
with pytest.raises(Conflict):
entries.claim({**flat, "period_key": "26AE"})
entries.get = real_get
def test_a_refresh_never_uses_a_savepoint_or_deletes_a_key_it_inserts_in_the_same_transaction(hmrc_world, engine):
from sqlalchemy import event
filing_ready()
frozen_sale("INV-R", [line(1, "1000.00")])
must(prepare("26AD"))
seen = []
event.listen(engine, "before_cursor_execute", lambda c, cur, stmt, p, ctx, many: seen.append(stmt))
must(api("POST", "/vat-returns/26AD/refresh", ADMIN, {}))
sql = " ".join(seen).upper()
assert "SAVEPOINT" not in sql and "DELETE FROM TAX_VAT_RETURN_ENTRY" not in sql # entries refreshed in place: one keyed upsert, not DELETE + INSERT
assert "INSERT INTO TAX_VAT_RETURN_ENTRY" in sql and "ON CONFLICT" in sql