dodil/erp-brick-product

tests/test_doors.py

"""The use cases through both doors, on SQLite: every use case is a route AND a tool, the same call answers the same
through the API and MCP, a missing row is 404, no permission means nothing is written, and the module's access rules
(module.yaml tests.access) hold: sales never sees cost or a draft, a second person publishes where the settings ask."""
from conftest import (
    ALICE,
    CATALOGUE,
    DEFAULT,
    NOBODY,
    OWNER,
    PRICING,
    SALES,
    SETUP,
    SYSTEM,
    api,
    mk_list,
    mk_product,
    must,
    on_day,
    publish,
    row,
    tool,
    tools,
)

ROUTES = [
    ("list_products", "GET /products"),
    ("get_product", "GET /products/{product_id}"),
    ("add_product", "POST /products"),
    ("ai_check_duplicate", "POST /products/{product_id}/ai-check-duplicate"),
    ("ai_suggest_details", "POST /products/{product_id}/ai-suggest-details"),
    ("complete_product", "POST /products/{product_id}/complete-product"),
    ("activate_product", "POST /products/{product_id}/activate-product"),
    ("rename_product", "POST /products/{product_id}/rename-product"),
    ("archive_product", "POST /products/{product_id}/archive-product"),
    ("reactivate_product", "POST /products/{product_id}/reactivate-product"),
    ("list_price_lists", "GET /price-lists"),
    ("get_price_list", "GET /price-lists/{price_list_id}"),
    ("list_price_rows", "GET /price-lists/{price_list_id}/rows"),
    ("list_price_rules", "GET /price-lists/{price_list_id}/rules"),
    ("open_price_list", "POST /price-lists"),
    ("set_prices", "POST /price-lists/{price_list_id}/set-prices"),
    ("ai_price_check", "POST /price-lists/{price_list_id}/ai-price-check"),
    ("publish_price_list", "POST /price-lists/{price_list_id}/publish-price-list"),
    ("change_prices", "POST /price-lists/{price_list_id}/change-prices"),
    ("assign_customer", "POST /price-lists/{price_list_id}/assign-customer"),
    ("retire_price_list", "POST /price-lists/{price_list_id}/retire-price-list"),
    ("list_customer_price_lists", "GET /customer-price-lists"),
    ("get_customer_price_list", "GET /customer-price-lists/{bp_id}/{currency}"),
    ("find_price", "GET /products/{product_id}/price"),
    ("import_catalogue", "POST /catalogue/import"),
    ("list_product_tasks", "GET /tasks"),
    ("close_product_task", "POST /tasks/{task_id}/close"),
    ("list_product_settings", "GET /settings"),
    ("set_product_setting", "POST /settings/{setting_key}"),
    ("list_product_categories", "GET /product-categories"),
    ("set_product_category", "POST /product-categories/{category_code}"),
    ("list_units", "GET /units"),
    ("set_unit", "POST /units/{unit_code}"),
]


def test_every_use_case_is_a_route_and_a_tool():
    status, spec = api("GET", "/openapi.json", ALICE)
    assert status == 200
    for name, route in ROUTES:
        method, path = route.split(" ", 1)
        assert method.lower() in spec["paths"].get(path, {}), route
    assert {n for n, _ in ROUTES} <= set(tools(ALICE))


def test_a_product_through_both_doors():
    made = must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": "HOST-M", "name": "Managed hosting"}))
    got = api("GET", "/products/HOST-M", CATALOGUE)[1]
    V4 = ("open_tasks", "lifecycle", "on", "roles", "roles_defaulted", "units", "product_type_code", "stocked_default", "valuation_class",
          "commodity_code", "country_of_origin", "classes", "characteristics")                      # what GET adds (v4); the add's answer is the v1 view
    assert {k: v for k, v in got.items() if k not in V4} == {k: v for k, v in made.items()
                                                                   if k not in ("likely_duplicates", "ai", "task_id", "product_type_code")}
    assert tool(CATALOGUE, "get_product", {"product_id": "HOST-M"})[1]["name"] == "Managed hosting"
    assert made["version"] == 1 and made["status"] == "draft"
    via_tool = tool(CATALOGUE, "add_product", {"product_id": "HOST-Q", "product_type_code": "SERVICE", "name": "Managed hosting, quarterly"})
    assert via_tool[0] == "ok" and via_tool[1]["version"] == 1
    assert tool(CATALOGUE, "add_product", {"product_id": "HOST-Q", "product_type_code": "SERVICE", "name": "Again"}) == ("refused", "409: HOST-Q already exists")


