dodil/erp-brick-pos

DODIL ERP brick: point of sale

tests/test_stories.py

"""The acceptance tests of slice A, one per story (module.yaml stories POS-03, POS-04, POS-05, POS-06, POS-08, the tender and tip
of POS-09, and the controls of POS-35), each named for it and written from its Given / When / Then through the doors."""
import pytest
from conftest import (
    CASHIER,
    CASHIER2,
    LEAD,
    LEAD2,
    OPS,
    OWNER,
    ROLES,
    api,
    finalise,
    freeze,
    make_shop,
    must,
    open_till,
    pay,
    person,
    refused,
    ring,
    sell,
)
from fakes import seed_de


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):
    body = {"kind": kind, "amount": amount, **({"reason_code": reason} if reason else {})}
    return api("POST", f"/sessions/{s['session_id']}/cash-moves", who or CASHIER, body)


def get_session(s, who=None):
    return must(api("GET", f"/sessions/{s['session_id']}", who or CASHIER))


# ── POS-03: open the till with a counted float ──────────────────────────────────────────────────────────

def test_pos_03_the_session_opens_for_the_business_date_with_the_float_counted(nb):
    make_shop()
    s = open_till(opening_float="150.00")
    assert (s["session_id"], s["status"], s["business_date"], s["opening_float"], s["currency"]) == \
        ("LON-SOHO-R1-0001", "open", "2026-11-14", "150.00", "GBP")
    assert s["opened_by"] == "cashier@shop.example" and s["float_difference"] is None


def test_pos_03_one_register_has_at_most_one_open_session(nb):
    make_shop()
    open_till()
    body = refused(api("POST", "/sessions", CASHIER2, {"shop_id": "LON-SOHO", "register_id": "R1", "opening_float": "100.00"}),
                   409, "already has session LON-SOHO-R1-0001 open", gate="one_open_session")
    assert body["fix"] == "close it first"
    assert open_till(register="R2", who=CASHIER2)["session_id"] == "LON-SOHO-R2-0001"      # another register is another drawer


def test_pos_03_a_session_opened_at_0130_belongs_to_the_previous_business_date(nb):
    make_shop()
    freeze("2026-11-15T01:30:00+00:00")                 # London is on GMT in November: 01:30 local, before the 04:00 cutoff
    assert open_till()["business_date"] == "2026-11-14"


def test_pos_03_a_float_that_differs_from_what_the_last_session_left_is_recorded(nb):
    make_shop()
    first = open_till(opening_float="150.00")
    closed = must(count(first, "150.00", float_left="150.00"))
    assert closed["status"] == "closed" and closed["closing_float"] == "150.00"
    second = open_till(opening_float="160.00")
    assert second["session_id"] == "LON-SOHO-R1-0002" and second["float_difference"] == "10.00"
    assert open_till(register="R2")["float_difference"] is None                      # a register with no history has none


def test_pos_03_a_register_whose_fiscal_device_is_not_ready_cannot_open(nb):
    seed_de(nb)
    make_shop("BER-MITTE", registers=("R1",), price_list="LIST-DE-SHOP", country="DE", fiscal="de_kassensichv")
    refused(api("POST", "/sessions", CASHIER, {"shop_id": "BER-MITTE", "register_id": "R1", "opening_float": "100.00"}), 409,
            "TSE or signature device is not registered", gate="fiscal_device_ready")
    must(api("POST", "/shops/BER-MITTE/registers", OPS, {"register_id": "R1", "fiscal_device_serial": "7f3a", "fiscal_device_status": "ready"}))
    assert open_till("BER-MITTE", "R1", "100.00")["currency"] == "EUR"


# ── POS-04: cash in and out of the drawer ───────────────────────────────────────────────────────────────

def test_pos_04_a_safe_drop_and_a_paid_out_are_rows_of_the_session(nb):
    make_shop()
    s = open_till()
    drop = must(move(s, "safe_drop", "200.00"))["cash_move"]
    out = must(move(s, "paid_out", "12.00", "MILK"))["cash_move"]
    assert (drop["cash_move_id"], drop["kind"], drop["amount"]) == ("LON-SOHO-R1-0001-cm1", "safe_drop", "200.00")
    assert (out["cash_move_id"], out["kind"], out["amount"], out["reason_code"]) == ("LON-SOHO-R1-0001-cm2", "paid_out", "12.00", "MILK")
    refused(move(s, "paid_out", "5.00"), 409, "names its reason", gate="reason_required")


