dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

scripts/live_check.py

"""ONE live check: the brick alone on a scratch DataK3 bucket (default w2purlab, org dodil) — apply the schema, then one happy-path story
through a ROUTE (supplier price -> draft order -> confirm -> read) and one through MCP (tool list -> source list -> readiness read).
The pg wire is real; the neighbours (business-partner, product) are the in-process fakes of tests/fakes.py.

    BUCKET=w2purlab python scripts/live_check.py

The schema is applied here (not with `erp-compose`, which needs the module signed in the published method). The token is the logged-in
CLI session's, read from `dodil data connect` and never printed.
"""
import base64
import datetime
import json
import os
import pathlib
import re
import subprocess
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
BUCKET = os.environ.get("BUCKET", "w2purlab")
out = subprocess.run(["dodil", "data", "connect", BUCKET], capture_output=True, text=True, check=True).stdout
os.environ["DODIL_TOKEN"] = re.search(r"token:([^@]+)@", out).group(1)
os.environ["BUCKET"] = BUCKET

import anyio  # noqa: E402
import httpx  # noqa: E402
import httpx2  # noqa: E402
from erp_platform import db  # noqa: E402
from fakes import FakeBp, FakeProduct  # noqa: E402
from mcp import ClientSession  # noqa: E402
from mcp.client.streamable_http import streamable_http_client  # noqa: E402

from brick_purchasing.app import app  # noqa: E402
from brick_purchasing.infrastructure import seams  # noqa: E402
from brick_purchasing.infrastructure.tables import Repository  # noqa: E402


def person(email, perms):
    payload = base64.urlsafe_b64encode(json.dumps({"permissions": perms}).encode()).decode().rstrip("=")
    return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": []}), "X-Dodil-User-JWT": f"h.{payload}.s"}


P = "purchasing:"
BUYER = person("buyer@live", [P + x for x in ("supplier_price:write", "order:write", "order:read", "order:confirm", "source:write", "supplier:read")])


OWNER = person("owner@live", [P + "settings:controls"])


def apply_schema():
    sql = (ROOT / "src" / "brick_purchasing" / "schema.sql").read_text()
    stmts = re.findall(r"(CREATE TABLE IF NOT EXISTS.*?\);)", sql, re.S)
    with db.engine.begin() as c:
        for s in stmts:
            c.exec_driver_sql(s)
        Repository(c).seed()
    print(f"schema applied: {len(stmts)} tables on {BUCKET}")


async def route(method, path, headers, body=None):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://brick") as c:
        r = await c.request(method, path, headers=headers, json=body) if method != "GET" else await c.get(path, headers=headers)
        return r.status_code, r.json()


def api(*a):
    return anyio.run(route, *a)


async def mcp(headers, name, args):
    async with app.router.lifespan_context(app):
        async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=headers, base_url="http://brick") as http:
            async with streamable_http_client("http://brick/mcp", http_client=http) as (r, w):
                async with ClientSession(r, w) as s:
                    await s.initialize()
                    names = [t.name for t in (await s.list_tools()).tools]
                    res = await s.call_tool(name, args)
                    return names, res.is_error, json.loads(res.content[0].text) if not res.is_error else res.content[0].text


def main():
    apply_schema()
    seams.ports.bp, seams.ports.product = FakeBp(), FakeProduct()
    today = datetime.date.today().isoformat()
    stamp = datetime.datetime.now().strftime("%H%M%S")
    s, price = api("POST", "/supplier-prices", BUYER, {"bp_id": 7001, "product_id": "COLA-075", "unit_code": "PK24", "unit_price": "18.0000",
                                                      "min_quantity": "1", "valid_from": today})
    print("route price:", s)
    api("POST", "/settings/controls", OWNER, {"setting_key": "order_approval_threshold", "value": "1000000", "valid_from": today})   # first row may start today
    assert s in (200, 409), price                                             # 409: the first run left today's row (dated rows are never back-dated)
    s, o = api("POST", "/purchase-orders", BUYER, {"bp_id": 7001, "currency": "GBP", "supplier_ref": f"LIVE-{stamp}", "lines": [{"product_id": "COLA-075", "quantity": "10"}]})
    assert s == 200, o
    s, c = api("POST", f"/purchase-orders/{o['purchase_order_id']}/confirm", BUYER, {})
    assert s == 200 and c["status"] == "confirmed" and c["order_no"], c
    s, got = api("GET", f"/purchase-orders/{o['purchase_order_id']}", BUYER)
    assert s == 200 and str(got["total"]) == "180.00", got
    print("route: confirmed", c["order_no"], "total", got["total"])
    names, err, listed = anyio.run(mcp, BUYER, "list_purchase_orders", {})
    assert not err and listed["total"] >= 1 and "set_source" in names and "draft_agreement" in names, (err, listed)
    names, err, src = anyio.run(mcp, BUYER, "set_source", {"product_id": f"LIVE-{stamp}", "bp_id": 7001, "valid_from": today, "required": True})
    assert not err and src["required"] is True, src
    print(f"mcp: {len(names)} tools; list_purchase_orders total {listed['total']}; set_source ok")
    print("LIVE OK")


if __name__ == "__main__":
    main()