dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_closing.py
"""Slice B, item 1: closing the day. Force-close (POS-07), the session's journal to gl (POS-20, POS-29) and the cash bag to payments
(POS-19, POS-26), each written from its Given / When / Then through the doors; the invariants that bite here (claim keys, side effects
only from pending rows after the commit, states that never go back, no self-approval on email AND sub, the business date, bounds,
flat statements)."""
import json
from datetime import datetime
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
CASHIER,
CASHIER2,
LEAD,
LEAD2,
OPS,
ROLES,
SYSTEM,
api,
as_role,
freeze,
make_shop,
must,
open_till,
person,
refused,
sell,
)
from erp_platform.seam import derived_id
from test_statement_budget import counting
from brick_pos.application import use_cases
from brick_pos.infrastructure.tables import ClaimRepository
def count(s, amount, who=None, **more):
return api("POST", f"/sessions/{s['session_id']}/count", who or CASHIER, {"counted_cash": amount, **more})
def move(s, kind, amount, reason=None, who=None):
return api("POST", f"/sessions/{s['session_id']}/cash-moves", who or CASHIER,
{"kind": kind, "amount": amount, **({"reason_code": reason} if reason else {})})
def force(s, counted="210.00", reason="tablet broken", who=None, **more):
return api("POST", f"/sessions/{s['session_id']}/force-close", who or LEAD, {"counted_cash": counted, "reason": reason, **more})
def post(s, who=None):
return api("POST", f"/sessions/{s['session_id']}/post", who or ACCOUNTANT, {})
def get(s, who=None):
return must(api("GET", f"/sessions/{s['session_id']}", who or CASHIER))
def outbox(status=None):
return must(api("GET", "/outbox", LEAD, {"status": status} if status else None))["rows"]
def a_day_of_takings(nb):
"""LON-SOHO R1 opens with 150.00. Cash sale 36.00 with a 2.00 cash tip; card sale 12.00 with a 1.00 tip; a paid-out of 12.00 (MILK) and a
safe drop of 20.00: the drawer should hold 156.00. The cashier counts 149.50, 6.50 short (above the 5.00 threshold), a shift lead approves."""
make_shop()
s = open_till(opening_float="150.00")
sid = s["session_id"]
sell(sid, [("LATTE-REG", 10)], tip="2.00")
sell(sid, [("CROISSANT", 5)], method="card", tip="1.00")
must(move(s, "paid_out", "12.00", "MILK", who=LEAD))
must(move(s, "safe_drop", "20.00"))
counted = must(count(s, "149.50"))
assert counted["status"] == "counted" and counted["expected_cash"] == "156.00" and counted["difference"] == "-6.50"
closed = must(api("POST", f"/sessions/{sid}/approve-difference", LEAD, {"reason": "till short, recounted twice"}))
assert closed["status"] == "closed"
return s
# ── POS-07: force-close a session whose register died ───────────────────────────────────────────────────
def a_dying_register(nb):
"""R2 holds local receipts up to 0005; the server holds 0001 and 0002. A shift lead other than the opener force-closes it."""
make_shop()
s = open_till(register="R2", who=CASHIER2)
sell(s["session_id"], [("LATTE-REG", 1)], who=CASHIER2)
sell(s["session_id"], [("CROISSANT", 1)], who=CASHIER2)
return s
def test_pos_07_force_close_records_the_claimed_gap_and_the_day_can_close(nb):
s = a_dying_register(nb)
done = must(force(s, "155.00", "tablet broken", register_last_receipt_no=5, float_left="100.00"))
assert done["status"] == "force_closed" and done["last_receipt_no"] == 2
assert (done["claimed_gap_from"], done["claimed_gap_to"]) == (3, 5)
assert (done["counted_cash"], done["expected_cash"], done["difference"], done["closing_float"]) == ("155.00", "156.00", "-1.00", "100.00")
assert (done["approved_by"], done["reason"], done["counted_by"]) == ("shift_lead@shop.example", "tablet broken", "shift_lead@shop.example")
refused(api("POST", "/receipts", CASHIER2, {"session_id": s["session_id"], "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}), 409,
"sales go in an open session", gate="session_open") # the closed session takes no new receipt
nxt = open_till(register="R2", who=CASHIER2, opening_float="100.00")
assert nxt["session_id"] == "LON-SOHO-R2-0002" and nxt["float_difference"] == "0.00" # the next float is compared with what this one left
must(post(s))
assert get(s)["status"] == "posted" # the day posts without the receipts that never came
def test_pos_07_no_gap_when_the_register_and_the_server_agree(nb):
s = a_dying_register(nb)
done = must(force(s, "156.00", register_last_receipt_no=2))
assert (done["claimed_gap_from"], done["claimed_gap_to"], done["last_receipt_no"]) == (None, None, 2)
def test_pos_07_a_force_close_needs_the_permission_a_reason_and_a_register_number_the_server_can_believe(nb):
s = a_dying_register(nb)
refused(force(s, who=CASHIER2), 403, "pos:session:force_close")
refused(force(s, reason=" "), 409, "needs the reason", gate="reason_required")
refused(force(s, register_last_receipt_no=1), 422, "cannot be lower", gate="bounds")
refused(force(s, float_left="999.00"), 409, "float left", gate="bounds")
assert get(s)["status"] == "open" # nothing of the refused attempts stuck
def test_pos_07_the_person_whose_drawer_it_is_cannot_force_it_closed_under_either_identity(nb):
make_shop()
s = open_till(register="R2", who=LEAD) # the lead opened this drawer
refused(force(s, who=LEAD), 409, "you opened this drawer", gate="second_person")
second_login = person("lead.other-address@shop.example", ROLES["shift_lead"], sub="shift_lead@shop.example") # same sub, another email
refused(force(s, who=second_login), 409, "you opened this drawer", gate="second_person")
same_address = person("shift_lead@shop.example", ROLES["shift_lead"], sub="another-sub") # same email, another sub
must(api("POST", "/shops/LON-SOHO/staff", OPS, {"sub": "another-sub", "email": "shift_lead@shop.example"}))
refused(force(s, who=same_address), 409, gate="second_person")
assert must(force(s, who=LEAD2))["status"] == "force_closed" # another shift lead can
def test_pos_07_a_force_closed_session_never_goes_back_and_the_same_force_close_again_changes_nothing(nb):
s = a_dying_register(nb)
first = must(force(s, "155.00"))
again = must(force(s, "155.00"))
assert again["version"] == first["version"]
refused(force(s, "100.00"), 409, "is force_closed", gate="session_open") # another count is another document, not an edit
refused(count(s, "155.00", who=CASHIER2), 409, "never goes back to open", gate="session_open")
refused(force(s, who=LEAD2, reason="second thoughts"), 409, gate="session_open")
# ── POS-20 / POS-29: the session's journal to gl ────────────────────────────────────────────────────────
def lines_of(body):
return sorted((ln["purpose"], ln.get("tax_code", ""), ln.get("debit", ""), ln.get("credit", "")) for ln in body["lines"])
def test_pos_20_the_session_posts_one_balanced_journal_with_tips_as_a_liability(nb):
s = a_day_of_takings(nb)
must(post(s))
posted = get(s) # gl is told after the commit: read the session again
assert posted["status"] == "posted" and posted["journal_id"] == str(derived_id(f"pos:{s['session_id']}", "day_close"))
[body] = nb.journal_calls
assert (body["source_module"], body["event"], body["source_document"], body["company_code"], body["currency"],
body["journal_date"]) == ("pos", "day_close", f"pos:{s['session_id']}", "UK01", "GBP", "2026-11-14")
assert lines_of(body) == sorted([
("pos_cash_in_till", "", "38.00", ""), # the cash taken, with its 2.00 tip
("card_clearing", "", "13.00", ""), # the card takings with their 1.00 tip
("cash_over_short", "", "6.50", ""), # the drawer was 6.50 short
("shop_supplies", "", "12.00", ""), # the milk paid out of the drawer
("pos_cash_in_safe", "", "20.00", ""), # the safe drop
("revenue", "", "", "40.00"), # 48.00 of sales net of 8.00 VAT
("tax", "S", "", "8.00"),
("tips_payable", "", "", "3.00"), # tips are staff's money, never revenue
("pos_cash_in_till", "", "", "12.00"), ("pos_cash_in_till", "", "", "20.00"), ("pos_cash_in_till", "", "", "6.50")])
debit = sum(Decimal(ln.get("debit", "0")) for ln in body["lines"])
credit = sum(Decimal(ln.get("credit", "0")) for ln in body["lines"])
assert debit == credit == Decimal("89.50")
assert list(nb.journals) == [posted["journal_id"]] # gl holds exactly one journal
def test_pos_20_tips_can_be_the_customers_choice_of_revenue(nb):
make_shop()
must(api("POST", "/settings/tips_treatment", OPS, {"value": "revenue"}))
s = open_till()
sell(s["session_id"], [("LATTE-REG", 10)], tip="2.00")
must(count(s, "188.00"))
must(post(s))
[body] = nb.journal_calls
assert [ln["purpose"] for ln in body["lines"] if ln["purpose"] == "tips_payable"] == []
assert [ln["credit"] for ln in body["lines"] if ln["purpose"] == "revenue"] == ["32.00"] # 30.00 net of VAT plus the 2.00 tip
def test_pos_20_the_retry_after_a_timeout_is_gls_replay_and_posts_once(nb):
s = a_day_of_takings(nb)
nb.gl_status = 500 # gl does not answer
must(post(s))
pending = get(s)
assert pending["status"] == "closed" and pending["posting"]["status"] == "pending" and "gl answered 500" in pending["posting"]["last_error"]
assert nb.journals == {}
nb.gl_status = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 1, "pending": 0, "failed": 0}
assert get(s)["status"] == "posted" and len(nb.journals) == 1
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0} # nothing twice
assert must(post(s))["status"] == "posted" and len(nb.journal_calls) == 2 # asking again sends nothing new
def test_pos_20_gl_is_called_only_from_a_pending_row_after_the_commit(monkeypatch, nb):
monkeypatch.setattr(use_cases, "after_response", lambda *a, **k: None) # the process dies after the commit
s = a_day_of_takings(nb)
must(post(s))
assert nb.journal_calls == [] and [(r["target"], r["status"]) for r in outbox("pending") if r["target"] == "gl.journal"] == [("gl.journal", "pending")]
assert get(s)["status"] == "closed"
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] >= 1
assert get(s)["status"] == "posted" and len(nb.journal_calls) == 1
def test_pos_20_a_session_that_is_not_closed_is_not_posted(nb):
make_shop()
s = open_till()
refused(post(s), 409, "only a closed session is posted", gate="session_closed")
sell(s["session_id"], [("LATTE-REG", 10)])
must(count(s, "100.00")) # counted, waiting for a second person
refused(post(s), 409, gate="session_closed")
assert nb.journal_calls == [] and all(r["target"] != "gl.journal" for r in outbox())
def test_pos_20_a_closed_gl_period_leaves_the_session_waiting_and_a_person_requeues_it(nb):
s = a_day_of_takings(nb)
nb.gl_status = 409
must(post(s))
waiting = get(s)
assert waiting["status"] == "closed" and waiting["posting"]["status"] == "failed"
assert "period_open" in waiting["posting"]["last_error"]
assert len(nb.journal_calls) == 1 and nb.journal_calls[0]["journal_date"] == "2026-11-14" # never into the next period
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0} # failed rows wait for a person
refused(api("POST", "/outbox/requeue", CASHIER, {"target": "gl.journal", "ref": s["session_id"]}), 403, "pos:day:post")
nb.gl_status = None # finance reopened the period
back = must(api("POST", "/outbox/requeue", ACCOUNTANT, {"target": "gl.journal", "ref": s["session_id"]}))
assert back["status"] == "pending" and get(s)["status"] == "posted" # the requeue is dispatched after its commit
assert [c["journal_date"] for c in nb.journal_calls] == ["2026-11-14", "2026-11-14"] and len(nb.journals) == 1
refused(api("POST", "/outbox/requeue", ACCOUNTANT, {"target": "gl.journal", "ref": s["session_id"]}), 409, "never goes back to pending",
gate="outbox_sent")
assert api("POST", "/outbox/requeue", ACCOUNTANT, {"target": "gl.journal", "ref": "nope"})[0] == 404
def test_pos_20_a_posted_session_never_goes_back(nb):
s = a_day_of_takings(nb)
must(post(s))
refused(count(s, "149.50"), 409, "never goes back to open", gate="session_open")
refused(move(s, "safe_drop", "1.00"), 409, gate="session_open")
refused(force(s), 409, gate="session_open")
assert must(post(s))["status"] == "posted" and len(nb.journals) == 1
def accountant_who_approved(sub_of=LEAD):
"""A person holding the posting permission AND the approval one, as a small shop's owner-operator does."""
return person("shift_lead@shop.example", [*ROLES["shift_lead"], "pos:day:post"])
def test_pos_20_whoever_approved_the_difference_does_not_post_it_under_either_identity(nb):
s = a_day_of_takings(nb)
refused(post(s, accountant_who_approved()), 409, "you approved this session's difference", gate="reviewer_not_approver")
second_login = person("lead.second@shop.example", [*ROLES["shift_lead"], "pos:day:post"], sub="shift_lead@shop.example")
refused(post(s, second_login), 409, gate="reviewer_not_approver")
assert nb.journal_calls == [] and get(s)["status"] == "closed"
must(post(s)) # another person, the accountant, posts it
assert get(s)["status"] == "posted"
def test_pos_20_a_shift_lead_cannot_post_and_a_cashier_sees_only_their_shop(nb):
s = a_day_of_takings(nb)
refused(post(s, LEAD), 403, "pos:day:post")
refused(post(s, CASHIER), 403, "pos:day:post")
kx = as_role("accountant", "kx.acc@shop.example")
assert api("POST", f"/sessions/{s['session_id']}/post", kx, {})[0] == 200 # an accountant sees every shop
nb.price_list("LIST-UK-KX", "GBP", [("LATTE-REG", "C62", "1", "3.60")])
other = as_role("cashier", "kx@shop.example")
make_shop("LON-KX", registers=("R1",), staff=(other,), price_list="LIST-UK-KX")
mine = open_till("LON-KX", "R1", who=other)
assert api("POST", f"/sessions/{mine['session_id']}/force-close", LEAD, {"counted_cash": "1.00", "reason": "x"})[0] == 404 # not LEAD's shop
def test_pos_20_the_journal_date_is_the_business_date_not_the_utc_date(nb):
make_shop()
freeze("2026-07-16T00:30:00+00:00") # 01:30 BST on the 16th: still the 15th's business day
s = open_till()
assert s["business_date"] == "2026-07-15"
sell(s["session_id"], [("LATTE-REG", 1)])
must(count(s, "153.60"))
freeze("2026-07-16T09:00:00+00:00")
must(post(s))
assert nb.journal_calls[0]["journal_date"] == "2026-07-15"
def test_pos_20_a_session_that_took_nothing_has_no_journal_and_a_shop_without_a_company_cannot_post(nb):
make_shop()
s = open_till()
must(count(s, "150.00"))
refused(post(s), 409, "took nothing", gate="nothing_to_post")
must(api("POST", "/shops", OPS, {"shop_id": "NOCO", "price_list_id": "LIST-UK-SHOP", "country": "GB"})) # no company_code
must(api("POST", "/shops/NOCO/registers", OPS, {"register_id": "R1"}))
must(api("POST", "/shops/NOCO/staff", OPS, {"sub": "cashier@shop.example", "email": "cashier@shop.example"}))
t = open_till("NOCO", "R1")
sell(t["session_id"], [("LATTE-REG", 1)])
must(count(t, "153.60"))
refused(post(t), 409, "has no company", gate="company_code")
def test_pos_20_the_journal_is_derived_from_its_own_session_whatever_else_is_selling(engine, nb):
"""Statement count is flat, and the journal does not change, with 20 receipts in the session or with 50 unrelated ones elsewhere."""
make_shop()
make_shop("LON-KX", registers=("R1",))
mine = {}
for n in (2, 20):
s = open_till(register="R1" if n == 2 else "R2", who=CASHIER2 if n == 20 else CASHIER)
for _ in range(n):
sell(s["session_id"], [("LATTE-REG", 1)], who=CASHIER2 if n == 20 else CASHIER)
must(count(s, format(Decimal("150.00") + Decimal("3.60") * n, "f"), who=CASHIER2 if n == 20 else CASHIER))
mine[n] = s
costs, bodies = {}, {}
for n, s in mine.items():
with counting(engine) as seen:
must(post(s))
costs[n] = len(seen)
bodies[n] = next(b for b in nb.journal_calls if b["source_document"] == f"pos:{s['session_id']}")
assert costs[2] == costs[20], costs
assert Decimal(next(ln["credit"] for ln in bodies[20]["lines"] if ln["purpose"] == "revenue")) == Decimal("60.00") # 72.00 less 12.00 VAT
kx = open_till("LON-KX", "R1")
for _ in range(50):
sell(kx["session_id"], [("FLAT-WHITE", 1)]) # 50 unrelated receipts in another shop
s3 = open_till(register="R1") # a new session after the 50 unrelated receipts
sell(s3["session_id"], [("LATTE-REG", 2)])
must(count(s3, "157.20"))
with counting(engine) as seen3:
must(post(s3))
assert len(seen3) == costs[2], (len(seen3), costs) # the same statements as with nothing else around
body3 = next(b for b in nb.journal_calls if b["source_document"] == f"pos:{s3['session_id']}")
assert [ln["credit"] for ln in body3["lines"] if ln["purpose"] == "revenue"] == ["6.00"] # its own 7.20, net of 1.20 VAT, nothing else
# ── POS-19 / POS-26: the cash bag for the bank ──────────────────────────────────────────────────────────
def a_safe_holding_600(nb):
"""LON-SOHO's safe holds 600.00 from safe drops on two days."""
make_shop()
s1 = open_till()
must(move(s1, "safe_drop", "350.00"))
must(count(s1, "150.00", float_left="150.00")) # the drawer holds 150.00 after the drop
freeze("2026-11-15T09:00:00+00:00")
s2 = open_till(opening_float="150.00")
must(move(s2, "safe_drop", "250.00"))
return s2
def bag(shop="LON-SOHO", bag_no="B-00451", amount="600.00", who=None, **more):
return api("POST", f"/shops/{shop}/cash-bags", who or LEAD, {"bag_no": bag_no, "amount": amount, "bank_account_id": "BARC-SHOP-UK", **more})
def test_pos_19_a_bag_is_sealed_from_the_safe_and_handed_to_payments(nb):
a_safe_holding_600(nb)
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["safe_balance"] == "600.00"
sealed = must(bag(expected_on="2026-11-16", denominations={"20.00": 20, "50.00": 4}))
assert (sealed["deposit_ref"], sealed["amount"], sealed["status"], sealed["sealed_by"]) == \
("dep:pos:LON-SOHO:B-00451", "600.00", "sealed", "shift_lead@shop.example") # sealed now; handed over after the commit
[call] = nb.batch_calls
assert call == {"batch_ref": "dep:pos:LON-SOHO:B-00451", "kind": "cash", "bank_account_id": "BARC-SHOP-UK", "company_code": "UK01",
"currency": "GBP", "amount": "600.00", "expected_on": "2026-11-16", "clearing_purpose": "cash_in_transit",
"source_module": "pos"}
listing = must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))
assert listing["safe_balance"] == "0.00" and [b["status"] for b in listing["bags"]] == ["handed_over"]
assert json.loads(listing["bags"][0]["denominations"]) == {"20.00": 20, "50.00": 4}
def test_pos_19_the_default_day_is_the_next_business_day_in_business_time(nb):
a_safe_holding_600(nb)
freeze("2026-07-16T00:30:00+00:00") # 01:30 BST on the 16th: the 15th's business day
must(bag(amount="100.00"))
assert nb.batch_calls[0]["expected_on"] == "2026-07-16" # not the UTC date plus one
def test_pos_19_a_bag_is_never_more_than_the_safe_holds_and_is_sealed_once(nb):
a_safe_holding_600(nb)
refused(bag(amount="600.01"), 409, "the safe of LON-SOHO holds 600.00", gate="safe_balance")
must(bag(amount="400.00"))
refused(bag(bag_no="B-00452", amount="200.01"), 409, "holds 200.00", gate="safe_balance") # what the first bag took is gone
again = must(bag(amount="400.00")) # the same bag again is the same bag
assert again["deposit_ref"] == "dep:pos:LON-SOHO:B-00451" and len(nb.batch_calls) == 1
refused(bag(amount="300.00"), 409, "a bag is sealed once", gate="bag_sealed")
assert sorted(nb.batches) == ["dep:pos:LON-SOHO:B-00451"]
def test_pos_19_a_float_topped_up_from_the_safe_is_not_in_the_bag(nb):
s = a_safe_holding_600(nb)
must(move(s, "float_top_up", "100.00"))
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["safe_balance"] == "500.00"
refused(bag(amount="600.00"), 409, gate="safe_balance")
def test_pos_19_two_bags_never_drain_one_safe_twice(engine, nb):
a_safe_holding_600(nb)
with engine.begin() as conn: # a racer sealed the shop's first bag a moment ago
ClaimRepository(conn).claim("safe:LON-SOHO:1", "bag", "", "racer", datetime(2026, 11, 15))
refused(bag(amount="600.00"), 409, "claimed a moment ago", gate="claim")
assert nb.batch_calls == [] and must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["bags"] == []
def test_pos_19_the_claims_of_a_bag_are_its_own_keys(engine, nb):
from sqlalchemy import select
from brick_pos.infrastructure.tables import pos_claim
a_safe_holding_600(nb)
must(bag(amount="100.00"))
with engine.begin() as conn:
keys = sorted(r[0] for r in conn.execute(select(pos_claim.c.claim_key)))
assert "bag:LON-SOHO:B-00451" in keys and "safe:LON-SOHO:1" in keys
assert not any(k.startswith(("pay:", "refund:", "voucher:")) and "B-00451" in k for k in keys)
def test_pos_26_payments_down_leaves_the_bag_sealed_and_the_row_pending_until_it_is_back(nb):
a_safe_holding_600(nb)
nb.payments_status = 500
sealed = must(bag())
assert sealed["status"] == "sealed" # the shift lead has their sealed bag
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["bags"][0]["status"] == "sealed"
[row] = [r for r in outbox("pending") if r["target"] == "payments.deposit_batch"]
assert row["ref"] == "dep:pos:LON-SOHO:B-00451" and "payments answered 500" in row["last_error"]
nb.payments_status = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] == 1
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["bags"][0]["status"] == "handed_over"
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0}
assert len(nb.batches) == 1
def test_pos_26_a_bag_payments_refuses_is_failed_with_its_reason_and_never_unsealed(nb):
a_safe_holding_600(nb)
nb.payments_status = 422
sealed = must(bag())
assert sealed["status"] == "sealed"
[row] = outbox("failed")
assert "bank_account_receives" in row["last_error"]
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["bags"][0]["status"] == "sealed" # a bag is never unsealed
nb.payments_status = None
must(api("POST", "/outbox/requeue", ACCOUNTANT, {"target": "payments.deposit_batch", "ref": "dep:pos:LON-SOHO:B-00451"}))
assert must(api("GET", "/shops/LON-SOHO/cash-bags", LEAD))["bags"][0]["status"] == "handed_over"
def test_pos_19_who_may_seal_and_which_shop(nb):
a_safe_holding_600(nb)
refused(bag(who=CASHIER), 403, "pos:deposit:prepare")
refused(bag(who=ACCOUNTANT), 403, "pos:deposit:prepare")
assert api("POST", "/shops/NOPE/cash-bags", LEAD, {"bag_no": "B1", "amount": "1.00", "bank_account_id": "X"})[0] == 404
stranger = as_role("shift_lead", "stranger@shop.example") # a shift lead who works at no shop
assert bag(shop="LON-SOHO", who=stranger)[0] == 404
BAD = ["-1.00", "0", "0.00", "1e20", "10000000000000000.00", "99999999999999999999999", "10.001", "abc", "", "NaN", "Infinity", 1.5, True]
@pytest.mark.parametrize("value", BAD)
def test_every_amount_of_the_closing_is_422_bounds_never_a_500(nb, value):
a_safe_holding_600(nb)
s = open_till(register="R2", who=CASHIER2)
def bounds(result):
status, body = result
assert status == 422 and body.get("gate") == "bounds", (status, body)
bounds(bag(amount=value))
bounds(api("POST", f"/sessions/{s['session_id']}/force-close", LEAD, {"counted_cash": value, "reason": "x"}) if value not in ("0", "0.00") else (422, {"gate": "bounds"}))
bounds(api("POST", f"/sessions/{s['session_id']}/force-close", LEAD, {"counted_cash": "150.00", "reason": "x", "float_left": value}) if value not in ("0", "0.00") else (422, {"gate": "bounds"}))
@pytest.mark.parametrize("extra", [{"denominations": {"20.00": 4}}, {"denominations": {"abc": 1}}, {"denominations": {"20.00": 0}},
{"denominations": {"20.00": True}}, {"denominations": {}}, {"expected_on": "tomorrow"},
{"denominations": {f"{i}.00": 1 for i in range(1, 40)}}])
def test_a_bags_denominations_and_date_are_bounded_too(nb, extra):
a_safe_holding_600(nb)
status, body = bag(amount="100.00", **extra)
assert status == 422 and body.get("gate") == "bounds", (status, body)
def test_a_bag_number_is_a_code(nb):
a_safe_holding_600(nb)
assert bag(bag_no="B 1 / 2")[0] == 422
def test_sealing_costs_the_same_statements_with_2_or_50_unrelated_receipts_around(engine, nb):
a_safe_holding_600(nb)
with counting(engine) as quiet:
must(bag(bag_no="B-1", amount="100.00"))
make_shop("LON-KX", registers=("R1",))
k = open_till("LON-KX", "R1")
for _ in range(50):
sell(k["session_id"], [("FLAT-WHITE", 1)])
must(move(k, "safe_drop", "1.00"))
with counting(engine) as busy:
must(bag(bag_no="B-2", amount="100.00"))
assert len(quiet) == len(busy), (len(quiet), len(busy))
refused(bag(bag_no="B-3", amount="400.01"), 409, "holds 400.00", gate="safe_balance") # the other shop's drops are not this safe's