def test_pos_04_a_paid_out_above_the_limit_needs_a_shift_lead_who_did_not_open_the_drawer(nb):
    make_shop()
    s = open_till()                                                                   # the cashier opened it
    refused(move(s, "paid_out", "60.00", "SUPPLIES"), 409, "needs a shift lead", gate="paid_out_limit")
    row = must(move(s, "paid_out", "60.00", "SUPPLIES", who=LEAD))["cash_move"]
    assert row["approved_by"] == "shift_lead@shop.example" and row["created_by"] == "shift_lead@shop.example"
    own = open_till(register="R2", who=LEAD2)                                         # a lead opening a drawer cannot approve its paid-outs
    refused(move(own, "paid_out", "60.00", "SUPPLIES", who=LEAD2), 409, "another person", gate="second_person")


# ── POS-05: count the drawer blind ──────────────────────────────────────────────────────────────────────

def a_session_expecting_132(nb):
    """opening 150.00; cash sales 94.00 (3 receipts); a safe drop of 100.00 and a paid-out of 12.00: the drawer should hold 132.00."""
    make_shop()
    s = open_till()
    for item in ([("LATTE-REG", 10)], [("CROISSANT", 10)], [("FLAT-WHITE", 10)]):
        sell(s["session_id"], item)
    must(move(s, "safe_drop", "100.00"))
    must(move(s, "paid_out", "12.00", "MILK"))
    return s


def test_pos_05_the_cashier_does_not_see_what_the_drawer_should_hold_until_the_count_is_saved(nb):
    s = a_session_expecting_132(nb)
    before = get_session(s)
    assert before["expected_cash"] is None and before["summary"]["cash"] is None and before["summary"]["expected_cash"] is None
    assert before["summary"]["receipts"] == 3 and before["summary"]["gross"] == "94.00"      # the sales are not secret, the cash is
    after = must(count(s, "125.50"))
    assert after["expected_cash"] == "132.00" and after["difference"] == "-6.50" and after["counted_cash"] == "125.50"


def test_pos_05_a_difference_above_the_threshold_waits_for_a_second_person(nb):
    s = a_session_expecting_132(nb)
    counted = must(count(s, "125.50"))
    assert counted["status"] == "counted" and counted["approved_by"] == ""


@pytest.mark.parametrize("counted,status", [("127.00", "closed"), ("137.00", "closed"), ("126.99", "counted"), ("137.01", "counted")])
def test_pos_05_within_the_threshold_it_closes_at_once_and_the_difference_is_still_recorded(nb, counted, status):
    s = a_session_expecting_132(nb)                                                  # threshold 5.00: at or below closes
    got = must(count(s, counted))
    assert got["status"] == status and got["difference"] == str(__import__("decimal").Decimal(counted) - 132)


def test_pos_05_a_session_cannot_be_counted_while_the_register_holds_receipts_the_server_has_not(nb):
    s = a_session_expecting_132(nb)
    body = refused(count(s, "132.00", register_last_receipt_no=5), 409, "0004 to 0005", gate="all_synced")
    assert body["fix"].startswith("bring the register online")
    assert must(count(s, "132.00", register_last_receipt_no=3))["status"] == "closed"


