"""livecheck — run a brick's real use cases against a real DataK3 bucket, with only a CLI login.
Unit tests run a brick's repository on SQLite, which does not share DataK3's differences
(ON CONFLICT writing every VALUES column — tables#417; no read-your-writes inside a transaction;
its own SQL dialect). A live check runs the same use cases on a bucket instead. Every statement is
compiled and sent through `dodil data pg` — the Postgres-wire adapter the deployed app uses — so no
token has to be put in the environment.
from erp_platform import livecheck as lc
lc.create_schema(ROOT / "schema.sql", tables) # drops then creates
use_cases.repository = lc.repository(MyRepository)
lc.check("the story holds", condition, got)
...
raise SystemExit(lc.finish(tables)) # drops; non-zero if anything failed
Each statement commits on its own (the CLI has no session): this checks what is WRITTEN and READ,
not transaction boundaries. Harness differences from a real driver are handled here and say so.
When transaction boundaries ARE the thing under test (an optimistic version claim, a single-use link —
both decided at COMMIT on DataK3), `engine()` gives a real SQLAlchemy engine over the pg wire, its
credential taken from `dodil data connect` (your CLI login) and never printed.
"""
import sys
import json
import os
import re
import shutil
import subprocess
import time
from contextlib import contextmanager
from datetime import date, datetime
from decimal import Decimal
from sqlalchemy import BigInteger, String, cast
from sqlalchemy.dialects import postgresql
from sqlalchemy.sql import Select
DODIL = shutil.which("dodil") or "dodil"
BUCKET = os.environ.get("BUCKET", "")
ORG = os.environ.get("DODIL_ORG", "")
def _cli_env() -> dict:
"""The environment for a `dodil` call, without DODIL_TOKEN: with it set, the CLI resolves the database
name differently (it gained the org slug and the gateway answered "not authorized for database",
2026-10-17) — and a script that also drives erp_platform.db locally has to set it."""
return {k: v for k, v in os.environ.items() if k != "DODIL_TOKEN"}
def pg(sql: str, attempts: int = 5):
"""One statement. DataK3 answers DDL during a writer roll with SQLSTATE 08006 "retry the DDL" —
transient, and it says so; that one error is retried with backoff."""
if not BUCKET:
raise SystemExit("set BUCKET (a scratch bucket) and DODIL_ORG")
cmd = [DODIL, "data", "pg", "-b", BUCKET, sql, "-o", "json"] + (["--org", ORG] if ORG else [])
env = _cli_env()
for attempt in range(attempts):
r = subprocess.run(cmd, capture_output=True, text=True, env=env)
err = (r.stderr or r.stdout).strip()
if r.returncode and "SQLSTATE 08006" in err and attempt + 1 < attempts:
time.sleep(2 ** attempt)
continue
break
if r.returncode:
raise RuntimeError(f"{sql[:200]}\n -> {err[-400:]}")
try:
return json.loads(r.stdout)
except ValueError:
return {}
def engine(pool_size: int = 12):
"""A real engine on BUCKET over the pg wire (TLS, sslmode=require), authenticated as your CLI login.
The URL `dodil data connect -o json` prints carries the login token as its password: it is parsed here
and handed to psycopg, never printed or logged. Its sslmode is ignored — the wire is TLS and the
password is a credential, so it is always `require` (see erp_platform.db)."""
from urllib.parse import unquote, urlparse
import psycopg
if not BUCKET:
raise SystemExit("set BUCKET (a scratch bucket) and DODIL_ORG")
cmd = [DODIL, "data", "connect", BUCKET, "-o", "json"] + (["--org", ORG] if ORG else [])
r = subprocess.run(cmd, capture_output=True, text=True, env=_cli_env())
if r.returncode:
raise SystemExit("dodil data connect failed — is the CLI logged in? (dodil auth status)")
u = urlparse(json.loads(r.stdout)["pg"])
kw = dict(host=u.hostname, port=u.port or 5432, dbname=unquote(u.path.lstrip("/")),
user=unquote(u.username or "token"), password=unquote(u.password or ""), sslmode="require")
from erp_platform.pool import pooled_engine
return pooled_engine(lambda: psycopg.connect(**kw), pool_size=pool_size)
def rows(sql: str) -> list:
return pg(sql).get("rows", [])
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:
# the CLI prints fractions with as few digits as needed; Python 3.10 parses only 3 or 6
s = re.sub(r"\.(\d{1,6})", lambda m: "." + m.group(1).ljust(6, "0"), str(v).replace("Z", "+00:00"))
return datetime.fromisoformat(s)
if py is Decimal:
return Decimal(str(v))
if py is bool:
return v if isinstance(v, bool) else str(v).lower() in ("t", "true", "1")
return py(v)
class _Row(dict):
@property
def _mapping(self):
return self
def __iter__(self):
return iter(self.values())
def __getitem__(self, k):
return list(self.values())[k] if isinstance(k, int) else dict.__getitem__(self, k)
class _Result(list):
def first(self):
return self[0] if self else None
class Connection:
"""Enough of a SQLAlchemy connection for a repository: execute() → rows in SELECT order."""
def execute(self, stmt):
original = stmt
if isinstance(stmt, Select):
# CLIs before 0.1.80 passed BIGINT through a float in `-o json` (and every hosted MCP tool):
# 54848654143250435 came back as 54848654143250430 (cli-shell #109, fixed in 0.1.80,
# verified 2026-10-16). Kept: reading BIGINT as TEXT costs nothing and protects anyone on an
# older CLI, where every derived partner id (above 2^53) would silently be wrong.
stmt = stmt.with_only_columns(*[
cast(c, String).label(c.name) if isinstance(c.type, BigInteger) else c
for c in stmt.selected_columns])
sql = str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
# Inlined values are escaped for the driver's %-paramstyle ("21%" → "21%%"); the CLI sends
# the text as written. A real driver binds parameters and never sees this.
sql = sql.replace("%%", "%")
res = pg(sql)
if not isinstance(stmt, Select) and not hasattr(stmt, "selected_columns"):
return _Result()
cols = list(original.selected_columns)
return _Result(_Row((c.name, typed(c, r[c.name])) for c in cols) for r in res.get("rows", []))
def repository(factory):
"""A use-case `repository` context manager that yields factory(live connection)."""
@contextmanager
def repo():
yield factory(Connection())
return repo
def drop_all(tables, rounds: int = 2) -> list:
"""Drop, children first; what refuses is tried once more after a pause."""
left = list(reversed(list(tables)))
for n in range(rounds):
stuck = []
for t in left:
try:
pg(f"DROP TABLE IF EXISTS {t}", attempts=3)
except RuntimeError:
stuck.append(t)
left = stuck
if not left or n + 1 == rounds:
break
time.sleep(3 * (n + 1))
return left
def create_schema(schema_path, tables) -> None:
stuck = drop_all(tables)
if stuck:
print(f" could not drop {stuck} first — continuing on CREATE IF NOT EXISTS")
for stmt in re.split(r";\s*\n", open(schema_path).read()):
body = "\n".join(l for l in stmt.splitlines() if not l.strip().startswith("--")).strip().rstrip(";")
if not body:
continue
try:
pg(body)
except RuntimeError as exc:
# A schema's explicit ALTER … ADD COLUMN (erp_platform.migrate's additive rule) meets the column its own
# CREATE just made: already there is the normal case, as in erp-migrate.
msg = str(exc).lower()
if body.upper().startswith("ALTER") and ("already exists" in msg or "duplicate column" in msg):
continue
raise
print(f"schema: {len(list(tables))} tables on {BUCKET}")
_results: list = []
def check(name: str, cond, got=None) -> None:
_results.append(bool(cond))
print(("PASS " if cond else "FAIL ") + name + ("" if cond else f" got: {got}"))
def refused(fn, *a, **kw):
"""(status, reason) if the use case refused, else None."""
from erp_platform.bricks import Refused
try:
fn(*a, **kw)
except Refused as r:
return r.status, r.reason
return None
def finish(tables) -> int:
# called from a `finally:` that raises SystemExit, which REPLACES an exception in flight: a step that crashed before it
# recorded a check would otherwise end "N/N passed" and exit 0 (a live check hid a 500 that way, 2026-10-23)
crashed = sys.exc_info()[1]
if crashed is not None and not isinstance(crashed, SystemExit):
_results.append(False)
print(f"FAIL the run stopped: {type(crashed).__name__}: {str(crashed)[:300]}")
left = drop_all(tables)
print(f"\n{sum(_results)}/{len(_results)} passed; " + (f"COULD NOT DROP {left}" if left else "tables dropped"))
return 0 if _results and all(_results) and crashed is None else 1