dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_invariants.py
"""The recurring bug classes, each tested by trying to break it: one claim key per money operation; side effects only after the
local commit, from pending rows; state machines never go backwards; the business date in business_timezone; money bounded (422,
never a 500); results derived from their own rows."""
import pytest
from conftest import (
ACCOUNTANT,
BOSS,
CASHIER,
CASHIER2,
LEAD,
NOBODY,
OPS,
OWNER,
SYSTEM,
api,
as_role,
freeze,
make_shop,
must,
open_till,
pay,
person,
refused,
ring,
sell,
)
from sqlalchemy import select
from brick_pos.application import use_cases
from brick_pos.domain.outbox import OutboxRow
from brick_pos.infrastructure.tables import pos_claim
def outbox(status=None):
return must(api("GET", "/outbox", LEAD, {"status": status} if status else None))["rows"]
def bill(session_id):
return ring(session_id, [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)]) # 10.00
# ── one claim key per money operation ───────────────────────────────────────────────────────────────────
def test_a_payment_takes_the_claim_pay_receipt_pay_no(engine, nb):
make_shop()
s = open_till()
r = bill(s["session_id"])
r = pay(r, "cash", "4.00")
pay(r, "cash", "6.00")
with engine.begin() as conn:
keys = sorted(row[0] for row in conn.execute(select(pos_claim.c.claim_key)))
assert keys == [f"pay:{r['receipt_id']}:1", f"pay:{r['receipt_id']}:2"]
def test_a_second_person_taking_the_same_payment_line_is_refused_not_double_counted(engine, nb):
from brick_pos.infrastructure.tables import ClaimRepository
make_shop()
s = open_till()
r = bill(s["session_id"])
with engine.begin() as conn: # a racer that committed claim 1 first
ClaimRepository(conn).claim(f"pay:{r['receipt_id']}:1", "pay", s["session_id"], "racer", __import__("datetime").datetime(2026, 11, 14))
body = refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "cash", "amount": "10.00", "version": r["version"]}),
409, "claimed a moment ago", gate="claim")
assert body["fix"] == "reload and retry"
after = must(api("GET", f"/receipts/{r['receipt_id']}", CASHIER))
assert after["payments"] == [] and after["paid"] == "0.00" # the refused attempt left nothing behind
# and two people holding the same version: one lands, the other is told it changed
one = pay(bill(s["session_id"]), "cash", "3.00")
stale = {"method": "cash", "amount": "3.00", "version": one["version"] - 1}
refused(api("POST", f"/receipts/{one['receipt_id']}/payments", CASHIER2, stale), 409, "changed since you loaded it")
# ── side effects only after the local commit, from pending rows ─────────────────────────────────────────
def test_stock_is_sent_from_a_pending_row_after_the_receipt_commits(nb):
make_shop()
s = open_till()
done = sell(s["session_id"], [("LATTE-REG", 2), ("CROISSANT", 1)])
[row] = outbox()
assert (row["target"], row["ref"], row["status"], row["attempts"]) == ("inventory.goods_issue", done["receipt_id"], "sent", 1)
[issue] = nb.issue_calls
assert issue["source_kind"] == "pos_sale" and issue["source_id"] == done["receipt_id"] and issue["location_id"] == "LON-SOHO/BAR"
assert issue["issued_on"] == "2026-11-14" and issue["company_code"] == "UK01"
assert [(ln["move_ref"], ln["product_id"], ln["quantity"], ln["unit_code"]) for ln in issue["lines"]] == [
(f"pos:{done['receipt_id']}:1:1", "LATTE-REG", "2", "C62"), (f"pos:{done['receipt_id']}:2:1", "CROISSANT", "1", "C62")]
def test_a_crash_between_the_commit_and_the_dispatch_sends_on_the_next_run(monkeypatch, nb):
monkeypatch.setattr(use_cases, "after_response", lambda *a, **k: None) # the process dies after the commit
make_shop()
s = open_till()
done = sell(s["session_id"], [("LATTE-REG", 1)])
assert done["status"] == "finalised" and nb.issue_calls == []
assert [(r["status"], r["attempts"]) for r in outbox()] == [("pending", 0)]
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 1, "pending": 0, "failed": 0}
assert list(nb.issues) == [done["receipt_id"]]
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0} # nothing twice
assert len(nb.issue_calls) == 1
def test_a_sale_is_never_refused_because_inventory_is_down_and_the_row_waits(nb):
nb.inventory_status = 500
make_shop()
s = open_till()
done = sell(s["session_id"], [("LATTE-REG", 1)])
assert done["status"] == "finalised" # the customer has their receipt
[row] = outbox()
assert (row["status"], row["attempts"]) == ("pending", 1) and "inventory answered 500" in row["last_error"]
nb.inventory_status = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] == 1
assert [(r["status"], r["attempts"], r["last_error"]) for r in outbox()] == [("sent", 2, "")]
assert list(nb.issues) == [done["receipt_id"]]
def test_a_request_inventory_refuses_for_good_is_failed_and_listed_not_retried(nb):
nb.inventory_status = 422
make_shop()
s = open_till()
sell(s["session_id"], [("LATTE-REG", 1)])
[row] = outbox("failed")
assert row["attempts"] == 1 and "inventory answered 422" in row["last_error"]
calls = len(nb.issue_calls)
nb.inventory_status = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0}
assert len(nb.issue_calls) == calls and outbox("pending") == [] # a failed row never goes back to pending by itself
def test_a_receipt_that_does_not_finalise_sends_nothing(nb):
make_shop()
s = open_till()
r = bill(s["session_id"])
refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}), 409, gate="paid_in_full")
assert outbox() == [] and nb.issue_calls == []
def test_finalising_twice_is_one_sale_and_one_stock_movement(nb):
make_shop()
s = open_till()
done = sell(s["session_id"], [("LATTE-REG", 1)])
again = must(api("POST", f"/receipts/{done['receipt_id']}/finalise", CASHIER, {}))
assert again["status"] == "finalised" and again["version"] == done["version"]
assert len(outbox()) == 1 and len(nb.issue_calls) == 1
# ── state machines never go backwards ───────────────────────────────────────────────────────────────────
def test_a_finalised_receipt_is_never_edited_or_reopened(nb):
make_shop()
s = open_till()
done = sell(s["session_id"], [("LATTE-REG", 1)])
refused(api("POST", f"/receipts/{done['receipt_id']}/payments", CASHIER, {"method": "cash", "amount": "1.00"}), 409,
"a finalised receipt is never changed", gate="receipt_open")
after = must(api("GET", f"/receipts/{done['receipt_id']}", CASHIER))
assert after["status"] == "finalised" and after["payments"] == done["payments"] and after["gross"] == done["gross"]
def test_a_closed_session_is_not_counted_again_and_a_sent_row_is_not_pending_again(nb):
make_shop()
s = open_till()
sid = s["session_id"]
assert must(api("POST", f"/sessions/{sid}/count", CASHIER, {"counted_cash": "150.00"}))["status"] == "closed"
refused(api("POST", f"/sessions/{sid}/count", CASHIER, {"counted_cash": "151.00"}), 409, "never goes back to open", gate="session_open")
row = OutboxRow.pending("inventory.goods_issue", "R", {})
row.sent(__import__("datetime").datetime(2026, 11, 14))
with pytest.raises(Exception, match="never goes back to pending"):
row.tried("late failure", permanent=False)
assert row.status == "sent"
# ── the business date, in business_timezone, never the UTC date ─────────────────────────────────────────
@pytest.mark.parametrize("instant,expected", [
("2026-11-15T00:30:00+00:00", "2026-11-14"), # 00:30 GMT on the 15th: before the 04:00 cutoff, still the 14th
("2026-11-15T04:00:00+00:00", "2026-11-15"), # the cutoff itself belongs to the new day
("2026-07-15T03:30:00+00:00", "2026-07-15"), # 04:30 BST: the 15th, though the UTC clock is before the cutoff
("2026-07-15T02:30:00+00:00", "2026-07-14"), # 03:30 BST: the 14th
("2026-03-29T00:30:00+00:00", "2026-03-28"), # the night the clocks go forward
])
def test_the_business_date_is_counted_in_business_timezone_with_the_cutoff(nb, instant, expected):
make_shop()
freeze(instant)
assert open_till()["business_date"] == expected
def test_a_dated_change_of_zone_takes_over_by_the_business_date_not_the_utc_date(nb):
make_shop()
must(api("POST", "/controls/business_timezone", OWNER, {"value": "Europe/Vienna", "valid_from": "2026-11-15"}))
freeze("2026-11-14T23:30:00+00:00") # London 23:30 on the 14th; Vienna would already say the 15th
assert open_till()["business_date"] == "2026-11-14"
freeze("2026-11-15T10:00:00+00:00") # London's date has reached the 15th: Vienna is in force
assert open_till(register="R2", who=CASHIER2)["business_date"] == "2026-11-15"
refused(api("POST", "/controls/business_timezone", OWNER, {"value": "Mars/Olympus", "valid_from": "2026-12-01"}), 422, "not an IANA time zone")
# ── money is bounded: 422 {gate: bounds}, never a 500 ───────────────────────────────────────────────────
BAD = ["-1.00", "1e20", "10000000000000000.00", "99999999999999999999999", "10.001", "abc", "", "NaN", "Infinity", 1.5, 1e300]
def bounds(result):
status, body = result
assert status == 422 and body.get("gate") == "bounds", (status, body)
@pytest.mark.parametrize("value", BAD)
def test_every_amount_the_books_cannot_hold_is_422_bounds(nb, value):
make_shop()
sid = open_till(register="R2", who=CASHIER2)["session_id"]
bounds(api("POST", "/sessions", CASHIER, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": value}))
bounds(api("POST", f"/sessions/{sid}/count", CASHIER2, {"counted_cash": value}))
bounds(api("POST", f"/sessions/{sid}/count", CASHIER2, {"counted_cash": "150.00", "float_left": value}))
bounds(api("POST", f"/sessions/{sid}/cash-moves", CASHIER2, {"kind": "safe_drop", "amount": value}))
r = bill(sid)
bounds(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER2, {"method": "cash", "amount": value}))
bounds(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER2, {"method": "cash", "amount": "10.00", "tip_amount": value}))
if value != "10.001": # three decimals are a fine quantity (10.001 kg)
bounds(api("POST", "/receipts", CASHIER2, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": value}]}))
bounds(api("POST", "/controls/difference_threshold", OWNER, {"value": value}))
@pytest.mark.parametrize("value", ["0", "0.00", "-0.01"])
def test_a_payment_a_move_and_a_quantity_are_more_than_zero(nb, value):
make_shop()
sid = open_till()["session_id"]
r = bill(sid)
bounds(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "cash", "amount": value}))
bounds(api("POST", f"/sessions/{sid}/cash-moves", CASHIER, {"kind": "safe_drop", "amount": value}))
bounds(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": value}]}))
def test_a_float_of_zero_and_a_tip_of_zero_are_fine_and_a_boolean_is_not_an_amount(nb):
make_shop()
assert open_till(opening_float="0.00")["opening_float"] == "0.00"
sid = "LON-SOHO-R1-0001"
r = bill(sid)
assert pay(r, "cash", "10.00", tip="0")["tips"] == "0.00"
bounds(api("POST", f"/sessions/{sid}/cash-moves", CASHIER, {"kind": "safe_drop", "amount": True}))
def test_a_yen_amount_has_no_decimals_and_a_two_place_currency_is_not_given_three(nb):
nb.price_list("LIST-JP", "JPY", [("LATTE-REG", "C62", "1", "480")])
nb.rate("JP", "hot_drink", "S10", "10.00")
make_shop("TOKYO", registers=("R1",), price_list="LIST-JP", country="JP")
s = open_till("TOKYO", "R1", "10000")
assert s["currency"] == "JPY"
bounds(api("POST", f"/sessions/{s['session_id']}/cash-moves", CASHIER, {"kind": "safe_drop", "amount": "100.50"}))
must(api("POST", f"/sessions/{s['session_id']}/cash-moves", CASHIER, {"kind": "safe_drop", "amount": "100"}))
r = ring(s["session_id"], [("LATTE-REG", 1)])
assert (r["gross"], r["vat"]) == ("480.00", "44.00") # 480 x 10/110 = 43.64, to the yen: 44
# ── results derive from their own rows ──────────────────────────────────────────────────────────────────
def test_the_expected_cash_is_this_sessions_own_rows_whatever_else_is_selling(nb):
make_shop()
make_shop("LON-KX", registers=("R1",))
a = open_till(register="R1")
sell(a["session_id"], [("LATTE-REG", 10)]) # 36.00 cash
b = open_till(register="R2", who=CASHIER2)
for _ in range(4):
sell(b["session_id"], [("CROISSANT", 3)], who=CASHIER2) # noise in the next drawer
other = open_till("LON-KX", "R1")
sell(other["session_id"], [("FLAT-WHITE", 7)]) # noise in another shop
freeze("2026-11-20T10:00:00+00:00") # and another business date
counted = must(api("POST", f"/sessions/{a['session_id']}/count", CASHIER, {"counted_cash": "186.00"}))
assert counted["expected_cash"] == "186.00" and counted["status"] == "closed"
assert counted["summary"]["receipts"] == 1 and counted["summary"]["gross"] == "36.00"
# ── who may, and which rows ─────────────────────────────────────────────────────────────────────────────
def test_a_cashier_sees_only_their_shop_and_a_row_of_another_shop_is_a_404(nb):
nb.price_list("LIST-UK-KX", "GBP", [("LATTE-REG", "C62", "1", "3.60")])
kx = as_role("cashier", "kx@shop.example")
make_shop()
make_shop("LON-KX", registers=("R1",), staff=(kx,), price_list="LIST-UK-KX")
mine, theirs = open_till(), open_till("LON-KX", "R1", who=kx)
r = bill(mine["session_id"])
assert must(api("GET", "/sessions", CASHIER))["sessions"] == [mine["session_id"]]
assert must(api("GET", "/shops", CASHIER))["shops"] == ["LON-SOHO"]
for call in (api("GET", f"/sessions/{theirs['session_id']}", CASHIER),
api("POST", f"/sessions/{theirs['session_id']}/count", CASHIER, {"counted_cash": "150.00"}),
api("POST", f"/sessions/{theirs['session_id']}/cash-moves", CASHIER, {"kind": "safe_drop", "amount": "1.00"}),
api("POST", "/receipts", CASHIER, {"session_id": theirs["session_id"], "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}),
api("GET", "/shops/LON-KX", CASHIER), api("POST", "/sessions", CASHIER, {"shop_id": "LON-KX", "register_id": "R1", "opening_float": "1.00"})):
assert call[0] == 404, call
assert api("GET", f"/receipts/{r['receipt_id']}", kx)[0] == 404 # and the other way round
assert must(api("GET", "/sessions", kx))["sessions"] == [theirs["session_id"]]
unassigned = person("new@shop.example", ["pos:shop:read", "pos:session:open"])
assert must(api("GET", "/sessions", unassigned))["sessions"] == []
assert api("GET", f"/sessions/{mine['session_id']}", unassigned)[0] == 404
assert sorted(must(api("GET", "/sessions", ACCOUNTANT))["sessions"]) == sorted([mine["session_id"], theirs["session_id"]]) # sees every shop
def test_nobody_can_do_what_their_role_does_not_name(nb):
make_shop()
s = open_till()
sid = s["session_id"]
sell(sid, [("LATTE-REG", 10)])
must(api("POST", f"/sessions/{sid}/count", CASHIER, {"counted_cash": "100.00"}))
refused(api("POST", f"/sessions/{sid}/approve-difference", CASHIER, {"reason": "mine"}), 403, "pos:session:approve_difference")
refused(api("POST", f"/sessions/{sid}/approve-difference", ACCOUNTANT, {"reason": "mine"}), 403, "pos:session:approve_difference")
for who in (ACCOUNTANT, OWNER):
refused(api("POST", "/receipts", who, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}), 403, "pos:receipt:sell")
refused(api("POST", "/sessions", who, {"shop_id": "LON-SOHO", "register_id": "R2", "opening_float": "1.00"}), 403, "pos:session:open")
refused(api("POST", "/settings/day_cutoff_time", ACCOUNTANT, {"value": "05:00"}), 403, "pos:settings:write")
refused(api("POST", "/shops", CASHIER, {"shop_id": "X"}), 403, "pos:shop:write")
assert api("GET", "/shops", NOBODY)[0] == 403
assert api("GET", "/shops", {})[0] == 401
@pytest.mark.parametrize("who", [CASHIER, LEAD, OPS, ACCOUNTANT, OWNER, BOSS, person("root@shop.example", ["*:*:*"])])
def test_a_machine_permission_is_refused_to_every_person_role_and_to_a_wildcard(nb, who):
refused(api("POST", "/outbox/dispatch", who, {}), 403, "pos:outbox:dispatch")
def test_the_system_client_alone_dispatches(nb):
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0}