peter
6ea750a698
The test's own writes are committed before a request, so a handler's Core DML can no longer hide in a transaction the test had already opened. The entry-point probe answers on a marked line the API's logging cannot mix into. A listed commit whose function no longer commits fails the fence. CLAUDE.md and the backend README now name what an AST cannot see; the route-commit tests no longer claim their read-back holds the commit.
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""Every process the app runs as sees every table before its first query.
|
|
|
|
The units declare their tables in their own `models` modules, and a core model names a unit's by
|
|
string (`User.subscriptions` → `Subscription`). A fresh interpreter that never imported the unit's
|
|
module fails to configure its mappers on its first query: the download child once did exactly that,
|
|
and every job failed, while this suite stayed green — it imports `app.tables` in-process, which hides
|
|
the gap. So each entry point is imported here in a clean interpreter, and must register the whole
|
|
schema on its own.
|
|
"""
|
|
|
|
import ast
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import app.tables # noqa: F401
|
|
from app.db import Base
|
|
|
|
BACKEND = Path(__file__).parent.parent
|
|
APP = BACKEND / "app"
|
|
|
|
# Each `python -m` module the deployment starts, and the API's app module.
|
|
ENTRY_POINTS = ("app.main", "app.worker", "app.downloads.runner")
|
|
|
|
# The table list goes out on one marked line: the API's logging writes to the same stdout, and a
|
|
# line it logs while importing is not the probe's answer.
|
|
_MARK = "entry-point-tables:"
|
|
_PROBE = (
|
|
"import importlib, json, sys\n"
|
|
"importlib.import_module(sys.argv[1])\n"
|
|
"from sqlalchemy.orm import configure_mappers\n"
|
|
"from app.db import Base\n"
|
|
"configure_mappers()\n"
|
|
f"print({_MARK!r} + json.dumps(sorted(Base.metadata.tables)))\n"
|
|
)
|
|
|
|
|
|
def test_each_entry_point_registers_the_whole_schema_on_its_own():
|
|
whole = sorted(Base.metadata.tables)
|
|
for module in ENTRY_POINTS:
|
|
r = subprocess.run(
|
|
[sys.executable, "-c", _PROBE, module],
|
|
cwd=BACKEND,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
check=False,
|
|
)
|
|
assert r.returncode == 0, (module, r.stderr[-2000:])
|
|
answer = [ln for ln in r.stdout.splitlines() if ln.startswith(_MARK)]
|
|
assert len(answer) == 1, (module, r.stdout[-2000:])
|
|
assert json.loads(answer[0].removeprefix(_MARK)) == whole, module
|
|
|
|
|
|
def test_every_module_that_runs_as_a_process_is_listed():
|
|
mains = set()
|
|
for path in APP.rglob("*.py"):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
for node in tree.body:
|
|
if (
|
|
isinstance(node, ast.If)
|
|
and isinstance(node.test, ast.Compare)
|
|
and isinstance(node.test.left, ast.Name)
|
|
and node.test.left.id == "__name__"
|
|
):
|
|
rel = path.relative_to(BACKEND).with_suffix("").as_posix()
|
|
mains.add(rel.replace("/", "."))
|
|
assert mains <= set(ENTRY_POINTS), mains - set(ENTRY_POINTS)
|