"""The one neighbour this brick asks: the partner register, to check that the customer a list is assigned to exists
(infrastructure/seams.py). business-partner is a RECOMMENDED dependency (module.yaml), so every answer is a reply, never an
exception: known, unknown (refused), or unchecked (not installed / down — the assignment is kept and says so)."""
import pytest
from conftest import PRICING, api, mk_list, must
from erp_platform import seam
from fastapi import FastAPI, Response
def register(status):
app = FastAPI()
@app.get("/partners/{bp_id}")
def partner(bp_id: int, response: Response):
response.status_code = status
return {"bp_id": bp_id, "can_transact": True} if status == 200 else {"detail": "x"}
seam.register_local("business-partner", app)
@pytest.fixture(autouse=True)
def no_neighbour(partner_register):
yield
def assign(bp_id="7311"):
return api("POST", "/price-lists/LIST-GBP/assign-customer", PRICING, {"bp_id": bp_id})
def test_with_no_partner_register_a_customer_cannot_be_assigned():
"""The signed spec: without the register a customer price list 'cannot be assigned'. Nothing is kept unchecked."""
seam.unregister_local("business-partner")
mk_list("LIST-GBP")
status, body = assign()
assert status == 409 and "not part of this installation" in body["detail"] and "cannot be assigned" in body["detail"]
assert api("GET", "/customer-price-lists/7311/GBP", PRICING)[0] == 404 # nothing was written
def test_a_customer_the_register_knows_is_accepted_and_one_it_does_not_is_refused():
mk_list("LIST-GBP")
register(200)
got = must(assign())
assert got["customer_checked"] is True and got["price_list_id"] == "LIST-GBP"
register(404)
status, body = assign("9")
assert status == 422 and "there is no business partner 9" in body["detail"]
assert api("GET", "/customer-price-lists/9/GBP", PRICING)[0] == 404 # nothing was written
def test_a_register_that_fails_or_refuses_is_reported_as_what_it_did():
mk_list("LIST-GBP")
for answer, status, words in ((500, 502, "failed (500"), (503, 502, "failed (503"), (403, 403, "refused the question (403")):
register(answer)
got_status, body = assign()
assert got_status == status and words in body["detail"] and "was not assigned" in body["detail"], (answer, body)
assert "did not answer" not in body["detail"]
assert api("GET", "/customer-price-lists/7311/GBP", PRICING)[0] == 404 # nothing was written
def test_a_register_that_cannot_be_reached_did_not_answer_503(monkeypatch):
mk_list("LIST-GBP")
monkeypatch.setattr(seam.Seam, "get", lambda self, *a, **kw: seam.Reply(False, reason="bp unreachable: timed out"))
status, body = assign()
assert status == 503 and "did not answer (bp unreachable: timed out)" in body["detail"]
def test_assigning_again_changes_nothing_and_a_later_assignment_is_a_new_row():
mk_list("LIST-GBP")
mk_list("OTHER-GBP")
first = must(assign())
assert first["valid_from"] == "2026-11-21" and must(assign()) == first # tomorrow; again: nothing
status, body = api("POST", "/price-lists/OTHER-GBP/assign-customer", PRICING, {"bp_id": "7311", "valid_from": "2026-11-21"})
assert status == 409 and "never edited" in body["detail"] # the same day, another list
later = must(api("POST", "/price-lists/OTHER-GBP/assign-customer", PRICING, {"bp_id": "7311", "valid_from": "2026-12-01"}))
assert later["price_list_id"] == "OTHER-GBP"
got = must(api("GET", "/customer-price-lists/7311/GBP", PRICING))["assignments"]
assert [(a["valid_from"], a["price_list_id"]) for a in got] == [("2026-11-21", "LIST-GBP"), ("2026-12-01", "OTHER-GBP")]
assert must(api("GET", "/customer-price-lists", PRICING))["customer_price_lists"] == [[7311, "GBP"]]
def test_later_bricks_read_the_odoo_map_through_one_route(engine):
"""shared.yaml conventions.odoo_migration: GET /migration-map/{db}/{model}?ids= -> [{odoo_model, odoo_id, ours}]."""
from conftest import SELLING, api
def must_list(r):
assert r[0] == 200, r
return r[1]
from brick_product.infrastructure.tables import Repository
with engine.begin() as conn:
Repository(conn).migration_map.put("odoo17", "product.product", 201, "WIDGET")
s, got = api("GET", "/migration-map/odoo17/product.product", SELLING, {"ids": "201,202"})
assert s == 200 and got == [{"odoo_model": "product.product", "odoo_id": "201", "ours": "WIDGET"}]
assert must_list(api("GET", "/migration-map/odoo17/uom.uom", SELLING, {"ids": "1"})) == []