dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_snapshot.py
"""Slice B, item 5: the register snapshot (POS-30). When a register opens a session pos reads the shop's prices, the item master's tax
categories and tax's rates ONCE and keeps them as a dated, versioned copy with a digest; the register sells with it (so an offline till keeps
selling), and a price or a rate dated tomorrow is in tomorrow's snapshot, never today's. Written from Given / When / Then through the doors."""
import re
from conftest import (
CASHIER,
CASHIER2,
LEAD,
api,
as_role,
freeze,
make_shop,
must,
open_till,
refused,
ring,
sell,
)
from sqlalchemy import select, text
from test_statement_budget import counting
from brick_pos.infrastructure.tables import pos_register_snapshot
def snap(session_id, who=None):
return must(api("GET", f"/sessions/{session_id}/snapshot", who or CASHIER))
def test_pos_30_a_register_opens_with_the_dated_prices_and_rates_it_read_once(nb):
make_shop()
s = open_till()
assert (s["snapshot_no"], bool(re.fullmatch(r"[0-9a-f]{64}", s["snapshot_digest"]))) == (1, True)
got = snap(s["session_id"])
assert (got["shop_id"], got["business_date"], got["snapshot_no"], got["digest"], got["currency"], got["country"], got["price_list_id"]) == \
("LON-SOHO", "2026-11-14", 1, s["snapshot_digest"], "GBP", "GB", "LIST-UK-SHOP")
latte = next(p for p in got["products"] if p["product_id"] == "LATTE-REG")
assert (latte["unit_code"], latte["unit_price"], latte["tax_category"], latte["name"], latte["pricing_quantity"]) == ("C62", "3.60", "hot_drink", "LATTE-REG", 1)
assert sorted((r["category_code"], r["tax_code"], r["rate_pct"]) for r in got["rates"]) == \
[("food_takeaway", "S", "20.00"), ("hot_drink", "S", "20.00"), ("packaged_food", "Z", "0.00")]
assert {c["on"] for c in nb.price_calls} == {"2026-11-14"} and nb.rate_calls[-1]["on"] == "2026-11-14" and nb.rate_calls[-1]["consumption"] is None
assert nb.rate_calls[-1]["categories"] == "food_takeaway,hot_drink,packaged_food" and len(nb.rate_calls) == 1 # tax asked once, for every category
def test_pos_30_a_price_or_rate_dated_tomorrow_is_in_tomorrows_snapshot_never_todays(nb):
rows = nb.price_lists["LIST-UK-SHOP"]["prices"]
rows[:] = [r for r in rows if r[0] != "LATTE-REG"] + [("LATTE-REG", "C62", "1", "3.60", "2026-01-01"), ("LATTE-REG", "C62", "1", "4.00", "2026-11-15")]
nb.rate("GB", "hot_drink", "S25", "25.00", valid_from="2026-11-15")
make_shop()
today = open_till()
got = snap(today["session_id"])
assert [p["unit_price"] for p in got["products"] if p["product_id"] == "LATTE-REG"] == ["3.60"]
assert got["valid_until"] == "2026-11-15" # refresh before the day a price or rate changes
assert [(r["tax_code"], r["rate_pct"]) for r in got["rates"] if r["category_code"] == "hot_drink"] == [("S", "20.00")]
assert ring(today["session_id"], [("LATTE-REG", 1)])["gross"] == "3.60"
must(api("POST", f"/sessions/{today['session_id']}/count", CASHIER, {"counted_cash": "150.00"}))
freeze("2026-11-15T10:00:00+00:00")
tomorrow = open_till()
got2 = snap(tomorrow["session_id"])
assert (tomorrow["business_date"], tomorrow["snapshot_no"]) == ("2026-11-15", 1) # numbered per business date
assert [p["unit_price"] for p in got2["products"] if p["product_id"] == "LATTE-REG"] == ["4.00"]
assert [(r["tax_code"], r["rate_pct"]) for r in got2["rates"] if r["category_code"] == "hot_drink"] == [("S25", "25.00")]
assert tomorrow["snapshot_digest"] != today["snapshot_digest"]
assert snap(today["session_id"])["digest"] == today["snapshot_digest"] # today's snapshot is not touched
def test_pos_30_registers_that_open_on_the_same_answers_share_one_snapshot_and_a_change_is_the_next_number(nb):
make_shop()
first = open_till(register="R1")
second = open_till(register="R2", who=CASHIER2)
assert (first["snapshot_no"], second["snapshot_no"], first["snapshot_digest"]) == (1, 1, second["snapshot_digest"])
nb.price_lists["LIST-UK-SHOP"]["prices"] = [r for r in nb.price_lists["LIST-UK-SHOP"]["prices"] if r[0] != "LATTE-REG"] + [("LATTE-REG", "C62", "1", "3.80")]
must(api("POST", f"/sessions/{second['session_id']}/count", CASHIER2, {"counted_cash": "150.00"}))
third = open_till(register="R2", who=CASHIER2)
assert (third["snapshot_no"], third["snapshot_digest"] != first["snapshot_digest"]) == (2, True)
def test_pos_30_a_price_changed_after_the_register_opened_never_changes_the_days_sales(nb):
make_shop()
s = open_till()
sid = s["session_id"]
sold = sell(sid, [("LATTE-REG", 1)])
nb.price_lists["LIST-UK-SHOP"]["prices"] = [r for r in nb.price_lists["LIST-UK-SHOP"]["prices"] if r[0] != "LATTE-REG"] + [("LATTE-REG", "C62", "1", "9.99")]
nb.rates["GB"] = [{**r, "tax_code": "S25", "rate_pct": "25.00"} if r["category_code"] == "hot_drink" else r for r in nb.rates["GB"]]
later = ring(sid, [("LATTE-REG", 1)])
assert (later["gross"], later["vat"], later["lines"][0]["tax_code"]) == ("3.60", "0.60", "S") # still this register's snapshot
assert must(api("GET", f"/receipts/{sold['receipt_id']}", CASHIER))["gross"] == "3.60"
other = open_till(register="R2", who=CASHIER2) # a register that opens now reads the new answers
assert ring(other["session_id"], [("LATTE-REG", 1)], who=CASHIER2)["gross"] == "9.99"
def test_pos_30_the_snapshot_follows_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" and snap(s["session_id"])["business_date"] == "2026-07-15"
assert {c["on"] for c in nb.price_calls} == {"2026-07-15"} and {c["on"] for c in nb.rate_calls} == {"2026-07-15"}
def test_pos_30_a_refused_opening_leaves_no_snapshot_behind(engine, nb):
make_shop()
nb.tax_status = 500
assert api("POST", "/sessions", CASHIER, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": "1.00"})[0] == 502
with engine.begin() as conn:
assert conn.execute(select(pos_register_snapshot.c.snapshot_no)).all() == []
def test_pos_30_who_sees_a_snapshot(nb):
make_shop()
make_shop("LON-KX", registers=("R1",), staff=(as_role("cashier", "kx@shop.example"),))
s = open_till()
assert api("GET", f"/sessions/{s['session_id']}/snapshot", as_role("cashier", "kx@shop.example"))[0] == 404
assert api("GET", "/sessions/NOPE/snapshot", CASHIER)[0] == 404
assert api("GET", f"/sessions/{s['session_id']}/snapshot", LEAD)[0] == 200
def test_pos_30_a_session_opened_before_snapshots_is_still_rung_from_the_neighbours(engine, nb):
make_shop()
s = open_till()
with engine.begin() as conn: # as a session of slice A is: no snapshot
conn.execute(text("UPDATE pos_session SET snapshot_no = NULL, snapshot_digest = NULL"))
refused(api("GET", f"/sessions/{s['session_id']}/snapshot", CASHIER), 404, "opened before snapshots")
nb.price_calls.clear()
assert ring(s["session_id"], [("LATTE-REG", 1)])["gross"] == "3.60" and nb.price_calls # it reads, as it did
def test_ringing_costs_the_same_statements_with_50_unrelated_snapshots_around(engine, nb):
make_shop()
sid = open_till()["session_id"]
with counting(engine) as quiet:
ring(sid, [("LATTE-REG", 1)])
for i in range(1, 6):
make_shop(f"SHOP-{i}", registers=("R1",), staff=(CASHIER,))
for day in range(10):
freeze(f"2026-12-{day + 1:02d}T10:00:00+00:00")
who = open_till(f"SHOP-{i}", "R1")
must(api("POST", f"/sessions/{who['session_id']}/count", CASHIER, {"counted_cash": "150.00"}))
freeze("2026-11-14T10:00:00+00:00")
nb.price_calls.clear()
with counting(engine) as busy:
ring(sid, [("LATTE-REG", 1)])
assert nb.price_calls == [] # still from its own snapshot, not another shop's
with engine.begin() as conn:
assert len(conn.execute(select(pos_register_snapshot.c.snapshot_no)).all()) >= 50
assert len(quiet) == len(busy), (len(quiet), len(busy))