#!/usr/bin/env python
"""Live check of erp_platform.repository against a real DataK3 bucket.
The unit tests (tests/test_repository.py) check the SQL the repository DECIDES to send; this
checks that DataK3 executes it the way the repository assumes. Each statement is compiled and run
through `dodil data pg` — the same Postgres-wire adapter apps use — so it needs only a CLI login,
no token in the environment. It creates four scratch tables and drops them at the end.
dodil auth use-org dodil
BUCKET=erp-platform-test python scripts/live_repository_check.py
Run it after changing repository.py, and after any DataK3 release: two of its checks exist
because the engine differs from Postgres (ON CONFLICT ignores the SET list — see upsert()).
"""
import json, subprocess, sys
from datetime import datetime, timezone, date
from decimal import Decimal
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parents[1] / "tests"))
from sqlalchemy.dialects import postgresql
from sqlalchemy.sql import Select
from test_repository import QUOTE, contacts # same tables the unit tests use
from erp_platform.repository import upsert
import os, shutil
D, B = shutil.which("dodil") or "dodil", os.environ["BUCKET"]
def pg(sql):
r = subprocess.run([D, "data", "pg", "-b", B, sql, "-o", "json"], capture_output=True, text=True)
if r.returncode:
raise RuntimeError(f"{sql[:160]}\n -> {(r.stderr or r.stdout).strip()[-400:]}")
out = r.stdout.strip()
try: return json.loads(out)
except ValueError: return out
def lit(stmt):
return str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
def typed(col, v):
"""What psycopg would hand back: the column's Python type, not the CLI's JSON rendering."""
if v is None: return None
py = col.type.python_type
if py is date: return date.fromisoformat(str(v)[:10])
if py is datetime: return datetime.fromisoformat(str(v).replace("Z", "+00:00"))
if py is Decimal: return Decimal(str(v))
return py(v)
class Live:
"""A connection that sends each statement to DataK3; SELECT rows come back in SELECT order."""
def execute(self, stmt):
res = pg(lit(stmt))
if isinstance(stmt, Select):
cols = list(stmt.selected_columns)
rs = res.get("rows", []) if isinstance(res, dict) else []
return [tuple(typed(c, r[c.name]) for c in cols) for r in rs]
return []
def rows(sql):
res = pg(sql)
return res.get("rows", res) if isinstance(res, dict) else res
DDL = [
"CREATE TABLE quote_quotes (quote_id VARCHAR PRIMARY KEY, owner VARCHAR, created_at TIMESTAMP, updated_at TIMESTAMP)",
"CREATE TABLE quote_lines (quote_id VARCHAR, version INTEGER, line_no INTEGER, amount DECIMAL(18,2), created_at TIMESTAMP, updated_at TIMESTAMP, PRIMARY KEY (quote_id, version, line_no))",
"CREATE TABLE quote_schedule (quote_id VARCHAR, version INTEGER, line_no INTEGER, month DATE, amount DECIMAL(18,2), PRIMARY KEY (quote_id, version, line_no, month))",
"CREATE TABLE crm_contacts (id INTEGER PRIMARY KEY, name VARCHAR, phone VARCHAR, created_at TIMESTAMP, created_by VARCHAR)",
]
for t in ("quote_schedule", "quote_lines", "quote_quotes", "crm_contacts"):
try: pg(f"DROP TABLE IF EXISTS {t}")
except RuntimeError as e: print("drop:", e)
for s in DDL: pg(s)
print("schema created")
def mem(lines, months, Q="Q-1"):
return {"quote_quotes": [{"quote_id": Q, "owner": "alice"}],
"quote_lines": [{"quote_id": Q, "version": 1, "line_no": n, "amount": Decimal("100.50")} for n in lines],
"quote_schedule": [{"quote_id": Q, "version": 1, "line_no": n, "month": date(2026, m, 1), "amount": Decimal("50.25")} for n, m in months]}
conn, ok = Live(), []
def check(name, cond, got=None):
ok.append(cond); print(("PASS " if cond else "FAIL ") + name + ("" if cond else f" got: {got}"))
# 1. first save: 3 lines, schedule on all three, line 1 has two months; plus a neighbour quote
t1 = datetime(2026, 10, 16, 9, 0, tzinfo=timezone.utc)
QUOTE.save(conn, mem([1, 2, 3], [(1, 1), (1, 2), (2, 1), (3, 1)]), now=t1)
QUOTE.save(conn, mem([1], [(1, 1)], Q="Q-2"), now=t1)
check("first save stored 4 lines' schedule", len(rows("SELECT * FROM quote_schedule WHERE quote_id='Q-1'")) == 4)
# 2. second save: drop line 3 entirely, trim month 2 from line 1 (both are known failure modes)
t2 = datetime(2026, 10, 16, 10, 0, tzinfo=timezone.utc)
QUOTE.save(conn, mem([1, 2], [(1, 1), (2, 1)]), now=t2)
sch = rows("SELECT line_no, month FROM quote_schedule WHERE quote_id='Q-1' ORDER BY line_no, month")
check("schedule of removed line 3 deleted, trimmed month deleted", len(sch) == 2, sch)
lns = rows("SELECT line_no FROM quote_lines WHERE quote_id='Q-1' ORDER BY line_no")
check("removed line 3 deleted", len(lns) == 2, lns)
check("neighbour quote Q-2 untouched", len(rows("SELECT * FROM quote_schedule WHERE quote_id='Q-2'")) == 1)
# 3. created_at survives, updated_at moves
r = rows("SELECT created_at, updated_at FROM quote_quotes WHERE quote_id='Q-1'")[0]
vals = list(r.values()) if isinstance(r, dict) else list(r)
ca, ua = (datetime.fromisoformat(str(v).replace("Z", "+00:00")) for v in vals)
check("created_at kept from first save, updated_at from second", ca == t1 and ua == t2, vals)
cl = rows("SELECT created_at FROM quote_lines WHERE quote_id='Q-1' AND line_no=1")[0]
check("a kept line's created_at kept too",
datetime.fromisoformat(str(list(cl.values())[0]).replace("Z", "+00:00")) == t1, cl)
# 4. save again unchanged: idempotent (no 23505, same state)
QUOTE.save(conn, mem([1, 2], [(1, 1), (2, 1)]), now=t2)
check("re-save is idempotent", len(rows("SELECT * FROM quote_schedule WHERE quote_id='Q-1'")) == 2)
# 5. money exact
amt = rows("SELECT amount FROM quote_lines WHERE quote_id='Q-1' AND line_no=1")[0]
check("DECIMAL(18,2) exact", "100.5" in json.dumps(amt), amt)
# 6. load returns the whole aggregate
class LiveLoad(Live):
def execute(self, stmt):
res = pg(lit(stmt)); rs = res.get("rows", []) if isinstance(res, dict) else []
cols = list(stmt.selected_columns)
class R(dict):
@property
def _mapping(self): return self
return [R({c.name: typed(c, x[c.name]) for c in cols}) for x in rs]
agg = QUOTE.load(LiveLoad(), {"quote_id": "Q-1"})
check("load returns root + 2 lines + 2 schedule rows",
agg and len(agg["quote_quotes"]) == 1 and len(agg["quote_lines"]) == 2 and len(agg["quote_schedule"]) == 2,
{k: len(v) for k, v in (agg or {}).items()})
check("load of a missing root is None", QUOTE.load(LiveLoad(), {"quote_id": "NOPE"}) is None)
# 7. upsert: a batched row never nulls a column it did not name
upsert(conn, contacts, [{"id": 1, "name": "A", "phone": "111"}, {"id": 2, "name": "B", "phone": "222"}], "id")
upsert(conn, contacts, [{"id": 1, "name": "A2", "phone": "999"}, {"id": 2, "name": "B2"}], "id")
c2 = rows("SELECT name, phone FROM crm_contacts WHERE id=2")[0]
check("row 2's phone kept (old db.upsert would set NULL)", "222" in json.dumps(c2) and "B2" in json.dumps(c2), c2)
# 8. remove
QUOTE.remove(conn, {"quote_id": "Q-1"})
left = [len(rows(f"SELECT * FROM {t} WHERE quote_id='Q-1'")) for t in ("quote_quotes", "quote_lines", "quote_schedule")]
check("remove clears every part", left == [0, 0, 0], left)
check("remove left Q-2 alone", len(rows("SELECT * FROM quote_quotes WHERE quote_id='Q-2'")) == 1)
for t in ("quote_schedule", "quote_lines", "quote_quotes", "crm_contacts"):
pg(f"DROP TABLE IF EXISTS {t}")
print(f"\n{sum(ok)}/{len(ok)} passed; tables dropped")
sys.exit(0 if all(ok) else 1)