def test_pos_05_a_counted_session_never_goes_back_to_open(nb):
    s = a_session_expecting_132(nb)
    counted = must(count(s, "125.50"))
    refused(move(s, "safe_drop", "10.00"), 409, "open session", gate="session_open")
    refused(count(s, "130.00"), 409, "never goes back to open", gate="session_open")
    refused(api("POST", "/receipts", CASHIER, {"session_id": s["session_id"], "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}),
            409, "sales go in an open session", gate="session_open")
    again = must(count(s, "125.50"))                                                 # the same count again is a replay: nothing changes
    assert again["status"] == "counted" and again["version"] == counted["version"]


# ── POS-06: a second person approves the difference ─────────────────────────────────────────────────────

def test_pos_06_the_counter_cannot_approve_under_a_second_login_but_another_lead_can(nb):
    same_sub = person("a2@shop.example", ROLES["shift_lead"], sub="cashier@shop.example")           # other email, same sub
    same_email = person("cashier@shop.example", ROLES["shift_lead"], sub="other-sub")               # same email, other sub
    make_shop(staff=(CASHIER, LEAD, same_sub, same_email))
    s = open_till()
    for item in ([("LATTE-REG", 10)],):
        sell(s["session_id"], item)
    counted = must(count(s, "180.00"))                                                # expected 186.00: difference -6.00
    assert counted["status"] == "counted" and counted["difference"] == "-6.00"
    for alias in (same_sub, same_email):
        body = refused(api("POST", f"/sessions/{s['session_id']}/approve-difference", alias, {"reason": "coins short"}), 409,
                       "you counted this session", gate="second_person")
        assert body["fix"] == "ask another shift lead"
    done = must(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD, {"reason": "coins short in the 2.00 tube"}))
    assert (done["status"], done["approved_by"], done["approved_sub"], done["reason"]) == \
        ("closed", "shift_lead@shop.example", "shift_lead@shop.example", "coins short in the 2.00 tube")
    assert done["approved_at"] is not None and done["difference"] == "-6.00"


def test_pos_06_approving_takes_a_reason_a_permission_and_a_counted_session(nb):
    make_shop()
    s = open_till()
    sell(s["session_id"], [("LATTE-REG", 10)])
    refused(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD, {"reason": "x"}), 409, "only a counted session",
            gate="session_counted")
    must(count(s, "180.00"))
    refused(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD, {"reason": "  "}), 409, "the reason", gate="reason_required")
    refused(api("POST", f"/sessions/{s['session_id']}/approve-difference", CASHIER2, {"reason": "fine"}), 403,
            "pos:session:approve_difference")
    must(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD, {"reason": "ok"}))
    again = must(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD, {"reason": "ok"}))   # a repeat changes nothing
    assert again["status"] == "closed"
    refused(api("POST", f"/sessions/{s['session_id']}/approve-difference", LEAD2, {"reason": "second thoughts"}), 409, "only a counted",
            gate="session_counted")


# ── POS-08: ring up drinks and food and take the payment ────────────────────────────────────────────────

def test_pos_08_a_till_sale_prices_from_product_and_takes_the_vat_from_tax(nb):
    make_shop()
    s = open_till()
    r = ring(s["session_id"], [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)], eat_in=False)
    assert r["receipt_id"] == "LON-SOHO-R1-0001-0001"
    assert [(ln["unit_price"], ln["line_gross"], ln["tax_code"], ln["rate_pct"], ln["vat"]) for ln in r["lines"]] == \
        [("3.60", "3.60", "S", "20.00", "0.60"), ("4.00", "4.00", "S", "20.00", "0.67"), ("2.40", "2.40", "S", "20.00", "0.40")]
    assert (r["gross"], r["net"], r["vat"]) == ("10.00", "8.33", "1.67")              # 10.00 x 20/120 = 1.67, rounded once
    assert nb.rate_calls[-1]["consumption"] is None and nb.price_calls[-1]["on"] == "2026-11-14"        # read once, when the session opened
    done = finalise(pay(r, "cash", "10.00"))
    assert done["status"] == "finalised" and done["payments"][0]["change_amount"] == "0.00" and done["remaining"] == "0.00"


