dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_seams.py
"""The neighbours - product (prices, item master), tax (rates) and inventory (goods issues) - each behind one adapter: every
answer is a reply, never an exception. Not installed (409), failed (502), silent (503). And the contract check itself: pos's
calls are held to the callee's signed spec, and the three defects we found in those specs are still defects."""
from datetime import date
from decimal import Decimal
import pytest
from conftest import CASHIER, METHOD_DIR, SYSTEM, api, make_shop, must, open_till, refused, ring, sell
from erp_platform import contract, seam
from brick_pos.domain import outbox
from brick_pos.domain.receipt import Receipt, ReceiptLine
def setup_session():
make_shop()
return open_till()["session_id"]
def test_without_the_item_master_a_session_cannot_open(nb):
make_shop()
seam.unregister_local("product")
refused(api("POST", "/sessions", CASHIER, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": "1.00"}), 409,
"the price list (product) is not part of this installation")
def test_without_tax_a_session_cannot_open_because_pos_prints_no_vat_it_did_not_get(nb):
make_shop()
seam.unregister_local("tax")
refused(api("POST", "/sessions", CASHIER, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": "1.00"}), 409,
"the tax rules (tax) is not part of this installation")
@pytest.mark.parametrize("which,words", [("tax_status", "the tax rules (tax) failed (500"), ("product_status", "failed (500")])
def test_a_neighbour_that_fails_is_a_502_never_a_500_and_the_register_opens_when_it_is_back(nb, which, words):
make_shop()
setattr(nb, which, 500)
refused(api("POST", "/sessions", CASHIER, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": "1.00"}), 502, words)
assert must(api("GET", "/sessions", CASHIER))["sessions"] == [] # nothing of the refused opening stuck
setattr(nb, which, None)
assert ring(open_till()["session_id"], [("LATTE-REG", 1)])["gross"] == "3.60" # and it recovers without a restart
def test_once_open_the_register_sells_with_its_snapshot_whatever_the_neighbours_do(nb):
sid = setup_session()
nb.tax_status, nb.product_status = 500, 500 # both neighbours go down after the register opened
nb.price_calls.clear(), nb.rate_calls.clear(), nb.product_calls.clear()
assert ring(sid, [("LATTE-REG", 1)])["gross"] == "3.60"
assert (nb.price_calls, nb.rate_calls, nb.product_calls) == ([], [], []) # a sale reads nothing
seam.unregister_local("tax") # not even installed
assert ring(sid, [("CROISSANT", 1)])["vat"] == "0.40"
def test_a_product_the_item_master_does_not_know_is_left_out_of_the_snapshot_and_refused_by_name(nb):
make_shop()
nb.price_lists["LIST-UK-SHOP"]["prices"].append(("GHOST", "C62", "1", "1.00")) # priced, but product has no such item
sid = open_till()["session_id"]
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "GHOST", "quantity": "1"}]}), 409,
"GHOST is not on the shop's price list", gate="price_known")
assert "GHOST" not in {p["product_id"] for p in must(api("GET", f"/sessions/{sid}/snapshot", CASHIER))["products"]}
def test_a_price_list_of_thousands_is_read_once_a_page_at_a_time_when_the_session_opens(nb):
make_shop()
for i in range(1100):
nb.price_lists["LIST-UK-SHOP"]["prices"].append((f"Z{i:05d}", "C62", "1", "1.00"))
nb.product_row(f"Z{i:05d}", "hot_drink")
nb.price_calls.clear()
sid = open_till()["session_id"]
assert [c["offset"] for c in nb.price_calls] == [0, 1000] and all(c["limit"] == 1000 for c in nb.price_calls)
nb.price_calls.clear()
ring(sid, [("Z01099", 1)]) # the last product, past the first page: no read at all
assert nb.price_calls == []
def test_the_goods_issue_pos_sends_is_the_one_inventorys_spec_describes():
r = Receipt(receipt_id="LON-SOHO-R1-0001-0001", business_date=date(2026, 11, 14),
lines=[ReceiptLine(1, "LATTE-REG", Decimal("2"), "C62", Decimal("3.60"), Decimal("7.20"), "hot_drink", "S", Decimal("20"))])
body = outbox.goods_issue(r, company_code="UK01", location_id="LON-SOHO/BAR")
assert contract.check("inventory", "POST", "/goods-issues/from-source", body, caller="pos", method_dir=METHOD_DIR) == []
assert contract.check("inventory", "POST", "/goods-issues/from-source", {**body, "source_kind": "sale"}, caller="pos",
method_dir=METHOD_DIR) != [] # the check bites: a wrong kind is refused
assert contract.check("inventory", "POST", "/goods-issues/from-source", {k: v for k, v in body.items() if k != "company_code"},
caller="pos", method_dir=METHOD_DIR) != []
def test_the_contract_check_holds_every_pos_call_to_the_callees_spec(nb):
sid = setup_session()
done = sell(sid, [("LATTE-REG", 1)])
assert done["status"] == "finalised" and len(nb.issue_calls) == 1 and nb.rate_calls and nb.price_calls and nb.product_calls
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0}
# The defects in the callees' specs that conftest allows by name. When one is fixed its test fails: remove the allowance.
def test_gap_product_lists_the_item_read_for_inventory_not_for_pos():
bad = contract.check("product", "GET", "/products/LATTE-REG", {"on": "2026-11-14"}, caller="pos", method_dir=METHOD_DIR)
assert bad and "not for pos" in bad[0]
def test_gap_product_lists_the_unit_conversion_for_others_not_for_pos():
bad = contract.check("product", "GET", "/products/BEANS/convert", {"quantity": "18", "from_unit": "GRM", "to_unit": "KGM"}, caller="pos",
method_dir=METHOD_DIR)
assert bad and "not for pos" in bad[0]
def test_gap_gl_lists_the_company_read_for_others_not_for_pos():
bad = contract.check("gl", "GET", "/companies/UK01", {}, caller="pos", method_dir=METHOD_DIR)
assert bad and "not for pos" in bad[0]
def test_gap_inventory_lists_the_warehouse_and_location_reads_for_others_not_for_pos():
for path, params in (("/warehouses/LON-SOHO", {}), ("/locations", {"usage": "internal", "warehouse_id": "LON-SOHO"})):
bad = contract.check("inventory", "GET", path, params, caller="pos", method_dir=METHOD_DIR)
assert bad and "not for pos" in bad[0], (path, bad)
def test_the_setup_reads_are_held_to_the_schema_of_a_caller_the_owner_does_list():
"""While pos is not listed, conftest checks each gap read as the caller that is (LENDERS): a wrong query is still caught."""
from conftest import LENDERS
assert contract.check("inventory", "GET", "/locations", {"usage": "internal", "warehouse_id": "LON-SOHO"}, caller=LENDERS[("inventory", r"/locations")],
method_dir=METHOD_DIR) == []
assert contract.check("inventory", "GET", "/locations", {"usage": "bogus"}, caller=LENDERS[("inventory", r"/locations")], method_dir=METHOD_DIR) != []
def test_gap_tax_v6_entries_for_pos_sit_under_external_services_not_seams():
bad = contract.check("tax", "GET", "/rates", {"country": "GB", "categories": "hot_drink", "on": "2026-11-14"}, caller="pos",
method_dir=METHOD_DIR)
assert bad and bad[0].startswith("tax: no seam GET /rates")
spec, _ = contract.load_callee("tax", METHOD_DIR)
assert any(e.get("with") == "pos" and e["route"].startswith("GET /rates") for e in spec["external_services"])