def test_a_missing_product_is_not_found():
    assert api("GET", "/products/HOST-M", CATALOGUE)[0] == 404
    assert tool(CATALOGUE, "get_product", {"product_id": "HOST-M"})[0] == "refused"


def test_find_price_answers_the_same_through_both_doors():
    mk_product("HOST-M")
    mk_list("LIST-GBP", rows=[row("HOST-M", "2026-11-01", "120.00")], default_from=DEFAULT)
    args = {"product_id": "HOST-M", "currency": "GBP", "quantity": "3", "bp_id": "7311", "priced_on": "2026-12-01"}
    status, http = api("GET", "/products/HOST-M/price", SALES, {k: v for k, v in args.items() if k != "product_id"})
    assert status == 200 and http["unit_price"] == "120.00"
    assert tool(SALES, "find_price", args) == ("ok", http)
    assert tool(SALES, "find_price", {**args, "currency": "EUR"})[0] == "refused"


def test_a_price_list_through_both_doors():
    made = must(api("POST", "/price-lists", PRICING, {"price_list_id": "LIST-GBP", "currency": "GBP"}))
    assert made["status"] == "draft" and made["version"] == 1 and made["rounding"] == "0.01"
    assert api("GET", "/price-lists/LIST-GBP", PRICING)[1] == made
    assert tool(PRICING, "get_price_list", {"price_list_id": "LIST-GBP"}) == ("ok", made)
    assert must(api("POST", "/price-lists", PRICING, {"price_list_id": "LIST-GBP", "currency": "GBP"})) == made     # the same again
    assert api("POST", "/price-lists", PRICING, {"price_list_id": "LIST-GBP", "currency": "EUR"})[0] == 409
    assert api("POST", "/price-lists", PRICING, {"price_list_id": "BAD", "currency": "pounds"})[0] in (409, 422)
    assert api("GET", "/price-lists/NOPE", PRICING)[0] == 404


def test_without_the_permission_nothing_is_written():
    assert api("POST", "/products", NOBODY, {"product_type_code": "SERVICE", "product_id": "HOST-M", "name": "x"})[0] == 403
    assert api("POST", "/products", SALES, {"product_type_code": "SERVICE", "product_id": "HOST-M", "name": "x"})[0] == 403
    assert api("GET", "/products/HOST-M", CATALOGUE)[0] == 404
    assert api("POST", "/price-lists", SALES, {"price_list_id": "L", "currency": "GBP"})[0] == 403
    assert api("POST", "/price-lists", CATALOGUE, {"price_list_id": "L", "currency": "GBP"})[0] == 403    # keeps products, not prices
    assert api("GET", "/price-lists/L", PRICING)[0] == 404


def test_access_sales_reads_active_products_and_published_prices_and_never_cost():
    mk_product("HOST-M", cost_amount="61.50")
    mk_product("DRAFT-1", activate=False)
    mk_list("LIST-GBP", rows=[row("HOST-M", "2026-11-01", "120.00")], default_from=DEFAULT)
    mk_list("DRAFT-GBP", publish=False)
    assert [p["product_id"] for p in must(api("GET", "/products", SALES))["products"]] == ["HOST-M"]
    assert api("GET", "/products/DRAFT-1", SALES)[0] == 404 and api("GET", "/products/DRAFT-1", PRICING)[0] == 200
    assert api("GET", "/products", SALES, {"status": "draft"})[1]["products"] == []
    assert "cost" not in str(must(api("GET", "/products/HOST-M", SALES))) + str(must(api("GET", "/products", SALES)))
    assert [p["price_list_id"] for p in must(api("GET", "/price-lists", SALES))["price_lists"]] == ["LIST-GBP"]
    assert len(must(api("GET", "/price-lists", PRICING))["price_lists"]) == 2
    assert api("GET", "/price-lists/DRAFT-GBP", SALES)[0] == 404                      # a draft list is a 404, not a 403
    assert api("GET", "/price-lists/DRAFT-GBP/rows", SALES)[0] == 404
    assert [r["unit_price"] for r in must(api("GET", "/price-lists/LIST-GBP/rows", SALES, {"status": "draft"}))["rows"]] == ["120.00"]
    assert api("POST", "/price-lists/LIST-GBP/set-prices", SALES, {})[0] == 403        # a sales rep cannot write a price
    assert api("POST", "/products/HOST-M/archive-product", SALES, {})[0] == 403
    assert "opened_by" not in must(api("GET", "/price-lists/LIST-GBP", SALES))
    assert must(api("GET", "/price-lists/LIST-GBP", PRICING))["opened_by"] == "pricing_manager@shop.example"