def test_pos_08_the_till_never_types_a_price_and_a_product_off_the_list_is_refused_by_name(nb):
    make_shop()
    s = open_till()
    sid = s["session_id"]
    status, body = api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "LATTE-REG", "quantity": "1", "unit_price": "0.01"}]})
    assert status == 422                                                              # there is no field for a price
    nb.product_row("SECRET-MENU", "hot_drink")
    refused(api("POST", "/receipts", CASHIER, {"session_id": sid, "lines": [{"product_id": "SECRET-MENU", "quantity": "1"}]}), 409,
            "SECRET-MENU is not on the shop's price list", gate="price_known")
    nb.price_lists["LIST-UK-SHOP"]["incl"] = False                                       # the register opens on a list without VAT
    again = open_till(register="R2", who=CASHIER2)
    refused(api("POST", "/receipts", CASHIER2, {"session_id": again["session_id"], "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}), 409,
            "without VAT", gate="prices_include_tax")


def test_pos_08_eat_in_or_takeaway_decides_the_rate_where_the_rules_distinguish_it(nb):
    seed_de(nb)
    make_shop("BER-TEST", registers=("R1",), price_list="LIST-DE-SHOP", country="DE")
    s = open_till("BER-TEST", "R1", "100.00")
    body = {"session_id": s["session_id"], "lines": [{"product_id": "PRETZEL", "quantity": "1"}]}
    refused(api("POST", "/receipts", CASHIER, body), 409, "eating in or taking away", gate="eat_in_known")
    eat = must(api("POST", "/receipts", CASHIER, {**body, "eat_in": True}))
    away = must(api("POST", "/receipts", CASHIER, {**body, "eat_in": False}))
    assert (eat["lines"][0]["tax_code"], eat["lines"][0]["rate_pct"], eat["vat"]) == ("S19", "19.00", "0.40")      # 2.50 x 19/119
    assert (away["lines"][0]["tax_code"], away["lines"][0]["rate_pct"], away["vat"]) == ("R7", "7.00", "0.16")     # 2.50 x 7/107
    drink = must(api("POST", "/receipts", CASHIER, {"session_id": s["session_id"], "lines": [{"product_id": "LATTE-REG", "quantity": "1"}]}))
    assert drink["lines"][0]["tax_code"] == "S19"                                     # a drink has one rate: nothing to ask


def test_pos_08_a_category_tax_does_not_know_stops_the_sale(nb):
    make_shop()
    nb.product_row("MYSTERY", "no_such_category")
    nb.price_lists["LIST-UK-SHOP"]["prices"].append(("MYSTERY", "C62", "1", "1.00"))
    s = open_till()
    refused(api("POST", "/receipts", CASHIER, {"session_id": s["session_id"], "lines": [{"product_id": "MYSTERY", "quantity": "1"}]}), 409,
            "cannot print VAT it did not get from tax", gate="tax_rate_known")


def test_pos_08_cash_over_the_bill_gives_change_and_the_drawer_keeps_only_the_bill(nb):
    make_shop()
    s = open_till()
    r = ring(s["session_id"], [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)])
    line = pay(r, "cash", "20.00")["payments"][0]
    assert (line["amount"], line["change_amount"]) == ("10.00", "10.00")              # 20.00 handed over, 10.00 applied, 10.00 back
    done = must(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}))
    assert done["status"] == "finalised"
    must(count(s, "160.00"))                                                          # 150.00 float + 10.00 applied
    assert get_session(s)["expected_cash"] == "160.00"


def test_pos_08_a_receipt_finalises_only_when_the_payments_equal_the_bill(nb):
    make_shop()
    s = open_till()
    r = ring(s["session_id"], [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)])
    body = refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}), 409, "payments come to 0.00, the bill is 10.00",
                   gate="paid_in_full")
    assert body["fix"] == "take 10.00 more"
    part = pay(r, "cash", "4.00")
    assert part["remaining"] == "6.00" and part["readiness"][0]["met"] is False
    refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}), 409, "come to 4.00", gate="paid_in_full")
    assert finalise(pay(part, "cash", "6.00"))["status"] == "finalised"


def test_pos_08_a_shop_that_must_sign_receipts_cannot_finalise_until_the_capability_exists(nb):
    seed_de(nb)
    make_shop("BER-MITTE", registers=("R1",), price_list="LIST-DE-SHOP", country="DE", fiscal="de_kassensichv")
    must(api("POST", "/shops/BER-MITTE/registers", OPS, {"register_id": "R1", "fiscal_device_serial": "7f3a", "fiscal_device_status": "ready"}))
    s = open_till("BER-MITTE", "R1", "100.00")
    r = pay(ring(s["session_id"], [("LATTE-REG", 1)]), "cash", "3.90")
    refused(api("POST", f"/receipts/{r['receipt_id']}/finalise", CASHIER, {}), 409, "de_kassensichv", gate="fiscal_signature")


# ── POS-09 (the tender and the tip): card, declined card, tips ───────────────────────────────────────────

