dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_vouchers.py
"""Slice B, item 3: gift vouchers (POS-13) and the rest of POS-09 - a bill split across a voucher, cash and a card, a declined card, and a
terminal that times out - each written from its Given / When / Then through the doors, with the invariants that bite here: one claim per
movement of a voucher (voucher:<code>:<n>, never a pay: key), the terminal's idempotency key on the payment LINE, states that never go
back, 422 bounds, flat statements."""
from datetime import datetime
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
CASHIER,
CASHIER2,
OPS,
ROLES,
api,
as_role,
finalise,
freeze,
make_shop,
must,
open_till,
pay,
person,
refused,
ring,
)
from sqlalchemy import select
from test_statement_budget import counting
from brick_pos.infrastructure.tables import ClaimRepository, pos_claim
def sell_voucher(session_id, code, amount, method="card", who=None):
"""Sell a gift voucher on its own receipt and take payment for it."""
r = must(api("POST", "/receipts", who or CASHIER, {"session_id": session_id, "vouchers": [{"voucher_code": code, "amount": amount}]}))
more = {"terminal_txn_ref": f"T-{r['receipt_id']}-1", "terminal_status": "authorised"} if method == "card" else {}
return finalise(pay(r, method, **more, who=who), who)
def voucher(code, who=None):
return must(api("GET", f"/vouchers/{code}", who or CASHIER))
def redeem(receipt, code, amount, who=None):
return api("POST", f"/receipts/{receipt['receipt_id']}/payments", who or CASHIER,
{"method": "voucher", "voucher_code": code, "amount": amount, "version": receipt["version"]})
def with_prices(nb):
nb.price_lists["LIST-UK-SHOP"]["prices"] += [("COMBO", "C62", "1", "9.20"), ("BRUNCH", "C62", "1", "23.80")]
nb.product_row("COMBO", "hot_drink")
nb.product_row("BRUNCH", "hot_drink")
def a_till(nb):
with_prices(nb)
make_shop()
return open_till()["session_id"]
# ── POS-13: sell a gift voucher and redeem one ──────────────────────────────────────────────────────────
def test_pos_13_a_voucher_is_sold_outside_vat_and_is_live_when_the_receipt_is_final(nb):
sid = a_till(nb)
r = must(api("POST", "/receipts", CASHIER, {"session_id": sid, "vouchers": [{"voucher_code": "GV-UK-000813", "amount": "25.00"}]}))
[ln] = r["lines"]
assert (ln["kind"], ln["voucher_code"], ln["line_gross"], ln["vat"], ln["tax_code"], ln["tax_category"]) == \
("voucher_sale", "GV-UK-000813", "25.00", "0.00", "", "voucher")
assert (r["gross"], r["vat"], r["net"]) == ("25.00", "0.00", "25.00")
assert api("GET", "/vouchers/GV-UK-000813", CASHIER)[0] == 404 # not live while the receipt is open
sell_pay = finalise(pay(r, "card", terminal_txn_ref="T-1", terminal_status="authorised"))
v = voucher("GV-UK-000813")
assert (v["face_value"], v["balance"], v["state"], v["company_code"], v["currency"], v["profile"], v["issued_receipt_id"]) == \
("25.00", "25.00", "active", "UK01", "GBP", "multi_purpose", sell_pay["receipt_id"])
assert v["expires_on"] == "2028-11-14" # 24 months from the day it was sold
assert [(c["claim_no"], c["operation"], c["amount"], c["balance_after"]) for c in v["claims"]] == [(1, "issue", "25.00", "25.00")]
assert nb.issue_calls == [] # a voucher takes no stock
def test_pos_13_the_voucher_is_redeemed_for_part_of_a_bill_and_the_vat_is_on_the_goods(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-UK-000813", "25.00")
r = ring(sid, [("COMBO", 1)]) # 9.20
r = must(redeem(r, "GV-UK-000813", "9.20"))
[p] = r["payments"]
assert (p["method"], p["amount"], p["voucher_code"], p["terminal_status"], p["idempotency_key"]) == \
("voucher", "9.20", "GV-UK-000813", "", f"{r['receipt_id']}:1")
assert (r["paid"], r["remaining"]) == ("9.20", "0.00")
done = finalise(r)
assert (done["gross"], done["vat"], done["net"]) == ("9.20", "1.53", "7.67") # VAT at redemption, on the goods
v = voucher("GV-UK-000813")
assert (v["balance"], v["state"]) == ("15.80", "active")
assert [(c["claim_no"], c["operation"], c["receipt_id"], c["pay_no"], c["amount"], c["balance_after"]) for c in v["claims"]][1:] == \
[(2, "redeem", done["receipt_id"], 1, "-9.20", "15.80")]
def test_pos_13_a_redemption_above_the_balance_or_on_an_expired_voucher_is_refused(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-UK-000813", "10.00")
r = ring(sid, [("BRUNCH", 1)])
refused(redeem(r, "GV-UK-000813", "10.01"), 409, "holds 10.00, not 10.01", gate="voucher_balance")
refused(redeem(r, "GV-NOPE", "1.00"), 409, "no voucher GV-NOPE", gate="voucher_known")
assert voucher("GV-UK-000813")["balance"] == "10.00" and must(api("GET", f"/receipts/{r['receipt_id']}", CASHIER))["payments"] == []
freeze("2028-11-14T10:00:00+00:00")
last_day = open_till(register="R2", who=CASHIER2)
r2 = ring(last_day["session_id"], [("BRUNCH", 1)], who=CASHIER2)
assert must(redeem(r2, "GV-UK-000813", "5.00", who=CASHIER2))["paid"] == "5.00" # the last day it is valid
must(api("POST", f"/receipts/{r2['receipt_id']}/void", CASHIER2, {"reason": "x"}))
must(api("POST", f"/sessions/{last_day['session_id']}/count", CASHIER2, {"counted_cash": "150.00"}))
freeze("2028-11-15T10:00:00+00:00")
r3 = ring(open_till(register="R2", who=CASHIER2)["session_id"], [("BRUNCH", 1)], who=CASHIER2)
refused(redeem(r3, "GV-UK-000813", "1.00", who=CASHIER2), 409, "expired on 2028-11-14", gate="voucher_expired")
assert voucher("GV-UK-000813")["state"] == "expired"
def test_pos_13_a_voucher_is_a_debt_of_one_company_in_one_currency(nb):
from fakes import seed_de
sid = a_till(nb)
sell_voucher(sid, "GV-UK-000813", "25.00")
seed_de(nb)
make_shop("BER-UK", registers=("R1",), price_list="LIST-DE-SHOP", country="DE", company="UK01") # the same company, euros
eur = open_till("BER-UK", "R1")
r = ring(eur["session_id"], [("LATTE-REG", 1)])
refused(redeem(r, "GV-UK-000813", "1.00"), 409, "is in GBP, the receipt in EUR", gate="voucher_currency")
make_shop("DE-SHOP", registers=("R1",), company="DE01")
other = open_till("DE-SHOP", "R1")
r = ring(other["session_id"], [("LATTE-REG", 1)])
refused(redeem(r, "GV-UK-000813", "1.00"), 409, "sold by company UK01", gate="voucher_company")
assert voucher("GV-UK-000813")["balance"] == "25.00"
def test_pos_13_a_code_is_sold_once_and_a_voucher_pays_at_most_the_bill_with_no_tip(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-1", "50.00")
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "vouchers": [{"voucher_code": "GV-1", "amount": "5.00"}]}), 409,
"was issued already", gate="voucher_code")
twice = api("POST", "/receipts", CASHIER, {"session_id": sid, "vouchers": [{"voucher_code": "GV-2", "amount": "5.00"},
{"voucher_code": "GV-2", "amount": "5.00"}]})
assert twice[0] == 409 and twice[1]["gate"] == "voucher_code"
r = ring(sid, [("COMBO", 1)])
refused(redeem(r, "GV-1", "9.21"), 409, "pays at most the bill", gate="within_remainder")
refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "voucher", "voucher_code": "GV-1", "amount": "9.20",
"tip_amount": "1.00"}), 409, "not with a voucher", gate="tip_method")
assert voucher("GV-1")["balance"] == "50.00"
def test_pos_13_a_sold_voucher_is_not_refunded_and_a_voucher_and_a_product_share_a_receipt(nb):
sid = a_till(nb)
r = must(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "COMBO", "quantity": "1"}],
"vouchers": [{"voucher_code": "GV-3", "amount": "20.00"}]}))
assert [(ln["line_no"], ln["kind"]) for ln in r["lines"]] == [(1, "product"), (2, "voucher_sale")] and r["gross"] == "29.20"
done = finalise(pay(r, "cash", "29.20"))
assert [i["lines"] for i in nb.issue_calls] == [[{"move_ref": f"pos:{done['receipt_id']}:1:1", "product_id": "COMBO", "quantity": "1",
"unit_code": "C62"}]] # the voucher line took no stock
refused(api("POST", f"/receipts/{done['receipt_id']}/refund", CASHIER, {"session_id": sid, "lines": [{"line_no": 2, "quantity": "1"}],
"method": "cash", "reason": "x"}), 409, "money we owe", gate="voucher_not_refunded")
def test_pos_13_selling_a_voucher_needs_its_own_permission(nb):
sid = a_till(nb)
seller_only = person("till@shop.example", [p for p in ROLES["cashier"] if p != "pos:voucher:sell"])
must(api("POST", "/shops/LON-SOHO/staff", OPS, {"sub": "till@shop.example", "email": "till@shop.example"}))
refused(api("POST", "/receipts", seller_only, {"session_id": sid, "vouchers": [{"voucher_code": "GV-9", "amount": "5.00"}]}), 403, "pos:voucher:sell")
assert ring(sid, [("COMBO", 1)], who=seller_only)["gross"] == "9.20" # ordinary sales are theirs
def test_pos_13_a_voided_receipt_gives_the_voucher_its_balance_back(engine, nb):
sid = a_till(nb)
sell_voucher(sid, "GV-1", "25.00")
r = must(redeem(ring(sid, [("COMBO", 1)]), "GV-1", "9.20"))
assert voucher("GV-1")["balance"] == "15.80"
must(api("POST", f"/receipts/{r['receipt_id']}/void", CASHIER, {"reason": "changed their mind"}))
v = voucher("GV-1")
assert v["balance"] == "25.00"
assert [(c["claim_no"], c["operation"], c["amount"]) for c in v["claims"]] == [(1, "issue", "25.00"), (2, "redeem", "-9.20"), (3, "refund_to_voucher", "9.20")]
with engine.begin() as conn:
keys = sorted(k for (k,) in conn.execute(select(pos_claim.c.claim_key)) if k.startswith("voucher:"))
assert keys == ["voucher:GV-1:1", "voucher:GV-1:2", "voucher:GV-1:3"] # each operation its own key, none a pay: key
assert not any(k.startswith("pay:") and "GV-1" in k for k in keys)
def test_pos_13_two_tills_spending_one_voucher_at_once_never_both_succeed(engine, nb):
sid = a_till(nb)
sell_voucher(sid, "GV-1", "10.00")
r = ring(sid, [("BRUNCH", 1)])
with engine.begin() as conn: # another till claimed the voucher's claim 2 a moment ago
ClaimRepository(conn).claim("voucher:GV-1:2", "voucher", "x", "racer", datetime(2026, 11, 14))
body = refused(redeem(r, "GV-1", "10.00"), 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 voucher("GV-1")["balance"] == "10.00" # nothing of ours stuck: no line, no claim
def test_pos_13_a_voucher_is_seen_by_the_companys_shops_only(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-1", "10.00")
make_shop("DE-SHOP", registers=("R1",), company="DE01", staff=(as_role("cashier", "de@shop.example"),))
assert api("GET", "/vouchers/GV-1", as_role("cashier", "de@shop.example"))[0] == 404
assert api("GET", "/vouchers/GV-1", ACCOUNTANT)[0] == 200 and api("GET", "/vouchers/GV-1", CASHIER2)[0] == 200
def test_pos_13_the_validity_is_a_setting(nb):
sid = a_till(nb)
must(api("POST", "/settings/voucher_validity_months", OPS, {"value": 12}))
sell_voucher(sid, "GV-1", "10.00")
assert voucher("GV-1")["expires_on"] == "2027-11-14"
for bad in (0, 121, "12", True, 1.5):
assert api("POST", "/settings/voucher_validity_months", OPS, {"value": bad, "valid_from": "2026-12-01"})[0] == 422
def test_pos_13_the_journal_has_the_voucher_liability_both_ways(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-UK-000813", "25.00") # 25.00 by card: a liability, not revenue
finalise(must(redeem(ring(sid, [("COMBO", 1)]), "GV-UK-000813", "9.20"))) # 9.20 of goods paid by the voucher
must(api("POST", f"/sessions/{sid}/count", CASHIER, {"counted_cash": "150.00"}))
must(api("POST", f"/sessions/{sid}/post", ACCOUNTANT, {}))
[body] = nb.journal_calls
lines = sorted((ln["purpose"], ln.get("tax_code", ""), ln.get("debit", ""), ln.get("credit", "")) for ln in body["lines"])
assert lines == sorted([("card_clearing", "", "25.00", ""), ("voucher_liability", "", "9.20", ""),
("revenue", "", "", "7.67"), ("tax", "S", "", "1.53"), ("voucher_liability", "", "", "25.00")])
assert sum(Decimal(ln.get("debit", "0")) for ln in body["lines"]) == sum(Decimal(ln.get("credit", "0")) for ln in body["lines"]) == Decimal("34.20")
@pytest.mark.parametrize("value", ["-1.00", "0", "1e20", "10000000000000000.00", "10.001", "abc", "", "NaN", 1.5, True])
def test_every_voucher_amount_the_books_cannot_hold_is_422_bounds(nb, value):
sid = a_till(nb)
status, body = api("POST", "/receipts", CASHIER, {"session_id": sid, "vouchers": [{"voucher_code": "GV-1", "amount": value}]})
assert status == 422 and body.get("gate") == "bounds", (status, body)
sell_voucher(sid, "GV-OK", "20.00")
r = ring(sid, [("COMBO", 1)])
status, body = redeem(r, "GV-OK", value)
assert status == 422 and body.get("gate") == "bounds", (status, body)
assert api("POST", "/receipts", CASHIER, {"session_id": sid, "vouchers": [{"voucher_code": "bad code/1", "amount": "5.00"}]})[0] == 422
def test_redeeming_costs_the_same_statements_with_50_unrelated_vouchers_and_receipts_around(engine, nb):
sid = a_till(nb)
sell_voucher(sid, "GV-A", "100.00")
sell_voucher(sid, "GV-B", "100.00")
with counting(engine) as quiet:
must(redeem(ring(sid, [("COMBO", 1)]), "GV-A", "9.20"))
other = open_till(register="R2", who=CASHIER2)["session_id"]
for i in range(50):
sell_voucher(other, f"GV-N{i:02d}", "5.00", who=CASHIER2)
with counting(engine) as busy:
must(redeem(ring(sid, [("COMBO", 1)]), "GV-B", "9.20"))
assert len(quiet) == len(busy), (len(quiet), len(busy))
assert (voucher("GV-A")["balance"], voucher("GV-B")["balance"]) == ("90.80", "90.80") # each voucher's balance is its own claims
# ── POS-09: one bill across a voucher, cash and a card ──────────────────────────────────────────────────
def test_pos_09_a_bill_is_split_across_a_voucher_cash_and_a_card_with_a_tip(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-UK-000812", "15.00")
r = ring(sid, [("BRUNCH", 1)]) # 23.80
r = must(redeem(r, "GV-UK-000812", "15.00"))
r = pay(r, "cash", "5.00")
r = pay(r, "card", "3.80", tip="1.00", terminal_txn_ref="T-TXN-77", terminal_status="authorised")
assert [(p["pay_no"], p["method"], p["amount"], p["tip_amount"]) for p in r["payments"]] == \
[(1, "voucher", "15.00", "0.00"), (2, "cash", "5.00", "0.00"), (3, "card", "3.80", "1.00")]
assert r["payments"][2]["idempotency_key"] == f"{r['receipt_id']}:3" and r["remaining"] == "0.00" and r["tips"] == "1.00"
done = finalise(r)
assert done["status"] == "finalised" and voucher("GV-UK-000812")["balance"] == "0.00"
assert voucher("GV-UK-000812")["state"] == "exhausted"
r2 = ring(sid, [("COMBO", 1)])
refused(redeem(r2, "GV-UK-000812", "0.01"), 409, "holds 0.00", gate="voucher_balance")
def test_pos_09_a_declined_card_leaves_the_receipt_open_with_the_other_lines_intact(nb):
sid = a_till(nb)
sell_voucher(sid, "GV-1", "15.00")
r = must(redeem(ring(sid, [("BRUNCH", 1)]), "GV-1", "15.00"))
r = pay(r, "cash", "5.00")
r = pay(r, "card", "3.80", tip="1.00", terminal_txn_ref="T-1", terminal_status="declined")
assert (r["status"], r["paid"], r["remaining"], r["tips"]) == ("open", "20.00", "3.80", "0.00")
assert [p["terminal_status"] for p in r["payments"]] == ["", "", "declined"]
refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {"version": r["version"]}), 409, gate="paid_in_full")
done = finalise(pay(r, "cash", "3.80")) # the customer pays the rest in cash
assert done["status"] == "finalised" and voucher("GV-1")["balance"] == "0.00"
# ── POS-09: a terminal timeout ──────────────────────────────────────────────────────────────────────────
def test_pos_09_a_terminal_timeout_leaves_the_line_pending_and_nothing_else_is_taken_until_it_is_settled(nb):
sid = a_till(nb)
r = ring(sid, [("LATTE-REG", 1)])
r = pay(r, "card", "3.60", terminal_status="pending")
[p] = r["payments"]
assert (p["terminal_status"], p["terminal_txn_ref"], p["idempotency_key"]) == ("pending", "", f"{r['receipt_id']}:1")
assert (r["paid"], r["remaining"], r["status"]) == ("0.00", "3.60", "open") # pending is not money
refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "card", "amount": "3.60", "terminal_txn_ref": "T-2",
"terminal_status": "authorised"}), 409,
f"{r['receipt_id']}:1", gate="terminal_pending") # no second charge while the first is unknown
refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "cash", "amount": "3.60"}), 409, gate="terminal_pending")
refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}), 409, gate="paid_in_full")
refused(api("POST", f"/receipts/{r['receipt_id']}/void", CASHIER, {"reason": "x"}), 409, "has not answered", gate="terminal_pending")
def settle(r, pay_no, status, ref=None, who=None, **more):
return api("POST", f"/receipts/{r['receipt_id']}/payments/{pay_no}/outcome", who or CASHIER,
{"terminal_status": status, **({"terminal_txn_ref": ref} if ref else {}), **more})
def test_pos_09_the_terminals_answer_by_the_lines_key_authorises_it_and_the_receipt_finalises(nb):
sid = a_till(nb)
r = pay(ring(sid, [("LATTE-REG", 1)]), "card", "3.60", tip="0.50", terminal_status="pending")
settled = must(settle(r, 1, "authorised", "T-TXN-9"))
assert [(p["terminal_status"], p["terminal_txn_ref"], p["tip_amount"]) for p in settled["payments"]] == [("authorised", "T-TXN-9", "0.50")]
assert (settled["paid"], settled["remaining"], settled["tips"]) == ("3.60", "0.00", "0.50")
assert finalise(settled)["status"] == "finalised"
def test_pos_09_a_declined_answer_keeps_no_tip_and_the_retry_is_a_new_line_with_its_own_key(nb):
sid = a_till(nb)
r = pay(ring(sid, [("LATTE-REG", 1)]), "card", "3.60", tip="0.50", terminal_status="pending")
declined = must(settle(r, 1, "declined"))
assert [(p["terminal_status"], p["tip_amount"]) for p in declined["payments"]] == [("declined", "0.00")] and declined["remaining"] == "3.60"
retry = pay(declined, "card", "3.60", terminal_txn_ref="T-NEW", terminal_status="authorised")
assert [p["idempotency_key"] for p in retry["payments"]] == [f"{r['receipt_id']}:1", f"{r['receipt_id']}:2"] # the line is the key
assert finalise(retry)["tips"] == "0.00"
def test_pos_09_a_settled_line_never_goes_back_and_the_same_answer_again_changes_nothing(nb):
sid = a_till(nb)
r = pay(ring(sid, [("LATTE-REG", 1)]), "card", "3.60", terminal_status="pending")
first = must(settle(r, 1, "authorised", "T-1"))
again = must(settle(first, 1, "authorised", "T-1"))
assert again["version"] == first["version"]
refused(settle(first, 1, "declined"), 409, "already authorised: the terminal's answer is final", gate="terminal_final")
refused(settle(first, 1, "authorised", "T-OTHER"), 409, gate="terminal_final")
assert must(api("GET", f"/receipts/{r['receipt_id']}", CASHIER))["payments"][0]["terminal_status"] == "authorised"
def test_pos_09_settling_needs_the_right_line_an_answer_a_reference_and_an_open_receipt(nb):
sid = a_till(nb)
r = pay(ring(sid, [("LATTE-REG", 1)], who=CASHIER), "card", "3.60", terminal_status="pending")
cash = ring(sid, [("CROISSANT", 1)])
cash = pay(cash, "cash", "2.40")
refused(settle(cash, 1, "authorised", "T"), 409, "no card payment line 1", gate="payment_line")
refused(settle(r, 9, "authorised", "T"), 409, "no card payment line 9", gate="payment_line")
refused(settle(r, 1, "pending"), 409, "authorised or declined", gate="terminal_status")
refused(settle(r, 1, "authorised"), 409, "names the terminal's transaction", gate="terminal_ref")
assert settle(r, 0, "declined")[0] == 422 and settle(r, 5000, "declined")[0] == 422
refused(settle(r, 1, "declined", who=ACCOUNTANT), 403, "pos:receipt:sell")
must(api("POST", f"/sessions/{sid}/count", CASHIER, {"counted_cash": "150.00"}))
refused(settle(r, 1, "declined"), 409, "open session", gate="session_open")
def test_pos_09_a_voided_receipt_with_a_resolved_card_is_reversed_there_first(nb):
sid = a_till(nb)
r = pay(ring(sid, [("LATTE-REG", 1)]), "card", "3.60", terminal_status="pending")
r = must(settle(r, 1, "authorised", "T-77"))
refused(api("POST", f"/receipts/{r['receipt_id']}/void", CASHIER, {"reason": "x"}), 409, "still authorised on the terminal", gate="card_reversed")
done = must(api("POST", f"/receipts/{r['receipt_id']}/void", CASHIER, {"reason": "x", "terminal_reversals": ["T-77"]}))
assert done["status"] == "voided" and [p["terminal_status"] for p in done["payments"]] == ["reversed"]