def test_access_a_machine_permission_is_held_by_the_system_alone():
    mk_product("HOST-M", activate=False)
    assert api("POST", "/products/HOST-M/ai-check-duplicate", CATALOGUE, {})[0] == 403          # not even the catalogue manager
    assert api("POST", "/products/HOST-M/ai-check-duplicate", person_with("*:*:*"), {})[0] == 403    # no wildcard reaches it
    checked = must(api("POST", "/products/HOST-M/ai-check-duplicate", SYSTEM, {}))
    assert checked["product_id"] == "HOST-M" and checked["likely_duplicates"] == []
    suggested = must(api("POST", "/products/HOST-M/ai-suggest-details", SYSTEM, {}))
    assert suggested["fields_to_suggest"] == ["category_code"] and suggested["ai"]["kind"] == "manual" and suggested["task_id"]
    task, = must(api("GET", "/tasks", CATALOGUE))["tasks"]
    assert task["step"] == "ai_suggest_details" and "category_code" in task["plan"]
    assert api("GET", "/tasks", PRICING)[1]["tasks"] == []                                        # not theirs to see
    assert api("POST", f"/tasks/{task['task_id']}/close", PRICING, {"outcome": "accepted"})[0] == 404
    must(api("POST", "/settings/ai.ai_suggest_details.mode", CATALOGUE, {"value": "off"}))
    assert api("POST", "/products/HOST-M/ai-suggest-details", SYSTEM, {})[0] == 409
    mk_list("LIST-GBP", publish=False)
    assert must(api("POST", "/price-lists/LIST-GBP/ai-price-check", SYSTEM, {}))["flagged"] == []
    assert api("POST", "/price-lists/LIST-GBP/ai-price-check", PRICING, {})[0] == 403


def test_access_the_author_of_a_list_cannot_publish_it_where_the_setting_asks():
    mk_product("HOST-M")
    mk_list("LIST-GBP", rows=[row("HOST-M", "2026-11-01", "120.00")], publish=False)
    must(api("POST", "/pricing-controls/publish_needs_second_person", OWNER, {"value": True}))
    status, body = publish("LIST-GBP")
    assert status == 409 and "second person" in body["detail"]
    with on_day(SETUP):
        assert tool(PRICING, "publish_price_list", {"price_list_id": "LIST-GBP"})[0] == "refused"
    other = person_with("product:price_list:publish", "product:price_list:read", email="boss@shop.example")
    assert must(publish("LIST-GBP", other))["status"] == "published"
    # a scheduled change on a published list needs the same: its author may not put it live
    must(api("POST", "/price-lists/LIST-GBP/change-prices", PRICING, {"rows": [row("HOST-M", "2027-01-01", "126.00")]}))
    assert api("POST", "/price-lists/LIST-GBP/publish-price-list", PRICING, {})[0] == 409
    assert must(api("POST", "/price-lists/LIST-GBP/publish-price-list", other, {}))["contents"]["rows_pending"] == 0


def test_a_stale_save_is_a_conflict_not_a_lost_update():
    mk_product("HOST-M", activate=False)
    assert api("POST", "/products/HOST-M/complete-product", CATALOGUE, {"version": 1, "cost_amount": "1"})[0] == 200
    status, body = api("POST", "/products/HOST-M/complete-product", CATALOGUE, {"version": 1, "cost_amount": "2"})
    assert status == 409 and "changed since you opened it" in body["detail"]


def test_inputs_are_bounded():
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": "A" * 65, "name": "x"})[0] in (409, 422)
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": "A", "name": "x" * 201})[0] == 422
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": "A", "name": "x", "cost_amount": "1e30"})[0] == 422
    mk_list("LIST-GBP", publish=False)
    for bad in ({"unit_price": "1.23456"}, {"unit_price": 1.5}, {"unit_price": "-1"}, {"unit_price": "1e20"}):
        assert api("POST", "/price-lists/LIST-GBP/set-prices", PRICING, {"rows": [{"product_id": "A", "valid_from": "2026-11-01", **bad}]})[0] == 422
    assert api("POST", "/price-lists/LIST-GBP/set-prices", PRICING, {"rows": [{"product_id": "A", "valid_from": "not-a-date", "unit_price": "1"}]})[0] == 422
    assert api("GET", "/products/A/price", SALES, {"currency": "GBP", "bp_id": "9" * 30})[0] == 422


