dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_adapters.py
"""The REAL adapters over neighbour apps registered in process (erp_platform.seam): the routes this brick calls, as each
neighbour's spec names them. The rest of the suite uses port fakes; this proves the wiring that sits behind the ports."""
import pytest
from conftest import ADMIN, BILLER, REP, api, hand, must
from erp_platform import seam
from fakes import FakeProduct, FakeTax
from fastapi import FastAPI, Request, Response
from brick_sales_order.infrastructure import seams
from brick_sales_order.infrastructure.seams import Missing
@pytest.fixture
def real(monkeypatch):
prod, tax = FakeProduct(), FakeTax()
billed, drafts = {}, {}
apps = {k: FastAPI() for k in ("product", "tax", "payments", "ar", "business-partner")}
@apps["product"].get("/products/{pid}/price")
def price(pid: str, request: Request, response: Response):
q = request.query_params
from datetime import date
try:
return prod.find_price({}, product_id=pid, quantity=q["quantity"], unit_code=q.get("unit_code"), bp_id=q["bp_id"], currency=q["currency"],
priced_on=date.fromisoformat(q["priced_on"]))
except Missing as m:
response.status_code = m.status
return {"detail": m.reason}
@apps["tax"].post("/documents/compute")
async def compute(request: Request):
"""tax's compute_tax shape: lines [{line_no, category_code, amount}] -> {complete, lines [{line_no, tax_code, rate_pct, vat}], totals}."""
b = await request.json()
from datetime import date
got = tax.compute({}, lines=[{"ref": x["line_no"], "tax_category": x.get("category_code") or "", "net_amount": x["amount"]}
for x in b["lines"]], bp_id=b["bp_id"], tax_point=date.fromisoformat(b["tax_point"]), currency=b["currency"])
return {"complete": True, "lines": [{"line_no": x["ref"], "tax_code": x["tax_code"], "rate_pct": x["rate"], "vat": x["tax_amount"],
"refusal": None} for x in got["lines"]],
"totals": {"vat": got["tax_total"], "method": got["method"]}, "gates": []}
@apps["payments"].get("/customers/{bp}/credit-ok")
def credit(bp: int):
return {"gate": "credit_ok", "met": bp != 7399, "reason": "" if bp != 7399 else "an invoice 75 days overdue", "fix": "", "hold_mode": "block",
"bp_id": bp}
@apps["ar"].post("/invoices/raise")
async def raise_draft(request: Request):
b = await request.json()
drafts.setdefault(b["source_id"], b)
return {"invoice_id": b["source_id"], "lifecycle": "draft"}
@apps["ar"].get("/sources/sales_order/{sid}/billed")
def billed_per_source(sid: str):
return {"source_kind": "sales_order", "source_id": sid,
"lines": [{"so_line_no": int(k), "quantity_billed": v, "net_billed": "0.00"} for k, v in billed.get(sid, {}).items()]}
@apps["ar"].get("/invoice-totals")
def invoice_totals(invoice_ids: str):
return {"invoices": {i: {"lifecycle": "draft"} for i in invoice_ids.split(",") if i in drafts}}
@apps["ar"].post("/invoices/{iid}/cancel")
def cancel(iid: str, response: Response):
if drafts.pop(iid, None) is None:
response.status_code = 404
return {"detail": f"no invoice {iid}"}
return {"invoice_id": iid, "lifecycle": "cancelled"}
@apps["business-partner"].get("/partners/{bp}")
def partner(bp: int, response: Response):
if bp >= 9000:
response.status_code = 404
return {"detail": "no such partner"}
return {"bp_id": bp, "status": "active", "roles": ["customer", "ship_to"], "can_transact": True}
for name, app in apps.items():
seam.register_local(name, app)
for attr, cls in (("product", seams.RealProduct), ("tax", seams.RealTax), ("payments", seams.RealPayments),
("invoicing", seams.RealInvoicing), ("partners", seams.RealPartners)):
monkeypatch.setattr(seams.ports, attr, cls())
yield {"drafts": drafts, "billed": billed}
for name in apps:
seam.unregister_local(name)
def test_a_hand_order_is_priced_taxed_confirmed_and_invoiced_through_the_real_adapters(real):
o = hand(lines=[{"product_id": "FEE", "quantity": "2", "unit_code": "C62"}])
assert o["lines"][0]["priced_from"].startswith("LIST-GBP") and o["total"] in ("2400.00", "2400.0000") and o["lines"][0]["tax_code"] == "S"
sid = o["sales_order_id"]
c = must(api("POST", f"/sales-orders/{sid}/confirm", REP, {}))
assert c["order_no"] == "SO-2026-0001"
must(api("POST", f"/sales-orders/{sid}/request-invoice", BILLER, {}))
draft = real["drafts"][f"so:{sid}:1"]
assert draft["source_kind"] == "sales_order" and draft["lines"][0]["sales_order_id"] == sid and draft["bp_id"] == "7311"
assert draft["lines"][0]["so_line_no"] == 1 and "line_no" not in draft["lines"][0] and "net_amount" not in draft["lines"][0]
real["billed"][sid] = {"1": "2"}
assert must(api("GET", f"/sales-orders/{sid}", ADMIN))["invoice_status"] == "invoiced"
assert must(api("GET", "/sales-orders", ADMIN))["orders"][0]["invoice_status"] == "invoiced"
def test_a_neighbours_refusal_travels_with_its_reason(real):
s, body = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "SETUP", "quantity": "1", "unit_code": "C62"}]})
assert s == 422 and "not counted in C62" in body["detail"]
s, body = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "NOPE", "quantity": "1"}]})
assert s == 422 and "no product NOPE" in body["detail"]
def test_the_credit_hold_and_a_missing_partner_come_from_the_neighbours(real):
held = hand(bp_id=7399)
gates = {g["gate"]: g for g in must(api("GET", f"/sales-orders/{held['sales_order_id']}/readiness", REP))["readiness"]["confirm"]}
assert not gates["credit"]["met"] and "75 days overdue" in gates["credit"]["reason"]
ghost = hand(bp_id=9001)
gates = {g["gate"]: g for g in must(api("GET", f"/sales-orders/{ghost['sales_order_id']}/readiness", REP))["readiness"]["confirm"]}
assert not gates["customer_active"]["met"] and "no business partner 9001" in gates["customer_active"]["reason"]
def test_a_module_that_is_not_installed_is_said_so(monkeypatch):
for attr, cls in (("product", seams.RealProduct), ("invoicing", seams.RealInvoicing), ("tax", seams.RealTax)):
monkeypatch.setattr(seams.ports, attr, cls())
s, body = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "FEE", "quantity": "1"}]})
assert s == 409 and "not part of this installation" in body["detail"]