db1788bfe5
Batch-committing the in-flight app work (all modules compile clean): - app/payments.py, app/packs.py, app/discogs_mp.py, app/dealgod.py (new) - admin_routes, main, navigator/sales/shop routes, site/search.html - customer_woo_sync + disc_sync tweaks; test_dealgod_supply, test_startup_ddl Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
"""Self-check for the per-statement startup DDL guard (main._ensure_tables).
|
|
Proves the real guarantee: ONE failing statement is isolated — every other statement is still
|
|
attempted and the app boots. No live DB (fake engine). Run: .venv/bin/python test_startup_ddl.py"""
|
|
import asyncio
|
|
import app.main as m
|
|
|
|
|
|
class _FakeConn:
|
|
def __init__(self, boom):
|
|
self.boom, self.seen = boom, []
|
|
|
|
async def execute(self, stmt):
|
|
s = str(stmt)
|
|
self.seen.append(s)
|
|
if self.boom in s:
|
|
raise RuntimeError("simulated DDL failure")
|
|
|
|
|
|
class _FakeBegin:
|
|
def __init__(self, conn):
|
|
self.conn = conn
|
|
|
|
async def __aenter__(self):
|
|
return self.conn
|
|
|
|
async def __aexit__(self, *a):
|
|
return False # don't swallow — the loop's own try/except must handle it
|
|
|
|
|
|
class _FakeEngine:
|
|
def __init__(self, conn):
|
|
self.conn = conn
|
|
|
|
def begin(self):
|
|
return _FakeBegin(self.conn)
|
|
|
|
|
|
def test_isolation():
|
|
boom = "CREATE EXTENSION IF NOT EXISTS pg_trgm" # a real statement in the list
|
|
conn = _FakeConn(boom)
|
|
orig = m.engine
|
|
m.engine = _FakeEngine(conn)
|
|
try:
|
|
asyncio.run(m._ensure_tables()) # must NOT raise
|
|
finally:
|
|
m.engine = orig
|
|
assert any(boom in s for s in conn.seen), "the boom statement was never reached"
|
|
assert len(conn.seen) == len(m._STARTUP_DDL), \
|
|
f"only {len(conn.seen)}/{len(m._STARTUP_DDL)} attempted — a failure aborted the batch"
|
|
print(f"ok — all {len(conn.seen)} statements attempted; the failing one was isolated")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_isolation()
|