def test_pos_09_a_card_line_is_the_terminals_and_a_declined_card_leaves_the_receipt_open(nb):
    make_shop()
    s = open_till()
    r = ring(s["session_id"], [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)])
    refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "card", "amount": "10.00"}), 409, "terminal's transaction",
            gate="terminal_ref")
    refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER,
                {"method": "card", "amount": "11.00", "terminal_txn_ref": "T1", "terminal_status": "authorised"}), 409, "10.00 is left",
            gate="within_remainder")
    refused(api("POST", f"/receipts/{r['receipt_id']}/payments", CASHIER, {"method": "voucher", "amount": "1.00"}), 409, "names the voucher", gate="voucher_code")
    declined = pay(r, "card", "10.00", terminal_txn_ref="T1", terminal_status="declined")
    assert declined["paid"] == "0.00" and declined["status"] == "open"
    ok = pay(declined, "card", "10.00", terminal_txn_ref="T2", terminal_status="authorised")
    assert [(p["pay_no"], p["terminal_status"], p["idempotency_key"]) for p in ok["payments"]] == \
        [(1, "declined", f"{r['receipt_id']}:1"), (2, "authorised", f"{r['receipt_id']}:2")]       # the terminal's key is the LINE
    assert finalise(ok)["status"] == "finalised"


def test_pos_09_a_tip_is_a_liability_to_staff_never_revenue(nb):
    make_shop()
    s = open_till()
    sid = s["session_id"]
    r = ring(sid, [("LATTE-REG", 1), ("LATTE-OAT", 1), ("CROISSANT", 1)])
    r = finalise(pay(r, "card", "10.00", tip="1.50", terminal_txn_ref="T1", terminal_status="authorised"))
    assert (r["gross"], r["net"], r["vat"], r["tips"]) == ("10.00", "8.33", "1.67", "1.50")      # the bill and its VAT do not move
    c = sell(sid, [("LATTE-REG", 1)], tip="0.40")                                       # a cash tip
    assert c["tips"] == "0.40"
    summary = get_session(s)["summary"]
    assert (summary["tips"], summary["tips_treatment"], summary["tips_payable"]) == ("1.90", "liability", "1.90")
    assert (summary["gross"], summary["net"], summary["vat"], summary["card"]) == ("13.60", "11.33", "2.27", "10.00")
    must(api("POST", "/settings/tips_treatment", OPS, {"value": "revenue"}))                # the customer's choice, never the default
    again = get_session(s)["summary"]
    assert (again["tips_treatment"], again["tips_payable"]) == ("revenue", "0.00")
    must(count(s, "153.60"))                                                              # 150.00 + 3.60 cash sale + 0.40 cash tip
    assert get_session(s)["expected_cash"] == "154.00"


# ── POS-35: the controls belong to the owner ────────────────────────────────────────────────────────────

def test_pos_35_a_control_is_the_owners_and_changes_by_a_dated_row(nb):
    make_shop()
    refused(api("POST", "/controls/difference_threshold", LEAD, {"value": "50.00"}), 403, "pos:settings:controls")
    refused(api("POST", "/controls/difference_threshold", OPS, {"value": "50.00"}), 403, "pos:settings:controls")
    refused(api("POST", "/settings/difference_threshold", OPS, {"value": "50.00"}), 403, "pos:settings:controls")   # settings:write is not controls
    refused(api("POST", "/settings/blind_close", OWNER, {"value": False}), 403, "missing permission: pos:settings:write")
    first = must(api("POST", "/controls/difference_threshold", OWNER, {"value": "2.00"}))       # the first row of a key may start today
    assert first["valid_from"] == "2026-11-14"
    refused(api("POST", "/controls/difference_threshold", OWNER, {"value": "8.00"}), 422, "starts after today", gate="valid_from")
    later = must(api("POST", "/controls/difference_threshold", OWNER, {"value": "8.00", "valid_from": "2026-11-30"}))
    assert later["value"] == "8.00"
    listed = must(api("GET", "/settings", OWNER))
    assert {s["setting_key"]: s["value"] for s in listed["settings"]}["difference_threshold"] == "2.00"
    s = open_till()
    sell(s["session_id"], [("LATTE-REG", 1)])
    assert must(count(s, "153.00"))["status"] == "closed"                                   # expected 153.60: -0.60 is within 2.00
    freeze("2026-11-30T10:00:00+00:00")                                                  # from the 30th the threshold is 8.00
    t = open_till(register="R2", who=CASHIER2)
    sell(t["session_id"], [("LATTE-REG", 1)], who=CASHIER2)
    assert must(count(t, "145.00", who=CASHIER2))["status"] == "counted"                    # expected 153.60: -8.60 is above 8.00: waits
    assert must(api("POST", "/settings/day_cutoff_time", OPS, {"value": "05:00"}))["value"] == "05:00"       # not a control: retail operations