def test_the_process_view_shows_the_workflows_as_bpmn():
    status, body = api("GET", "/process/bpmn", ALICE)
    assert status == 200 and body["processes"] == ["product", "price_list"]


def test_the_settings_are_seeded_one_row_per_setting_typed():
    status, body = api("GET", "/settings", ALICE)
    assert status == 200
    assert {s["setting_key"]: s["kind"] for s in body["settings"]} == {
        "odoo_type.product": "str", "odoo_type.consu": "str", "odoo_type.goods": "str", "odoo_type.service": "str", "odoo_type.combo": "str",
        "ai.ai_check_duplicate.mode": "mode", "ai.ai_price_check.mode": "mode", "ai.ai_suggest_details.mode": "mode",
        "business_timezone": "str", "product_code_format": "str", "publish_needs_second_person": "bool",
        "unusual_price_pct": "int"}
    assert all(s["value"] is None for s in body["settings"])


def test_a_setting_of_the_wrong_kind_or_range_is_refused():
    path = "/settings/product_code_format"
    assert api("POST", path, ALICE, {"value": 7})[0] == 422
    assert api("POST", path, ALICE, {"value": "whatever"})[0] == 422
    assert api("POST", "/pricing-controls/unusual_price_pct", OWNER, {"value": 0})[0] == 422
    assert api("POST", "/pricing-controls/publish_needs_second_person", OWNER, {"value": 7})[0] == 422
    assert api("POST", "/settings/ai.ai_check_duplicate.mode", ALICE, {"value": "auto"})[0] == 422
    assert api("POST", "/settings/nope", ALICE, {"value": 1})[0] == 404


def test_a_code_pattern_makes_the_next_code_and_holds_typed_ones_to_it():
    must(api("POST", "/settings/product_code_format", CATALOGUE, {"value": "PRD-{seq:5}"}))
    assert api("POST", "/settings/product_code_format", PRICING, {"value": "typed"})[0] == 200
    must(api("POST", "/settings/product_code_format", PRICING, {"value": "PRD-{seq:5}"}))
    first = must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "One"}))
    second = must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "Two"}))
    assert (first["product_id"], second["product_id"]) == ("PRD-00001", "PRD-00002")
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "Three", "product_id": "HOST-M"})[0] == 422
    assert must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "Four", "product_id": "PRD-00010"}))["product_id"] == "PRD-00010"
    assert must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "Five"}))["product_id"] == "PRD-00011"
    must(api("POST", "/settings/product_code_format", PRICING, {"value": "typed"}))
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "name": "Six"})[0] == 422


def test_units_and_categories():
    units = {u["unit_code"]: u["decimals"] for u in must(api("GET", "/units", SALES))["units"]}
    assert units["MON"] == 0 and units["HUR"] == 2 and units["KGM"] == 3 and "C62" in units
    assert api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": "X", "name": "x", "base_unit_code": "ZZZ"})[0] == 422
    must(api("POST", "/units/ZZZ", PRICING, {"name": "zed"}))
    assert api("POST", "/units/not-a-unit", PRICING, {})[0] == 422
    must(api("POST", "/product-categories/services", CATALOGUE, {}))
    must(api("POST", "/product-categories/hosting", CATALOGUE, {"parent_code": "services", "name": "Hosting"}))
    assert api("POST", "/product-categories/services", CATALOGUE, {"parent_code": "hosting"})[0] == 409     # no loops
    assert api("POST", "/product-categories/x", CATALOGUE, {"parent_code": "nope"})[0] == 422
    mk_product("HOST-M", category_code="hosting")
    assert [p["product_id"] for p in must(api("GET", "/products", SALES, {"category_code": "services"}))["products"]] == ["HOST-M"]


def test_product_search_pages_by_code():
    for i in range(5):
        mk_product(f"S{i}", name=f"Searchable item {i}")
    page = must(api("GET", "/products", SALES, {"limit": 2}))
    assert [p["product_id"] for p in page["products"]] == ["S0", "S1"] and page["next"] == "S1"
    nxt = must(api("GET", "/products", SALES, {"limit": 2, "after": "S1"}))
    assert [p["product_id"] for p in nxt["products"]] == ["S2", "S3"]
    assert [p["product_id"] for p in must(api("GET", "/products", SALES, {"q": "item 4"}))["products"]] == ["S4"]


def person_with(*permissions, email="someone@shop.example"):
    from conftest import person
    return person(email, permissions)