dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/test_recipes.py
"""Slice B, item 4: recipes (POS-15) and what a sold drink takes out of stock (POS-16, POS-27, POS-31): the till sells a latte and inventory
is told about its beans, milk, cup and lid - from pos's own dated recipe, or from manufacturing's usage-sale bill of materials. Written
from Given / When / Then through the doors, with the invariants that bite here: the recipe version used is frozen on the line, side
effects only from pending rows after the commit, no backward states, 422 bounds, flat statements."""
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
CASHIER,
LEAD,
METHOD_DIR,
OPS,
SYSTEM,
api,
finalise,
freeze,
make_shop,
must,
open_till,
pay,
refused,
ring,
)
from erp_platform import contract, seam
from test_statement_budget import counting, lines, many_products
from brick_pos.application import use_cases
from brick_pos.domain.recipe import Recipe, RecipeLine
from brick_pos.infrastructure.tables import RecipeRepository
def a_cafe(nb):
"""The coffee shop's stock: beans in kilograms (issued in grams), milk in litres (issued in millilitres), cups and lids."""
nb.product_row("BEANS-HOUSE", "packaged_food", base_unit="KGM", units={"GRM": "0.001"})
nb.product_row("MILK-WHOLE", "packaged_food", base_unit="LTR", units={"MLT": "0.001"})
nb.product_row("MILK-OAT", "packaged_food", base_unit="LTR", units={"MLT": "0.001"})
nb.product_row("CUP-12OZ", "packaged_food")
nb.product_row("LID-12OZ", "packaged_food")
nb.product_row("PAPER-BAG", "packaged_food", stocked=False)
LATTE = {"product_id": "LATTE-REG", "lines": [
{"component_id": "BEANS-HOUSE", "quantity": "0.018", "unit_code": "KGM"},
{"component_id": "MILK-WHOLE", "quantity": "0.250", "unit_code": "LTR"},
{"component_id": "CUP-12OZ", "quantity": "1", "unit_code": "C62"},
{"component_id": "LID-12OZ", "quantity": "1", "unit_code": "C62"}],
"modifiers": [{"modifier_id": "MOD-OAT", "replaces": "MILK-WHOLE", "component_id": "MILK-OAT", "quantity": "0.250", "unit_code": "LTR"}]}
def write(body=None, who=None, **more):
return api("POST", "/recipes", who or OPS, {**(body or LATTE), **more})
def set_source(source, who=None):
must(api("POST", "/settings/recipe_source", who or OPS, {"value": source}))
# ── POS-15: write each drink's recipe once ──────────────────────────────────────────────────────────────
def test_pos_15_a_recipe_is_a_dated_version_with_its_modifier(nb):
a_cafe(nb)
freeze("2026-10-25T10:00:00+00:00")
seam.unregister_local("manufacturing") # manufacturing is not installed
v1 = must(write(valid_from="2026-11-01"))
assert (v1["product_id"], v1["version"], v1["valid_from"]) == ("LATTE-REG", 1, "2026-11-01")
assert [(ln["component_id"], ln["quantity"], ln["unit_code"], ln["modifier_id"], ln["replaces"]) for ln in v1["lines"]] == [
("BEANS-HOUSE", "0.018", "KGM", "", ""), ("MILK-WHOLE", "0.25", "LTR", "", ""), ("CUP-12OZ", "1", "C62", "", ""),
("LID-12OZ", "1", "C62", "", ""), ("MILK-OAT", "0.25", "LTR", "MOD-OAT", "MILK-WHOLE")]
assert must(api("GET", "/recipes", CASHIER)) == {"products": ["LATTE-REG"], "recipe_source": "pos"}
assert must(api("GET", "/recipes/LATTE-REG", CASHIER))["versions"][0]["version"] == 1
assert api("GET", "/recipes/NOPE", CASHIER)[0] == 404
def test_pos_15_every_quantity_converts_exactly_to_its_base_unit_or_is_refused(nb):
a_cafe(nb)
grams = {**LATTE, "lines": [{"component_id": "BEANS-HOUSE", "quantity": "18", "unit_code": "GRM"}], "modifiers": []}
assert must(write(grams))["lines"][0]["unit_code"] == "GRM"
assert nb.convert_calls[-1] == {"product_id": "BEANS-HOUSE", "quantity": "18", "from_unit": "GRM", "to_unit": "KGM"}
half = {**LATTE, "product_id": "CROISSANT", "lines": [{"component_id": "BEANS-HOUSE", "quantity": "0.5", "unit_code": "GRM"}], "modifiers": []}
refused(write(half), 409, "does not convert exactly to KGM", gate="units_exact") # 0.0005 KGM: refused, never rounded
unknown = {**half, "lines": [{"component_id": "BEANS-HOUSE", "quantity": "1", "unit_code": "STONE"}]}
refused(write(unknown), 409, "does not know BEANS-HOUSE")
assert must(api("GET", "/recipes", CASHIER))["products"] == ["LATTE-REG"] # nothing of the refused ones stuck
def test_pos_15_a_component_that_is_not_stocked_or_not_known_is_refused(nb):
a_cafe(nb)
bag = {**LATTE, "lines": [{"component_id": "PAPER-BAG", "quantity": "1", "unit_code": "C62"}], "modifiers": []}
refused(write(bag), 409, "PAPER-BAG is not a stocked product", gate="component_stocked")
ghost = {**LATTE, "lines": [{"component_id": "GHOST", "quantity": "1", "unit_code": "C62"}], "modifiers": []}
refused(write(ghost), 409, "does not know /products/GHOST")
refused(write({**LATTE, "product_id": "GHOST-DRINK"}), 409, "does not know /products/GHOST-DRINK")
swap = {**LATTE, "modifiers": [{**LATTE["modifiers"][0], "replaces": "MILK-SOYA"}]}
refused(write(swap), 409, "not a component of the recipe", gate="replaces_known")
def test_pos_15_a_change_is_the_next_dated_version_never_an_edit_and_the_same_content_is_the_same_version(nb):
a_cafe(nb)
freeze("2026-10-25T10:00:00+00:00")
v1 = must(write(valid_from="2026-11-01"))
assert must(write(valid_from="2026-12-01"))["version"] == 1 # the same content again: still version 1
richer = {**LATTE, "lines": [{**LATTE["lines"][0], "quantity": "0.020"}, *LATTE["lines"][1:]]}
refused(write(richer), 422, "send valid_from", gate="valid_from") # a changed recipe starts after today
refused(write(richer, valid_from="2026-10-25"), 422, "starts after", gate="valid_from")
refused(write(richer, valid_from="2026-11-01"), 422, "starts after 2026-11-01", gate="valid_from") # and after the version before it
v2 = must(write(richer, valid_from="2026-12-01"))
assert (v2["version"], v2["valid_from"], v2["lines"][0]["quantity"]) == (2, "2026-12-01", "0.02")
assert [v["version"] for v in must(api("GET", "/recipes/LATTE-REG", CASHIER))["versions"]] == [1, v1["version"] + 1]
refused(write(valid_from="soon"), 422, "valid_from is a date", gate="bounds")
freeze("2026-11-14T09:00:00+00:00") # version 1 is in force: a change cannot start today either
even_richer = {**richer, "lines": [{**LATTE["lines"][0], "quantity": "0.040"}, *LATTE["lines"][1:]]}
refused(write(even_richer, valid_from="2026-11-14"), 422, "starts after 2026-12-01", gate="valid_from")
refused(write(even_richer, valid_from="2026-12-01"), 422, "starts after 2026-12-01", gate="valid_from")
def test_pos_15_a_version_in_force_is_not_replaced_from_today(nb):
a_cafe(nb)
freeze("2026-10-25T10:00:00+00:00")
must(write(valid_from="2026-11-01"))
freeze("2026-11-14T09:00:00+00:00") # version 1 has been in force for two weeks
richer = {**LATTE, "lines": [{**LATTE["lines"][0], "quantity": "0.020"}, *LATTE["lines"][1:]]}
refused(write(richer, valid_from="2026-11-14"), 422, "starts after 2026-11-14", gate="valid_from")
assert must(write(richer, valid_from="2026-11-15"))["version"] == 2
def test_pos_15_who_may_write_a_recipe(nb):
a_cafe(nb)
refused(write(who=CASHIER), 403, "pos:recipe:write")
refused(write(who=LEAD), 403, "pos:recipe:write")
refused(write(who=ACCOUNTANT), 403, "pos:recipe:write")
assert api("GET", "/recipes", ACCOUNTANT)[0] == 200
BAD = ["-1", "0", "1e20", "10000000000000000", "1.0001", "abc", "", "NaN", "Infinity", 1.5, True, "1000000"]
@pytest.mark.parametrize("value", BAD)
def test_every_recipe_quantity_the_books_cannot_hold_is_422_bounds(nb, value):
a_cafe(nb)
for body in ({**LATTE, "lines": [{"component_id": "CUP-12OZ", "quantity": value, "unit_code": "C62"}], "modifiers": []},
{**LATTE, "modifiers": [{**LATTE["modifiers"][0], "quantity": value}]}):
status, got = write(body)
assert status == 422 and got.get("gate") == "bounds", (status, got)
for body in ({**LATTE, "lines": []}, {**LATTE, "lines": LATTE["lines"] * 8}, {**LATTE, "product_id": "bad code"},
{**LATTE, "lines": [{"component_id": "a/b", "quantity": "1", "unit_code": "C62"}], "modifiers": []}):
assert write(body)[0] in (409, 422)
assert write(body)[0] != 500
# ── POS-16: the till sells a drink, inventory is told about its beans and milk ──────────────────────────
def a_shop_with_the_latte_recipe(nb):
a_cafe(nb)
make_shop()
freeze("2026-10-25T10:00:00+00:00")
must(write(valid_from="2026-11-01"))
freeze("2026-11-14T09:00:00+00:00")
return open_till()["session_id"]
def ring_modified(sid, lines_, who=None):
return must(api("POST", "/receipts", who or CASHIER, {"session_id": sid, "lines": lines_}))
def test_pos_16_a_latte_takes_its_components_out_of_the_shops_stock_and_the_croissant_itself(nb):
sid = a_shop_with_the_latte_recipe(nb)
r = ring_modified(sid, [{"product_id": "LATTE-REG", "quantity": "1"}, {"product_id": "LATTE-REG", "quantity": "1", "modifiers": ["MOD-OAT"]},
{"product_id": "CROISSANT", "quantity": "1"}])
assert r["gross"] == "9.60" # a modifier changes the recipe, never the price
assert [(ln["recipe_source"], ln["recipe_ref"], ln["recipe_version"], ln["modifiers"]) for ln in r["lines"]] == \
[("pos", "LATTE-REG", 1, ""), ("pos", "LATTE-REG", 1, "MOD-OAT"), ("", "", None, "")]
done = finalise(pay(r, "cash", "10.00"))
rid = done["receipt_id"]
[call] = nb.issue_calls
assert call["source_kind"] == "pos_sale" and call["location_id"] == "LON-SOHO/BAR" and call["issued_on"] == "2026-11-14"
assert [(ln["move_ref"], ln["product_id"], ln["quantity"], ln["unit_code"]) for ln in call["lines"]] == [
(f"pos:{rid}:1:1", "BEANS-HOUSE", "0.018", "KGM"), (f"pos:{rid}:1:2", "MILK-WHOLE", "0.25", "LTR"),
(f"pos:{rid}:1:3", "CUP-12OZ", "1", "C62"), (f"pos:{rid}:1:4", "LID-12OZ", "1", "C62"),
(f"pos:{rid}:2:1", "BEANS-HOUSE", "0.018", "KGM"), (f"pos:{rid}:2:2", "CUP-12OZ", "1", "C62"),
(f"pos:{rid}:2:3", "LID-12OZ", "1", "C62"), (f"pos:{rid}:2:4", "MILK-OAT", "0.25", "LTR"),
(f"pos:{rid}:3:1", "CROISSANT", "1", "C62")]
assert must(api("POST", "/outbox/dispatch", SYSTEM, {})) == {"sent": 0, "pending": 0, "failed": 0} # the second dispatch is a replay: nothing
assert len(nb.issue_calls) == 1
def test_pos_16_two_lattes_take_twice_the_components(nb):
sid = a_shop_with_the_latte_recipe(nb)
done = finalise(pay(ring(sid, [("LATTE-REG", 2)]), "cash", "7.20"))
[call] = nb.issue_calls
assert [(ln["product_id"], ln["quantity"]) for ln in call["lines"]] == [("BEANS-HOUSE", "0.036"), ("MILK-WHOLE", "0.5"), ("CUP-12OZ", "2"), ("LID-12OZ", "2")]
assert [(c["component_no"], c["product_id"], c["quantity"]) for c in done["components"]][0] == (1, "BEANS-HOUSE", "0.036")
def test_pos_16_a_later_recipe_never_changes_a_past_sales_consumption(monkeypatch, nb):
sid = a_shop_with_the_latte_recipe(nb)
monkeypatch.setattr(use_cases, "after_response", lambda *a, **k: None) # the stock row waits, pending
old = finalise(pay(ring(sid, [("LATTE-REG", 1)]), "cash", "3.60"))
richer = {**LATTE, "lines": [{**LATTE["lines"][0], "quantity": "0.030"}, *LATTE["lines"][1:]]}
must(write(richer, valid_from="2026-11-15")) # version 2 starts tomorrow
freeze("2026-11-15T09:00:00+00:00")
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] == 1
assert nb.issue_calls[0]["lines"][0]["quantity"] == "0.018" # the old sale still takes version 1's beans
assert (must(api("GET", f"/receipts/{old['receipt_id']}", CASHIER))["lines"][0]["recipe_version"]) == 1
freeze("2026-11-15T09:01:00+00:00")
nxt = open_till(register="R2", who=CASHIER)["session_id"]
new = must(api("POST", "/receipts", CASHIER, {"session_id": nxt, "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}))
assert [(ln["recipe_version"]) for ln in new["lines"]] == [2]
assert [c["quantity"] for c in new["components"]][0] == "0.03"
def test_pos_16_the_stock_is_sent_from_a_pending_row_after_the_commit_and_never_refuses_the_sale(nb):
sid = a_shop_with_the_latte_recipe(nb)
nb.inventory_status = 500
done = finalise(pay(ring(sid, [("LATTE-REG", 1)]), "cash", "3.60"))
assert done["status"] == "finalised"
[row] = must(api("GET", "/outbox", LEAD))["rows"]
assert row["status"] == "pending" and [ln["product_id"] for ln in row["payload"]["lines"]][:2] == ["BEANS-HOUSE", "MILK-WHOLE"]
nb.inventory_status = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] == 1
def test_pos_16_a_modifier_the_recipe_does_not_define_is_refused_not_ignored(nb):
sid = a_shop_with_the_latte_recipe(nb)
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1", "modifiers": ["MOD-SOYA"]}]}),
409, "has no modifier MOD-SOYA", gate="modifier_unknown")
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "CROISSANT", "quantity": "1", "modifiers": ["MOD-OAT"]}]}),
409, "has no recipe", gate="modifier_unknown")
assert must(api("GET", "/receipts", CASHIER, {"session_id": sid}))["receipts"] == []
def test_pos_16_before_a_recipe_starts_the_product_is_consumed_itself(nb):
a_cafe(nb)
make_shop()
freeze("2026-10-25T10:00:00+00:00")
must(write(valid_from="2026-11-20"))
freeze("2026-11-14T09:00:00+00:00")
sid = open_till()["session_id"]
finalise(pay(ring(sid, [("LATTE-REG", 1)]), "cash", "3.60"))
assert [(ln["product_id"], ln["quantity"]) for ln in nb.issue_calls[0]["lines"]] == [("LATTE-REG", "1")] # the recipe is dated tomorrow, not today
def test_pos_16_a_made_item_is_never_restocked_as_its_ingredients(nb):
sid = a_shop_with_the_latte_recipe(nb)
must(api("POST", "/category-rules/hot_drink", OPS, {"restock_on_refund": True})) # even where the category would say yes
sale = finalise(pay(ring(sid, [("LATTE-REG", 1)]), "cash", "3.60"))
refused(api("POST", f"/receipts/{sale['receipt_id']}/refund", CASHIER, {"session_id": sid, "lines": [{"line_no": 1, "quantity": "1", "restock": True}],
"method": "cash", "reason": "x"}), 409, "a made item", gate="restock_refused")
done = must(api("POST", f"/receipts/{sale['receipt_id']}/refund", CASHIER, {"session_id": sid, "lines": [{"line_no": 1, "quantity": "1"}],
"method": "cash", "reason": "cold"}))
assert done["lines"][0]["restock"] is False and nb.receipt_calls == []
# ── POS-31: manufacturing's bill of materials as the recipe source ──────────────────────────────────────
def bom(nb):
nb.bom("LATTE-REG", [("ROAST-HOUSE", "0.018", "KGM"), ("MILK-WHOLE", "0.25", "LTR"), ("CUP-12OZ", "1", "C62"), ("LID-12OZ", "1", "C62")],
valid_from="2026-12-01", modifiers={"MOD-OAT": {"replaces": "MILK-WHOLE", "adds": ("MILK-OAT", "0.25", "LTR")}})
def test_pos_31_manufacturings_bom_is_the_recipe_when_the_setting_says_so(nb):
make_shop()
bom(nb)
set_source("manufacturing")
freeze("2026-12-02T10:00:00+00:00")
sid = open_till()["session_id"]
r = ring_modified(sid, [{"product_id": "LATTE-REG", "quantity": "1", "modifiers": ["MOD-OAT"]}])
assert nb.bom_calls == [{"product_id": "LATTE-REG", "usage": "sale", "quantity": "1", "unit_code": "C62", "modifiers": "MOD-OAT", "on": "2026-12-02"}]
assert [(ln["recipe_source"], ln["recipe_ref"], ln["recipe_version"], ln["modifiers"]) for ln in r["lines"]] == [("manufacturing", "LATTE-REG:sale", 1, "MOD-OAT")]
rid = finalise(pay(r, "cash", "3.60"))["receipt_id"]
assert [(ln["move_ref"], ln["product_id"], ln["quantity"], ln["unit_code"]) for ln in nb.issue_calls[0]["lines"]] == [
(f"pos:{rid}:1:1", "ROAST-HOUSE", "0.018", "KGM"), (f"pos:{rid}:1:2", "CUP-12OZ", "1", "C62"),
(f"pos:{rid}:1:3", "LID-12OZ", "1", "C62"), (f"pos:{rid}:1:4", "MILK-OAT", "0.25", "LTR")]
def test_pos_31_a_product_with_no_released_bom_is_consumed_itself_and_a_date_before_it_starts_too(nb):
make_shop()
bom(nb)
set_source("manufacturing")
sid = open_till()["session_id"] # 2026-11-14: the BOM starts on 2026-12-01
done = finalise(pay(ring(sid, [("LATTE-REG", 1), ("CROISSANT", 1)]), "cash", "6.00"))
assert [(ln["product_id"], ln["quantity"]) for ln in nb.issue_calls[0]["lines"]] == [("LATTE-REG", "1"), ("CROISSANT", "1")]
assert [ln["recipe_source"] for ln in done["lines"]] == ["", ""]
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1", "modifiers": ["MOD-OAT"]}]}),
409, "has no recipe", gate="modifier_unknown")
def test_pos_31_manufacturing_being_down_is_a_502_never_a_500_and_it_recovers(nb):
make_shop()
bom(nb)
set_source("manufacturing")
freeze("2026-12-02T10:00:00+00:00")
sid = open_till()["session_id"]
nb.manufacturing_status = 500
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}), 502, "manufacturing answered 500")
nb.manufacturing_status = None
assert ring(sid, [("LATTE-REG", 1)])["gross"] == "3.60"
assert must(api("GET", "/receipts", CASHIER, {"session_id": sid}))["receipts"] == [f"{sid}-0001"]
def test_pos_31_recipes_from_manufacturing_need_it_installed_and_are_read_only_here(nb):
a_cafe(nb)
make_shop()
set_source("manufacturing")
sid = open_till()["session_id"]
refused(write(), 409, "read-only here", gate="recipe_source")
seam.unregister_local("manufacturing")
refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}), 409,
"not part of this installation")
assert must(api("GET", "/recipes", CASHIER))["recipe_source"] == "manufacturing"
refused(api("POST", "/settings/recipe_source", OPS, {"value": "oracle", "valid_from": "2026-12-01"}), 422, "pos, manufacturing", gate="bounds")
def test_pos_31_pos_calls_manufacturing_exactly_as_its_signed_spec_describes():
ask = {"usage": "sale", "quantity": "1", "unit_code": "C62", "modifiers": "MOD-OAT", "on": "2026-12-02"}
assert contract.check("manufacturing", "GET", "/boms/LATTE-REG/explode", ask, caller="pos", method_dir=METHOD_DIR) == []
assert contract.check("manufacturing", "GET", "/boms/LATTE-REG/explode", {**ask, "usage": "bogus"}, caller="pos", method_dir=METHOD_DIR) != [] # the check bites
assert contract.check("manufacturing", "GET", "/boms/LATTE-REG/explode", {k: v for k, v in ask.items() if k != "on"}, caller="pos",
method_dir=METHOD_DIR) != []
def test_pos_31_a_component_quantity_the_stock_cannot_hold_is_422_bounds(nb):
make_shop()
nb.bom("LATTE-REG", [("ROAST-HOUSE", "0.0000001", "KGM")], valid_from="2026-01-01")
set_source("manufacturing")
sid = open_till()["session_id"]
status, body = api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]})
assert status == 422 and body["gate"] == "bounds"
def test_a_sale_costs_the_same_statements_for_2_lines_or_20_and_with_50_unrelated_recipes_around(engine, nb):
a_cafe(nb)
many_products(nb)
make_shop()
freeze("2026-10-25T10:00:00+00:00")
for i in range(1, 21):
nb.product_row(f"P{i:02d}", "hot_drink")
must(write({"product_id": f"P{i:02d}", "lines": [{"component_id": "CUP-12OZ", "quantity": "1", "unit_code": "C62"}], "modifiers": []}))
freeze("2026-11-14T09:00:00+00:00")
sid = open_till()["session_id"]
costs = {}
for n in (2, 20):
with counting(engine) as seen:
r = must(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": lines(n)}))
assert len(r["lines"]) == n and len(r["components"]) == n
costs[n] = len(seen)
assert costs[2] == costs[20], costs
with engine.begin() as conn: # 50 recipes of products nobody sells here
repo = RecipeRepository(conn)
for i in range(50):
repo.add(Recipe(product_id=f"UNRELATED-{i:02d}", version=1, valid_from=__import__("datetime").date(2026, 1, 1),
lines=[RecipeLine(1, "CUP-12OZ", Decimal(1), "C62")]), __import__("datetime").datetime(2026, 1, 1))
with counting(engine) as seen:
must(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": lines(2)}))
assert len(seen) == costs[2], (len(seen), costs)