#!/usr/bin/env python3
"""CI gate: a customer can build from the PUBLISHED artifacts.
From an empty folder: `erp-compose init --starter uk-wholesale --starter uk-wholesale-sample` (the method is THIS checkout, via ERP_METHOD_DIR;
the bricks and erp-platform are cloned from git.dodil.io at the tags the method pins), `migrate --local`,
`seed --local`, then one sale through the local server's MCP endpoint (order -> deliver -> invoice -> post to the
ledger -> payment -> allocation) and an assertion that the trial balance nets to 0 with the sale in it.
python3 scripts/ci_fresh_build.py [--keep DIR]
stdlib only here; the sale itself runs under the build folder's own .venv python (it has the `mcp` client the
platform installs): this file re-invokes itself with `--sale URL` for that part.
"""
import argparse
import asyncio
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.request
ROOT = pathlib.Path(__file__).resolve().parent.parent
PORT = 8000
URL = f"http://127.0.0.1:{PORT}"
def sh(cmd, cwd, env=None, check=True):
print("+", " ".join(map(str, cmd)), flush=True)
r = subprocess.run([str(c) for c in cmd], cwd=cwd, env=env)
if check and r.returncode:
sys.exit(f"FAIL: {' '.join(map(str, cmd[:3]))} exited {r.returncode}")
def platform_tag():
"""The erp-platform tag the method pins (role: platform in a module's reference.repos)."""
text = (ROOT / "modules/sales-order/module.yaml").read_text()
m = re.search(r"- name: erp-platform\n\s+role: platform\n\s+https: (\S+)\n(?:\s+ssh: \S+\n)?\s+tag: (\S+)", text)
if not m:
sys.exit("FAIL: modules/sales-order/module.yaml pins no erp-platform tag")
return m.group(1), m.group(2)
def build(work: pathlib.Path):
https, tag = platform_tag()
print(f"method {(ROOT / 'VERSION').read_text().strip()} (this checkout); erp-platform {tag}", flush=True)
tool = work / "tool"
sh(["uv", "venv", "--python", "3.12", "-q", tool], work)
sh(["uv", "pip", "install", "-q", "--python", tool / "bin/python", f"erp-platform @ git+{https}@{tag}"], work)
app = work / "app"
app.mkdir()
env = {**os.environ, "ERP_METHOD_DIR": str(ROOT)}
ec = tool / "bin/erp-compose"
sh([ec, "init", "--starter", "uk-wholesale", "--starter", "uk-wholesale-sample"], app, env)
ec = app / ".venv/bin/erp-compose"
sh([ec, "migrate", "--local", "compose.yaml"], app, env)
for starter in ("uk-wholesale", "uk-wholesale-sample"): # the sample (one customer, one product) after the base
sh([ec, "seed", "--local", "compose.yaml", f"config/{starter}.calls.yaml"], app, env)
log = open(work / "serve.log", "w")
srv = subprocess.Popen([str(ec), "serve", "--local", "compose.yaml"], cwd=app, env=env, stdout=log, stderr=subprocess.STDOUT)
try:
for _ in range(60):
if srv.poll() is not None:
break
try:
urllib.request.urlopen(URL + "/openapi.json", timeout=2)
break
except Exception:
time.sleep(1)
else:
sys.exit("FAIL: the local server never answered")
if srv.poll() is not None:
sys.exit("FAIL: the local server exited:\n" + (work / "serve.log").read_text()[-3000:])
sh([app / ".venv/bin/python", "-I", ROOT / "scripts/ci_fresh_build.py", "--sale", URL + "/mcp"], app, env)
finally:
srv.terminate()
try:
srv.wait(10)
except Exception:
srv.kill()
print("--- server log tail ---\n" + (work / "serve.log").read_text()[-1500:])
async def sale(url):
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
async with streamable_http_client(url) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
async def c(name, args=None):
res = await s.call_tool(name, args or {})
txt = "".join(x.text for x in res.content if hasattr(x, "text"))
try:
out = json.loads(txt)
except Exception:
out = txt
print(">>", name, json.dumps(out)[:220], flush=True)
if res.is_error:
sys.exit(f"FAIL: {name} returned an error")
return out
await sale_steps(c)
async def sale_steps(c):
from decimal import Decimal as D
def need(cond, msg):
if not cond:
sys.exit("FAIL: " + msg)
# the starter's one customer and one product (seeded by config/uk-wholesale-sample.calls.yaml), 4 units at 25.00 + 20% VAT
bp = (await c("find_partners"))["partners"][0]["bp_id"]
product = (await c("list_products"))["products"][0]["product_id"]
so = (await c("draft_by_hand", {"bp_id": bp, "currency": "GBP",
"lines": [{"product_id": product, "quantity": "4", "unit_code": "C62"}]}))["sales_order_id"]
await c("confirm_order", {"sales_order_id": so})
dlv = (await c("create_delivery", {"sales_order_id": so}))["document_id"]
await c("validate_delivery", {"document_id": dlv, "picks": [
{"line_no": 1, "location_id": "MAIN/A/01", "quantity": "4", "unit_code": "C62"}]})
await c("request_invoice", {"sales_order_id": so})
iid = f"so:{so}:1"
inv = await c("issue_invoice", {"invoice_id": iid})
need(inv["lifecycle"] == "issued" and D(inv["total"]) == D("120.00"), f"invoice total {inv.get('total')} != 120.00")
# no background worker runs locally: post the issued invoice to the ledger by hand
posted = await c("post_pending")
need(any(r["document_id"] == iid and r["posted"] for r in posted["results"]), "the invoice was not posted to the ledger")
rc = await c("record_receipt", {"bp_id": bp, "amount": inv["total"], "currency": "GBP", "received_on": inv["issue_date"],
"method": "bank_transfer", "bank_account_id": "BANK1", "bank_reference": "BACS CI",
"remittance_text": inv["invoice_no"]})
al = await c("allocate", {"payment_id": rc["payment_id"], "lines": [{"invoice_id": iid, "amount": inv["total"]}]})
need(all(D(i["outstanding"]) == 0 for i in al["invoices"]), "the invoice is not settled after the allocation")
await c("post_pending")
tb = await c("get_trial_balance", {"company_code": "UK1"})
need(D(tb["difference"]) == 0 and D(tb["total_debit"]) == D(tb["total_credit"]),
f"trial balance does not net to 0: {tb['total_debit']} vs {tb['total_credit']}")
by = {r["code"]: r for r in tb["rows"]}
need(D(by["1100"]["closing"]) == 0, "trade debtors are not cleared by the payment")
need(D(by["1200"]["closing"]) == D("120.00"), "the bank account does not hold the 120.00 paid")
need(D(by["4000"]["credit"]) == D("100.00"), "revenue is not 100.00")
print(f"sale ok: order {so}, invoice {inv['invoice_no']} 120.00 paid; trial balance {tb['total_debit']} = {tb['total_credit']}")
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--keep")
ap.add_argument("--sale")
a = ap.parse_args()
if a.sale:
asyncio.run(sale(a.sale))
else:
work = pathlib.Path(a.keep or tempfile.mkdtemp(prefix="fresh-build-")).resolve()
work.mkdir(exist_ok=True)
try:
build(work)
finally:
if not a.keep:
shutil.rmtree(work, ignore_errors=True)
print("OK")