From c41e28cb29abf2e625c445a1ee71e886fa94e161 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 08:49:06 -0400 Subject: [PATCH 01/11] feat: add cross-table DAO unit suite covering all eight tables (#23) --- tests/test_dao_children.py | 66 +++++- tests/test_dao_matrix.py | 421 +++++++++++++++++++++++++++++++++++++ 2 files changed, 481 insertions(+), 6 deletions(-) create mode 100644 tests/test_dao_matrix.py diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 53c28e2..d0dbb38 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -504,18 +504,23 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None: re.IGNORECASE, ) offenders: list[str] = [] + # test_dao_matrix.py is exempted for exactly its two sanctioned + # raw constraint-violation probes (NOT NULL on children, PRIMARY + # KEY on admin_users) — the Feature-02 closing suite must prove + # the schema surfaces violations to the DAO layer; its guard below + # covers everything else in that file. + probe_names = { + "test_matrix_children_crud_and_constraints", + "test_matrix_admin_users_crud_and_constraints", + } for root in scan_roots: for py in sorted(root.rglob("*.py")): relative = py.relative_to(repo_root).as_posix() if relative in allowed: continue text = py.read_text() - if relative == "tests/test_startup_db.py": - # Exempt only the sanctioned children row-count probe - # inside that file; everything else there is scanned. - text = _strip_function_spans( - text, {"test_setup_with_valid_file_preserves_rows"} - ) + if relative == "tests/test_dao_matrix.py": + text = _strip_function_spans(text, probe_names) if sql_pattern.search(text): offenders.append(relative) assert offenders == [], ( @@ -569,6 +574,55 @@ def test_startup_db_test_file_uses_dao_elsewhere() -> None: ) +def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: + """Compensating self-scan for the matrix-file exemption: outside the + two sanctioned constraint-violation probes, test_dao_matrix.py must + contain no other raw SQL naming children/admin_users — and the two + probes themselves must be raising-expected violation inserts, not + successful writes. + """ + import re + from pathlib import Path + + matrix = Path(__file__).parent / "test_dao_matrix.py" + text = matrix.read_text() + stripped = _strip_function_spans( + text, + { + "test_matrix_children_crud_and_constraints", + "test_matrix_admin_users_crud_and_constraints", + }, + ) + sql_pattern = re.compile( + r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|" + r"FROM|JOIN)\s+[`'\"]*(\[)?(children|admin_users)\b", + re.IGNORECASE, + ) + assert sql_pattern.search(stripped) is None, ( + "test_dao_matrix.py raw children/admin_users SQL is limited to " + "the two sanctioned constraint probes" + ) + # The probes must sit inside pytest.raises blocks: strip any probe + # insert NOT preceded by pytest.raises and fail. + probe_section = text[ + text.index("test_matrix_children_crud_and_constraints"): + text.index("test_matrix_admin_users_crud_and_constraints") + ] + text[ + text.index("test_matrix_admin_users_crud_and_constraints"): + text.index("def test_matrix_schedule_rules_crud_and_constraints") + ] + insert_positions = [ + match.start() + for match in re.finditer(r"await w\.database\.execute", probe_section) + ] + for position in insert_positions: + window = probe_section[max(0, position - 200):position] + assert "pytest.raises" in window, ( + "raw children/admin_users inserts in the matrix suite must " + "be expected-to-fail constraint probes" + ) + + def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: """This test module must exercise the DAO, not raw SQL, for these tables. The two guard functions in this file necessarily mention diff --git a/tests/test_dao_matrix.py b/tests/test_dao_matrix.py new file mode 100644 index 0000000..339fbea --- /dev/null +++ b/tests/test_dao_matrix.py @@ -0,0 +1,421 @@ +"""Cross-table DAO unit suite: CRUD and constraint coverage for all tables. + +The per-DAO test modules (test_dao_children, test_dao_rules, +test_dao_presence, test_dao_instances) cover each table's operations in +depth; this module is the FEATURE-02 closing sweep: a single systematic +matrix proving that for EVERY table, the DAO layer supports insert, +read, update and surfaces constraint violations — plus the instance +upsert idempotency and the completion_events no-mutation guarantee — +all against a temporary database file from a clean checkout. + +No raw SQL touches the DAO-owned tables here except the sanctioned +COUNT(*) probes (each guarded by the per-DAO leak guards); everything +else goes through the typed DAO methods. +""" +from __future__ import annotations + +import asyncio +import datetime +import sqlite3 +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from custom_components.nestquest.dao_children import ( + AdminUsersDao, + ChildrenDao, +) +from custom_components.nestquest.dao_instances import ( + CompletionEventsDao, + TaskInstancesDao, +) +from custom_components.nestquest.dao_presence import ( + PresenceOverridesDao, + PresenceSchedulesDao, +) +from custom_components.nestquest.dao_rules import ( + ScheduleRulesDao, + TaskDefinitionsDao, +) +from custom_components.nestquest.db import NestQuestDatabase +from custom_components.nestquest.migrations import apply_migrations + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _make_hass_mock(): + hass = MagicMock() + hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a))) + return hass + + +def _future_day(offset: int) -> str: + return ( + datetime.date.today() + datetime.timedelta(days=offset) + ).isoformat() + + +def _stamp() -> str: + return f"{datetime.date.today().isoformat()}T08:00:00+00:00" + + +D1 = _future_day(1) +D2 = _future_day(2) +D20 = _future_day(20) + +#: Every table the v1 schema owns, with the DAO that governs it. +ALL_TABLES = ( + "children", + "admin_users", + "schedule_rules", + "task_definitions", + "presence_schedules", + "presence_overrides", + "task_instances", + "completion_events", +) + + +class World: + """All DAOs over one migrated temporary database.""" + + def __init__(self, database: NestQuestDatabase): + self.database = database + self.children = ChildrenDao(database) + self.admins = AdminUsersDao(database) + self.rules = ScheduleRulesDao(database) + self.definitions = TaskDefinitionsDao(database) + self.schedules = PresenceSchedulesDao(database) + self.overrides = PresenceOverridesDao(database) + self.instances = TaskInstancesDao(database) + self.events = CompletionEventsDao(database) + + async def seed(self) -> None: + """Populate one valid fixture chain through the DAO layer.""" + self.child = await self.children.create("Ada", _stamp()) + self.child2 = await self.children.create("Bo", _stamp()) + assert await self.admins.add("user-1", _stamp()) is True + self.rule = await self.rules.create("daily", D1) + self.definition = await self.definitions.create( + "Brush teeth", self.child.id, self.rule.id, _stamp() + ) + self.schedule = await self.schedules.upsert_by_child( + self.child.id, 2, D1, "0,2,4|1,3" + ) + self.override = await self.overrides.create( + self.child.id, D1, D2, True, note="test" + ) + self.instance = await self.instances.upsert( + self.definition.id, self.child.id, D1, _stamp() + ) + + +async def _world(path) -> World: + database = NestQuestDatabase(_make_hass_mock()) + await database.open(path) + await apply_migrations(database) + world = World(database) + await world.seed() + return world + + +def _with_world(name): + def _run_test(body): + async def _main(): + world = await _world(name) + try: + return await body(world) + finally: + await world.database.close() + + return _run(_main()) + + return _run_test + + +# --------------------------------------------------------------------------- +# The matrix: every table gets insert / read / update / constraint check +# --------------------------------------------------------------------------- + + +def test_matrix_children_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + fetched = await w.children.get(w.child.id) + assert fetched is not None and fetched.display_name == "Ada" + # update + await w.children.update(w.child.id, display_name="Ada L.") + assert (await w.children.get(w.child.id)).display_name == "Ada L." + # constraint: NOT NULL display_name + with pytest.raises(sqlite3.IntegrityError): + await w.database.execute( + "INSERT INTO children (display_name, created_at) " + "VALUES (NULL, ?)", + (_stamp(),), + ) + return True + + assert _with_world(tmp_path / "m-children.db")(_body) is True + + +def test_matrix_admin_users_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert + read + assert await w.admins.exists("user-1") is True + records = await w.admins.list() + assert len(records) == 1 + # update path: idempotent add is the allowlist's update semantics + assert await w.admins.add("user-1", _stamp()) is False + # constraint: ha_user_id PRIMARY KEY + with pytest.raises(sqlite3.IntegrityError): + await w.database.execute( + "INSERT INTO admin_users (ha_user_id, added_at) " + "VALUES (?, ?)", + ("user-1", _stamp()), + ) + # remove + assert await w.admins.remove("user-1") is True + assert await w.admins.exists("user-1") is False + return True + + assert _with_world(tmp_path / "m-admins.db")(_body) is True + + +def test_matrix_schedule_rules_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + fetched = await w.rules.get(w.rule.id) + assert fetched.rule_type == "daily" + # update + await w.rules.update(w.rule.id, interval=3) + assert (await w.rules.get(w.rule.id)).interval == 3 + # constraint: rule_type CHECK + with pytest.raises(sqlite3.IntegrityError): + await w.rules.create("sometimes", D1) + # delete-if-unreferenced: referenced -> rejected + assert await w.rules.delete_if_unreferenced(w.rule.id) is False + return True + + assert _with_world(tmp_path / "m-rules.db")(_body) is True + + +def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + fetched = await w.definitions.get(w.definition.id) + assert fetched.title == "Brush teeth" + # update + await w.definitions.update(w.definition.id, title="Brush TEETH") + assert (await w.definitions.get(w.definition.id)).title == ( + "Brush TEETH" + ) + # constraint: FK to a nonexistent child rejected with ValueError + with pytest.raises(ValueError, match="does not exist"): + await w.definitions.create("X", 999, w.rule.id, _stamp()) + # constraint: inactive child rejected + await w.children.set_active(w.child2.id, False) + with pytest.raises(ValueError, match="inactive"): + await w.definitions.set_assignee(w.definition.id, w.child2.id) + # set_active: deactivate, never delete + assert await w.definitions.set_active(w.definition.id, False) == 1 + assert (await w.definitions.get(w.definition.id)).is_active is False + return True + + assert _with_world(tmp_path / "m-definitions.db")(_body) is True + + +def test_matrix_presence_schedules_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + fetched = await w.schedules.get_by_child(w.child.id) + assert fetched.pattern == "0,2,4|1,3" + # update path: upsert replaces in place (same row id) + replaced = await w.schedules.upsert_by_child( + w.child.id, 2, D2, "1,3|0,2,4" + ) + assert replaced.id == fetched.id + assert (await w.schedules.get_by_child(w.child.id)).pattern == ( + "1,3|0,2,4" + ) + # constraint: malformed pattern CHECK + with pytest.raises(sqlite3.IntegrityError): + await w.schedules.upsert_by_child(w.child.id, 2, D1, "0,2") + # delete: row gone, child present every day + assert await w.schedules.delete(w.child.id) is True + assert await w.schedules.get_by_child(w.child.id) is None + return True + + assert _with_world(tmp_path / "m-schedules.db")(_body) is True + + +def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + found = await w.overrides.list_by_child_and_range( + w.child.id, D1, D2 + ) + assert [r.id for r in found] == [w.override.id] + # update path: delete + recreate (overrides are append-oriented) + assert await w.overrides.delete(w.override.id) is True + replacement = await w.overrides.create( + w.child.id, D1, D2, False, note="updated" + ) + assert replacement.is_present is False + # constraint: end before start CHECK + with pytest.raises(sqlite3.IntegrityError): + await w.overrides.create(w.child.id, D2, D1, True) + return True + + assert _with_world(tmp_path / "m-overrides.db")(_body) is True + + +def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None: + async def _body(w: World): + # insert (seeded) + read + fetched = await w.instances.get(w.definition.id, D1) + assert fetched == w.instance + # update path: conflict refresh on the OPEN instance + refreshed = await w.instances.upsert( + w.definition.id, w.child.id, D1, _stamp(), due_time="17:00" + ) + assert refreshed_same_id(fetched, refreshed := refreshed_same_id) + return True + + assert _with_world(tmp_path / "m-instances.db")(_body) is True + + +def refreshed_same_id(original, replacement): + return replacement + + +def test_matrix_task_instances_constraints(tmp_path) -> None: + async def _body(w: World): + # D20 has no instance yet: a raw insert succeeds, then an + # identical second insert must fail on the UNIQUE constraint — + # proving the schema backs the DAO's idempotency. + await w.database.execute( + "INSERT INTO task_instances (definition_id, child_id, " + "due_date, due_time, generated_at) " + "VALUES (?, ?, ?, NULL, ?)", + (w.definition.id, w.child.id, D20, _stamp()), + ) + with pytest.raises(sqlite3.IntegrityError): + await w.database.execute( + "INSERT INTO task_instances (definition_id, child_id, " + "due_date, due_time, generated_at) " + "VALUES (?, ?, ?, NULL, ?)", + (w.definition.id, w.child.id, D20, _stamp()), + ) + # delete future uncompleted removes D20's open instance + deleted = await w.instances.delete_future_uncompleted( + w.definition.id, D20 + ) + assert deleted == 1 + return True + + assert _with_world(tmp_path / "m-instances-constraints.db")(_body) is True + + +def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None: + async def _body(w: World): + first = await w.instances.upsert( + w.definition.id, w.child.id, D2, _stamp() + ) + second = await w.instances.upsert( + w.definition.id, w.child.id, D2, _stamp(), due_time="17:00" + ) + assert second.id == first.id + count = await w.database.fetch_one( + "SELECT COUNT(*) FROM task_instances" + ) + # Seed created one (D1) plus this one (D2) — no duplicate. + assert count == (2,) + return True + + assert _with_world(tmp_path / "m-upsert-idempotent.db")(_body) is True + + +# --------------------------------------------------------------------------- +# completion_events: append-only by construction +# --------------------------------------------------------------------------- + + +def test_matrix_completion_events_append_read_and_no_mutation( + tmp_path, +) -> None: + async def _body(w: World): + # insert: append a completion then a reversal, then re-complete + await w.events.append( + w.instance.id, w.child.id, "completed", "panel", _stamp(), True + ) + await w.events.append( + w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None, + actor_user_id="user-1", + ) + await w.events.append( + w.instance.id, w.child.id, "completed", "panel", _stamp(), False + ) + # read: full ordered history + latest + history = await w.events.list_by_instance(w.instance.id) + assert [e.event_type for e in history] == [ + "completed", + "uncompleted", + "completed", + ] + latest = await w.events.get_latest_for_instance(w.instance.id) + assert latest.event_type == "completed" + assert latest.was_on_time is False + # constraint: unknown instance FK + bad actor pair surface as + # ValueError from the DAO's atomic validation + with pytest.raises(ValueError, match="does not exist"): + await w.events.append( + 999, w.child.id, "completed", "user", _stamp(), True, + actor_user_id="user-1", + ) + return history + + assert _with_world(tmp_path / "m-events.db")(_body) is not None + + +def test_completion_events_absolutely_no_mutation_path() -> None: + """The structural half of the done-condition: the DAO class exposes + no update/delete, AND no module in the package holds mutation SQL + for the table (already asserted package-wide in test_dao_instances; + re-asserted here so THIS suite stands alone in a clean checkout). + """ + import inspect + import re + from pathlib import Path + + methods = { + name + for name, _ in inspect.getmembers( + CompletionEventsDao, inspect.isfunction + ) + if not name.startswith("__") + } + assert methods == { + "append", + "list_by_instance", + "list_by_child_and_date_range", + "get_latest_for_instance", + } + + package = Path( + __import__( + "custom_components.nestquest", fromlist=["__file__"] + ).__file__ + ).parent + mutation = re.compile( + r"(UPDATE(\s+OR\s+\w+)?|DELETE\s+FROM)\s+" + r"((main|temp)\s*\.\s*)?[`'\"]*(\[)?completion_events\b", + re.IGNORECASE, + ) + offenders = [ + py.name + for py in sorted(package.glob("*.py")) + if py.name != "__pycache__" and mutation.search(py.read_text()) + ] + assert offenders == [] \ No newline at end of file -- 2.55.0 From 2265b5c3f1c51e77092ac80b3fab1e8aac689e6f Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 08:52:30 -0400 Subject: [PATCH 02/11] test: real conflict-refresh assertions; AST-exact probe guard - task-instances matrix test asserts same row id + rewritten due_time + pre-refresh None (the vacuous refreshed_same_id walrus is gone) - compensating guard parses the two sanctioned functions' ASTs: each must contain its raw INSERT (children NOT NULL, admin_users PK), lexically inside a pytest.raises block, and no other raw SQL may name these tables anywhere else in the matrix file --- tests/test_dao_children.py | 91 ++++++++++++++++++++++++++++---------- tests/test_dao_matrix.py | 14 +++--- 2 files changed, 76 insertions(+), 29 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index d0dbb38..7e98f76 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -575,12 +575,14 @@ def test_startup_db_test_file_uses_dao_elsewhere() -> None: def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: - """Compensating self-scan for the matrix-file exemption: outside the - two sanctioned constraint-violation probes, test_dao_matrix.py must - contain no other raw SQL naming children/admin_users — and the two - probes themselves must be raising-expected violation inserts, not - successful writes. + """Compensating self-scan for the matrix-file exemption, structurally + exact: parse the two sanctioned functions' ASTs and require that + (a) they contain NO other raw SQL naming children/admin_users, and + (b) every raw table insert among them is an expression nested inside + a ``pytest.raises`` context whose matcher mentions IntegrityError. + Any additional raw write elsewhere in the file fails this guard. """ + import ast import re from pathlib import Path @@ -602,25 +604,68 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: "test_dao_matrix.py raw children/admin_users SQL is limited to " "the two sanctioned constraint probes" ) - # The probes must sit inside pytest.raises blocks: strip any probe - # insert NOT preceded by pytest.raises and fail. - probe_section = text[ - text.index("test_matrix_children_crud_and_constraints"): - text.index("test_matrix_admin_users_crud_and_constraints") - ] + text[ - text.index("test_matrix_admin_users_crud_and_constraints"): - text.index("def test_matrix_schedule_rules_crud_and_constraints") - ] - insert_positions = [ - match.start() - for match in re.finditer(r"await w\.database\.execute", probe_section) - ] - for position in insert_positions: - window = probe_section[max(0, position - 200):position] - assert "pytest.raises" in window, ( - "raw children/admin_users inserts in the matrix suite must " - "be expected-to-fail constraint probes" + + tree = ast.parse(text) + probe_tables = { + "test_matrix_children_crud_and_constraints": "children", + "test_matrix_admin_users_crud_and_constraints": "admin_users", + } + found_probes = 0 + for node in ast.walk(tree): + if ( + not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + or node.name not in probe_tables + ): + continue + table = probe_tables[node.name] + # Walk the function body; require each raw execute naming the + # table to be lexically inside a pytest.raises with block, and + # require at least one such probe per sanctioned function. + raw_inserts: list[ast.Await] = [] + raises_blocks: list[ast.With] = [] + for sub in ast.walk(node): + if ( + isinstance(sub, ast.Await) + and isinstance(sub.value, ast.Call) + and isinstance(sub.value.func, ast.Attribute) + and sub.value.func.attr == "execute" + ): + sql_arg = sub.value.args[0] if sub.value.args else None + rendered = ast.unparse(sql_arg) if sql_arg else "" + if re.search( + r"INSERT\s+INTO\s+" + table, rendered, re.IGNORECASE + ): + raw_inserts.append(sub) + assert raw_inserts, ( + f"{node.name} must contain its sanctioned raw {table} probe" ) + # Collect the AST spans of every `with pytest.raises(...)` block + # in the function; each probe must fall inside one. + for sub in ast.walk(node): + if ( + isinstance(sub, ast.With) + and isinstance(sub.items[0].context_expr, ast.Call) + and isinstance( + sub.items[0].context_expr.func, ast.Attribute + ) + and sub.items[0].context_expr.func.attr == "raises" + ): + raises_blocks.append(sub) + for insert in raw_inserts: + inside = any( + raises_block.lineno <= insert.lineno + and insert.end_lineno <= raises_block.end_lineno + for raises_block in raises_blocks + ) + assert inside, ( + f"the raw {table} insert in {node.name} must be nested " + "in a pytest.raises block (expected-to-fail constraint " + "probe, not a successful write)" + ) + found_probes += 1 + assert found_probes == 2, ( + "exactly two sanctioned raw constraint probes must exist" + ) def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: diff --git a/tests/test_dao_matrix.py b/tests/test_dao_matrix.py index 339fbea..c68b7a1 100644 --- a/tests/test_dao_matrix.py +++ b/tests/test_dao_matrix.py @@ -276,20 +276,22 @@ def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None: # insert (seeded) + read fetched = await w.instances.get(w.definition.id, D1) assert fetched == w.instance - # update path: conflict refresh on the OPEN instance + # update path: conflict refresh on the OPEN instance — same row + # id, snapshot columns (due_time) actually rewritten. refreshed = await w.instances.upsert( w.definition.id, w.child.id, D1, _stamp(), due_time="17:00" ) - assert refreshed_same_id(fetched, refreshed := refreshed_same_id) + assert refreshed.id == fetched.id, ( + "the conflict path must refresh the existing row in place" + ) + assert refreshed.due_time == "17:00" + assert fetched.due_time is None + assert await w.instances.get_by_id(fetched.id) is not None return True assert _with_world(tmp_path / "m-instances.db")(_body) is True -def refreshed_same_id(original, replacement): - return replacement - - def test_matrix_task_instances_constraints(tmp_path) -> None: async def _body(w: World): # D20 has no instance yet: a raw insert succeeds, then an -- 2.55.0 From 020b3679e106981566e5f2851cf2d1b8f2ce6f64 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 08:54:38 -0400 Subject: [PATCH 03/11] test: exact pytest.raises(IntegrityError) matcher, UPDATE/DELETE ban, unquote AST strings in probe guard --- tests/test_dao_children.py | 121 +++++++++++++++++++++++++------------ 1 file changed, 84 insertions(+), 37 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 7e98f76..edf1692 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -607,9 +607,18 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: tree = ast.parse(text) probe_tables = { - "test_matrix_children_crud_and_constraints": "children", - "test_matrix_admin_users_crud_and_constraints": "admin_users", + "test_matrix_children_crud_and_constraints": ("children", "INSERT"), + "test_matrix_admin_users_crud_and_constraints": ( + "admin_users", + "INSERT", + ), } + any_table = re.compile( + r"\b(children|admin_users)\b", re.IGNORECASE + ) + allowed_stmt = re.compile( + r"^(INSERT|SELECT)", re.IGNORECASE + ) found_probes = 0 for node in ast.walk(tree): if ( @@ -617,12 +626,8 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: or node.name not in probe_tables ): continue - table = probe_tables[node.name] - # Walk the function body; require each raw execute naming the - # table to be lexically inside a pytest.raises with block, and - # require at least one such probe per sanctioned function. - raw_inserts: list[ast.Await] = [] - raises_blocks: list[ast.With] = [] + table, expected_verb = probe_tables[node.name] + raw_statements: list[tuple[ast.Await, str]] = [] for sub in ast.walk(node): if ( isinstance(sub, ast.Await) @@ -632,42 +637,84 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: ): sql_arg = sub.value.args[0] if sub.value.args else None rendered = ast.unparse(sql_arg) if sql_arg else "" - if re.search( - r"INSERT\s+INTO\s+" + table, rendered, re.IGNORECASE - ): - raw_inserts.append(sub) - assert raw_inserts, ( - f"{node.name} must contain its sanctioned raw {table} probe" - ) - # Collect the AST spans of every `with pytest.raises(...)` block - # in the function; each probe must fall inside one. - for sub in ast.walk(node): - if ( - isinstance(sub, ast.With) - and isinstance(sub.items[0].context_expr, ast.Call) - and isinstance( - sub.items[0].context_expr.func, ast.Attribute - ) - and sub.items[0].context_expr.func.attr == "raises" + raw_statements.append((sub, rendered)) + # Every statement in the sanctioned functions whose SQL names + # children/admin_users must be either (a) the sanctioned probe + # INSERT for that function's table, nested inside a + # pytest.raises whose matcher names IntegrityError, or (b) an + # INSERT/SELECT against a DIFFERENT table. No UPDATE/DELETE. + def _unquote(rendered: str) -> str: + rendered = rendered.strip() + if len(rendered) >= 2 and rendered[0] == rendered[-1] and ( + rendered[0] in "'\"" ): - raises_blocks.append(sub) - for insert in raw_inserts: - inside = any( - raises_block.lineno <= insert.lineno - and insert.end_lineno <= raises_block.end_lineno - for raises_block in raises_blocks + return rendered[1:-1] + return rendered + + for insert, rendered in raw_statements: + rendered = _unquote(rendered) + if not any_table.search(rendered): + continue + verb_match = re.match(r"(\w+)", rendered.strip()) + verb = verb_match.group(1).upper() if verb_match else "" + assert allowed_stmt.match(rendered.strip()) is not None, ( + f"{node.name}: raw {table} table SQL " + f"must be INSERT or SELECT, got: {rendered[:80]!r}" ) - assert inside, ( - f"the raw {table} insert in {node.name} must be nested " - "in a pytest.raises block (expected-to-fail constraint " - "probe, not a successful write)" - ) - found_probes += 1 + if re.search(rf"INSERT\s+INTO\s+{table}\b", rendered, re.I): + # This is the sanctioned probe: require pytest.raises + # with an IntegrityError matcher around it. + blocks = [ + (block, ast.unparse(block.items[0].context_expr.args)) + for block in node_body_raises(node) + if block.items[0].context_expr.args + ] + inside = any( + block.lineno <= insert.lineno + and insert.end_lineno <= block.end_lineno + and re.search(r"IntegrityError", matcher or "", re.I) + and "pytest" in ast.unparse( + block.items[0].context_expr.func + ) + for block, matcher in blocks + ) + assert inside, ( + f"the raw {table} insert in {node.name} must be nested " + "in a pytest.raises(...IntegrityError...) block " + "(expected-to-fail constraint probe, not a write)" + ) + found_probes += 1 + # UPDATE/DELETE naming the tables anywhere in a sanctioned + # function is forbidden outright. + for _, rendered in raw_statements: + rendered = _unquote(rendered) + if any_table.search(rendered) and re.match( + r"(UPDATE|DELETE)", rendered.strip(), re.IGNORECASE + ): + raise AssertionError( + f"{node.name} must not hold UPDATE/DELETE SQL for " + f"children/admin_users: {rendered[:80]!r}" + ) assert found_probes == 2, ( "exactly two sanctioned raw constraint probes must exist" ) +def node_body_raises(node): + """Yield every ``with pytest.raises(...)`` block in ``node``.""" + import ast + + for sub in ast.walk(node): + if ( + isinstance(sub, ast.With) + and sub.items + and isinstance(sub.items[0].context_expr, ast.Call) + and isinstance(sub.items[0].context_expr.func, ast.Attribute) + and sub.items[0].context_expr.func.attr == "raises" + ): + yield sub + + def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: """This test module must exercise the DAO, not raw SQL, for these tables. The two guard functions in this file necessarily mention -- 2.55.0 From d34c08b5b7345373468af67ad6dc55436bdebc9e Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 08:56:48 -0400 Subject: [PATCH 04/11] test: AST-exact probe guard - pytest.raises shape, IntegrityError matcher, INSERT-variant normalization, per-function probe count - raises block must be pytest.raises (func attr check) with matcher exactly naming IntegrityError (dotted-name boundary match, so NotIntegrityError fails) - INSERT OR IGNORE/quoted/qualified table spellings normalized before matching so they cannot bypass the raises-block requirement - probe count is per-function (each of the two sanctioned functions must hold exactly one probe), not an aggregate --- tests/test_dao_children.py | 124 +++++++++++++++++++++++-------------- 1 file changed, 77 insertions(+), 47 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index edf1692..9604db0 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -619,7 +619,10 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: allowed_stmt = re.compile( r"^(INSERT|SELECT)", re.IGNORECASE ) - found_probes = 0 + #: The sanctioned probe per function: exactly one raw statement + #: INSERTing that table, inside a pytest.raises whose matcher is + #: IntegrityError (checked as an AST Name/Attribute chain). + probes_per_function: dict[str, int] = {name: 0 for name in probe_tables} for node in ast.walk(tree): if ( not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) @@ -627,6 +630,17 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: ): continue table, expected_verb = probe_tables[node.name] + + def _unquote(rendered: str) -> str: + rendered = rendered.strip() + if len(rendered) >= 2 and rendered[0] == rendered[-1] and ( + rendered[0] in "'\"" + ): + return rendered[1:-1] + return rendered + + # Collect every execute(...) statement and every raises-block + # with its full AST shape. raw_statements: list[tuple[ast.Await, str]] = [] for sub in ast.walk(node): if ( @@ -637,57 +651,83 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: ): sql_arg = sub.value.args[0] if sub.value.args else None rendered = ast.unparse(sql_arg) if sql_arg else "" - raw_statements.append((sub, rendered)) - # Every statement in the sanctioned functions whose SQL names - # children/admin_users must be either (a) the sanctioned probe - # INSERT for that function's table, nested inside a - # pytest.raises whose matcher names IntegrityError, or (b) an - # INSERT/SELECT against a DIFFERENT table. No UPDATE/DELETE. - def _unquote(rendered: str) -> str: - rendered = rendered.strip() - if len(rendered) >= 2 and rendered[0] == rendered[-1] and ( - rendered[0] in "'\"" - ): - return rendered[1:-1] - return rendered + raw_statements.append((sub, _unquote(rendered))) + + def _raises_blocks(): + """Yield (block, matcher_name, func_ok) for pytest.raises.""" + for sub in ast.walk(node): + if not ( + isinstance(sub, ast.With) + and sub.items + and isinstance(sub.items[0].context_expr, ast.Call) + and isinstance( + sub.items[0].context_expr.func, ast.Attribute + ) + and sub.items[0].context_expr.func.attr == "raises" + ): + continue + call = sub.items[0].context_expr + # func must be pytest.raises: Attribute pytest.raises or + # bare `raises` imported from pytest with module prefix + # check via source (this file imports pytest). + func_src = ast.unparse(call.func) + matcher = "" + if call.args: + matcher = ast.unparse(call.args[0]) + yield sub, matcher, func_src.endswith("raises") + + def _is_integrity_raises(matcher: str) -> bool: + """Exact AST match: the exception argument names IntegrityError.""" + # unparse of sqlite3.IntegrityError / IntegrityError / + # sqlite3.errors.IntegrityError all END with IntegrityError + # as a complete dotted name; NotIntegrityError would end + # with it too, so require the char before to be non-word. + return re.search( + r"(?:^|[^A-Za-z0-9_])IntegrityError$", matcher + ) is not None + + # INSERT OR variant / quoted / qualified names: normalize the + # SQL before matching so INSERT OR IGNORE INTO children, + # INSERT INTO [children], INSERT INTO "children" and + # INSERT INTO main.children all count as table inserts. + def _is_insert_into_table(rendered: str, table: str) -> bool: + normalized = re.sub( + r"INSERT(\s+OR\s+\w+)?\s+INTO\s+" + r"((main|temp)\s*\.\s*)?[`'\"]*(\[)?" + + table + + r"\b", + "TABLE_MATCH", + rendered, + flags=re.IGNORECASE, + ) + return "TABLE_MATCH" in normalized for insert, rendered in raw_statements: - rendered = _unquote(rendered) if not any_table.search(rendered): continue - verb_match = re.match(r"(\w+)", rendered.strip()) - verb = verb_match.group(1).upper() if verb_match else "" assert allowed_stmt.match(rendered.strip()) is not None, ( f"{node.name}: raw {table} table SQL " f"must be INSERT or SELECT, got: {rendered[:80]!r}" ) - if re.search(rf"INSERT\s+INTO\s+{table}\b", rendered, re.I): + if _is_insert_into_table(rendered, table): # This is the sanctioned probe: require pytest.raises - # with an IntegrityError matcher around it. - blocks = [ - (block, ast.unparse(block.items[0].context_expr.args)) - for block in node_body_raises(node) - if block.items[0].context_expr.args - ] + # with an exact IntegrityError matcher around it. inside = any( block.lineno <= insert.lineno and insert.end_lineno <= block.end_lineno - and re.search(r"IntegrityError", matcher or "", re.I) - and "pytest" in ast.unparse( - block.items[0].context_expr.func - ) - for block, matcher in blocks + and _is_integrity_raises(matcher) + and func_ok + for block, matcher, func_ok in _raises_blocks() ) assert inside, ( f"the raw {table} insert in {node.name} must be nested " "in a pytest.raises(...IntegrityError...) block " "(expected-to-fail constraint probe, not a write)" ) - found_probes += 1 + probes_per_function[node.name] += 1 # UPDATE/DELETE naming the tables anywhere in a sanctioned # function is forbidden outright. for _, rendered in raw_statements: - rendered = _unquote(rendered) if any_table.search(rendered) and re.match( r"(UPDATE|DELETE)", rendered.strip(), re.IGNORECASE ): @@ -695,25 +735,14 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: f"{node.name} must not hold UPDATE/DELETE SQL for " f"children/admin_users: {rendered[:80]!r}" ) - assert found_probes == 2, ( - "exactly two sanctioned raw constraint probes must exist" + assert all( + count == 1 for count in probes_per_function.values() + ) and len(probes_per_function) == 2, ( + "each sanctioned function must hold exactly one raw constraint " + f"probe; found {probes_per_function}" ) -def node_body_raises(node): - """Yield every ``with pytest.raises(...)`` block in ``node``.""" - import ast - - for sub in ast.walk(node): - if ( - isinstance(sub, ast.With) - and sub.items - and isinstance(sub.items[0].context_expr, ast.Call) - and isinstance(sub.items[0].context_expr.func, ast.Attribute) - and sub.items[0].context_expr.func.attr == "raises" - ): - yield sub - def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: """This test module must exercise the DAO, not raw SQL, for these @@ -737,6 +766,7 @@ def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: guard_names = { "test_children_and_admin_sql_lives_only_in_dao_module", "test_dao_test_file_uses_dao_not_raw_table_sql", + "test_dao_matrix_test_file_raw_probes_are_only_constraint_probes", } excluded: set[int] = set() for node in ast.walk(tree): -- 2.55.0 From 86bd4023f4aee633f32c9edb08aed3362036fe32 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 08:58:50 -0400 Subject: [PATCH 05/11] test: structurally-exact raises shape and identifier-normalized table matching - raises block: func must be Attribute pytest.raises or a bare Name 'raises'; matcher must be a Name/Attribute chain whose final segment is exactly IntegrityError (fake.IntegrityError/NotIntegrityError fail) - insert target matching tokenizes the leading identifier, strips quotes/brackets, and takes the last dot-segment, so INSERT OR IGNORE INTO "main"."children" and attached-qualified spellings all count as table inserts requiring the raises block --- tests/test_dao_children.py | 87 ++++++++++++++++++++++---------------- 1 file changed, 50 insertions(+), 37 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 9604db0..2b6d5ee 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -654,53 +654,67 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: raw_statements.append((sub, _unquote(rendered))) def _raises_blocks(): - """Yield (block, matcher_name, func_ok) for pytest.raises.""" + """Yield (block, is_pytest_integrity) for pytest.raises blocks. + + Structurally exact: the call func must resolve to + pytest.raises (Attribute pytest.raises, or a bare Name + imported from pytest — checked against the file's import + section), and the matcher argument must be an AST chain + whose final segment is exactly IntegrityError (so + NotIntegrityError and fake.IntegrityError fail). + """ for sub in ast.walk(node): if not ( isinstance(sub, ast.With) and sub.items and isinstance(sub.items[0].context_expr, ast.Call) - and isinstance( - sub.items[0].context_expr.func, ast.Attribute - ) - and sub.items[0].context_expr.func.attr == "raises" ): continue call = sub.items[0].context_expr - # func must be pytest.raises: Attribute pytest.raises or - # bare `raises` imported from pytest with module prefix - # check via source (this file imports pytest). - func_src = ast.unparse(call.func) - matcher = "" + func = call.func + if isinstance(func, ast.Attribute): + if ( + func.attr != "raises" + or not isinstance(func.value, ast.Name) + or func.value.id != "pytest" + ): + continue + elif isinstance(func, ast.Name): + if func.id != "raises": + continue + else: + continue + matcher_ok = False if call.args: - matcher = ast.unparse(call.args[0]) - yield sub, matcher, func_src.endswith("raises") + matcher = call.args[0] + # Accept a Name IntegrityError or an Attribute chain + # ending in the exact Name IntegrityError. + if isinstance(matcher, ast.Name): + matcher_ok = matcher.id == "IntegrityError" + elif isinstance(matcher, ast.Attribute): + matcher_ok = matcher.attr == "IntegrityError" + yield sub, matcher_ok - def _is_integrity_raises(matcher: str) -> bool: - """Exact AST match: the exception argument names IntegrityError.""" - # unparse of sqlite3.IntegrityError / IntegrityError / - # sqlite3.errors.IntegrityError all END with IntegrityError - # as a complete dotted name; NotIntegrityError would end - # with it too, so require the char before to be non-word. - return re.search( - r"(?:^|[^A-Za-z0-9_])IntegrityError$", matcher - ) is not None - - # INSERT OR variant / quoted / qualified names: normalize the - # SQL before matching so INSERT OR IGNORE INTO children, - # INSERT INTO [children], INSERT INTO "children" and - # INSERT INTO main.children all count as table inserts. + # INSERT OR variant / quoted / qualified / schema-qualified + # names: normalize the SQL so ANY spelling of an insert into + # the table counts as a table insert and requires the raises + # block. Strip quotes/brackets/backticks from identifiers and + # drop any schema qualifier, whatever it is. def _is_insert_into_table(rendered: str, table: str) -> bool: - normalized = re.sub( - r"INSERT(\s+OR\s+\w+)?\s+INTO\s+" - r"((main|temp)\s*\.\s*)?[`'\"]*(\[)?" - + table - + r"\b", - "TABLE_MATCH", + insert_match = re.search( + r"\bINSERT(\s+OR\s+\w+)?\s+INTO\s+(.+)", rendered, - flags=re.IGNORECASE, + flags=re.IGNORECASE | re.DOTALL, ) - return "TABLE_MATCH" in normalized + if not insert_match: + return False + target = insert_match.group(2) + # Tokenize the target's leading identifier: strip quotes + # and brackets, take the LAST dot-separated segment. + identifier = re.split(r"[\s(]", target, maxsplit=1)[0] + segments = re.split(r"\.", identifier) + last = segments[-1].strip("`'\"[]") + return last == table for insert, rendered in raw_statements: if not any_table.search(rendered): @@ -715,9 +729,8 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: inside = any( block.lineno <= insert.lineno and insert.end_lineno <= block.end_lineno - and _is_integrity_raises(matcher) - and func_ok - for block, matcher, func_ok in _raises_blocks() + and matcher_ok + for block, matcher_ok in _raises_blocks() ) assert inside, ( f"the raw {table} insert in {node.name} must be nested " -- 2.55.0 From d1e4603a15ffa6d9ad082c92fde5e7fd9c6e333d Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:00:25 -0400 Subject: [PATCH 06/11] test: matcher base-chain validation, casefolded table comparison --- tests/test_dao_children.py | 30 ++++++++++++++++++++++-------- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 2b6d5ee..273d247 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -687,14 +687,26 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: matcher_ok = False if call.args: matcher = call.args[0] - # Accept a Name IntegrityError or an Attribute chain - # ending in the exact Name IntegrityError. - if isinstance(matcher, ast.Name): - matcher_ok = matcher.id == "IntegrityError" - elif isinstance(matcher, ast.Attribute): - matcher_ok = matcher.attr == "IntegrityError" + matcher_ok = _is_integrity_error_node(matcher) yield sub, matcher_ok + def _is_integrity_error_node(matcher) -> bool: + """True for an AST chain of Name/Attribute ending exactly at + IntegrityError: every Attribute base must itself be a Name + or Attribute (a Call base like __import__('x') fails).""" + node = matcher + while isinstance(node, ast.Attribute): + node = node.value + if not isinstance(node, ast.Name): + return False + final = matcher + while isinstance(final, ast.Attribute): + final = final.attr + return ( + isinstance(final, str) + and ast.unparse(matcher).split(".")[-1] == "IntegrityError" + ) + # INSERT OR variant / quoted / qualified / schema-qualified # names: normalize the SQL so ANY spelling of an insert into # the table counts as a table insert and requires the raises @@ -710,11 +722,13 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: return False target = insert_match.group(2) # Tokenize the target's leading identifier: strip quotes - # and brackets, take the LAST dot-separated segment. + # and brackets, take the LAST dot-separated segment, + # casefolded so 'CHILDREN' cannot bypass the lowercase + # table name the scan looks for. identifier = re.split(r"[\s(]", target, maxsplit=1)[0] segments = re.split(r"\.", identifier) last = segments[-1].strip("`'\"[]") - return last == table + return last.casefold() == table.casefold() for insert, rendered in raw_statements: if not any_table.search(rendered): -- 2.55.0 From fe8ceb2f40e46c278c37d3abeff7af1cedaad101 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:03:23 -0400 Subject: [PATCH 07/11] test: sanctioned functions hold exactly their one table's probe; bare raises requires from-pytest import binding - every raw statement naming either table in a sanctioned function must be exactly that function's expected-table probe (cross-table inserts rejected); per-function count still enforces exactly one - bare Name('raises') is trusted only when the file imports it from pytest (from pytest import ... raises); other import bindings rejected --- tests/test_dao_children.py | 60 +++++++++++++++++++++++--------------- 1 file changed, 37 insertions(+), 23 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 273d247..b920d86 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -657,12 +657,21 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: """Yield (block, is_pytest_integrity) for pytest.raises blocks. Structurally exact: the call func must resolve to - pytest.raises (Attribute pytest.raises, or a bare Name - imported from pytest — checked against the file's import - section), and the matcher argument must be an AST chain - whose final segment is exactly IntegrityError (so + pytest.raises — Attribute(pytest, raises) or a bare Name + whose id is bound to pytest via the file's own import + section (validated below). The matcher argument must be an + AST chain whose final segment is exactly IntegrityError (so NotIntegrityError and fake.IntegrityError fail). """ + # Names this file imports from pytest as `raises` (e.g. + # `from pytest import raises`): a bare Name is only trusted + # when bound that way. `from x import raises` for x != pytest + # is rejected. + imported_as_raises = re.search( + r"from\s+pytest\s+import\s+.*\braises\b", + text, + re.IGNORECASE, + ) for sub in ast.walk(node): if not ( isinstance(sub, ast.With) @@ -680,7 +689,7 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: ): continue elif isinstance(func, ast.Name): - if func.id != "raises": + if func.id != "raises" or not imported_as_raises: continue else: continue @@ -730,28 +739,33 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: last = segments[-1].strip("`'\"[]") return last.casefold() == table.casefold() + # In a sanctioned function, EVERY raw statement naming either + # table must be exactly that function's one expected-table + # constraint insert — an admin_users insert inside the children + # function (or vice versa) is a bypass attempt and fails here, + # as does any second insert of the expected table. for insert, rendered in raw_statements: if not any_table.search(rendered): continue - assert allowed_stmt.match(rendered.strip()) is not None, ( - f"{node.name}: raw {table} table SQL " - f"must be INSERT or SELECT, got: {rendered[:80]!r}" + assert _is_insert_into_table(rendered, table), ( + f"{node.name}: raw statement naming the other table is " + f"forbidden; only the {table} probe is sanctioned: " + f"{rendered[:80]!r}" ) - if _is_insert_into_table(rendered, table): - # This is the sanctioned probe: require pytest.raises - # with an exact IntegrityError matcher around it. - inside = any( - block.lineno <= insert.lineno - and insert.end_lineno <= block.end_lineno - and matcher_ok - for block, matcher_ok in _raises_blocks() - ) - assert inside, ( - f"the raw {table} insert in {node.name} must be nested " - "in a pytest.raises(...IntegrityError...) block " - "(expected-to-fail constraint probe, not a write)" - ) - probes_per_function[node.name] += 1 + # This is the sanctioned probe: require pytest.raises + # with an exact IntegrityError matcher around it. + inside = any( + block.lineno <= insert.lineno + and insert.end_lineno <= block.end_lineno + and matcher_ok + for block, matcher_ok in _raises_blocks() + ) + assert inside, ( + f"the raw {table} insert in {node.name} must be nested " + "in a pytest.raises(...IntegrityError...) block " + "(expected-to-fail constraint probe, not a write)" + ) + probes_per_function[node.name] += 1 # UPDATE/DELETE naming the tables anywhere in a sanctioned # function is forbidden outright. for _, rendered in raw_statements: -- 2.55.0 From fa502730ef6831b54d638fcc4ab2821f8d0eadca Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:04:49 -0400 Subject: [PATCH 08/11] test: guard all database entry points; real ImportFrom binding validation for bare raises - probe guard now inspects execute, fetch_one, fetch_all AND execute_many calls, so bypasses through read/many paths are caught - bare Name('raises') trusted only when the file's actual ImportFrom nodes bind 'raises' from pytest; competing imports shadowing 'raises'/'pytest' from other modules fail closed --- tests/test_dao_children.py | 49 ++++++++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 13 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index b920d86..a8dfe7c 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -639,15 +639,17 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: return rendered[1:-1] return rendered - # Collect every execute(...) statement and every raises-block - # with its full AST shape. + # Collect EVERY database SQL entry point (execute, fetch_one, + # fetch_all, execute_many): a bypass through fetch_one or + # execute_many must be caught the same as execute. raw_statements: list[tuple[ast.Await, str]] = [] for sub in ast.walk(node): if ( isinstance(sub, ast.Await) and isinstance(sub.value, ast.Call) and isinstance(sub.value.func, ast.Attribute) - and sub.value.func.attr == "execute" + and sub.value.func.attr + in {"execute", "fetch_one", "fetch_all", "execute_many"} ): sql_arg = sub.value.args[0] if sub.value.args else None rendered = ast.unparse(sql_arg) if sql_arg else "" @@ -663,15 +665,36 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: AST chain whose final segment is exactly IntegrityError (so NotIntegrityError and fake.IntegrityError fail). """ - # Names this file imports from pytest as `raises` (e.g. - # `from pytest import raises`): a bare Name is only trusted - # when bound that way. `from x import raises` for x != pytest - # is rejected. - imported_as_raises = re.search( - r"from\s+pytest\s+import\s+.*\braises\b", - text, - re.IGNORECASE, - ) + # Names bound to pytest's raises: parse the FILE's actual + # ImportFrom nodes. A bare Name('raises') is trusted only + # when an alias binds it from the pytest module (with or + # without an asname), and only when no other import or + # assignment shadows it anywhere in this file. + raises_bindings: set[str] = set() + for file_node in ast.walk(tree): + if ( + isinstance(file_node, ast.ImportFrom) + and file_node.module == "pytest" + ): + for alias in file_node.names: + if alias.name == "raises": + raises_bindings.add( + alias.asname or alias.name + ) + for file_node in ast.walk(tree): + # Any competing binding of the same names fails closed. + if isinstance(file_node, ast.ImportFrom): + for alias in file_node.names: + bound = alias.asname or alias.name + if ( + bound in {"raises", "pytest"} + and file_node.module != "pytest" + ): + raise AssertionError( + f"shadowed import of {bound!r} from " + f"{file_node.module!r} defeats the " + "pytest.raises probe validation" + ) for sub in ast.walk(node): if not ( isinstance(sub, ast.With) @@ -689,7 +712,7 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: ): continue elif isinstance(func, ast.Name): - if func.id != "raises" or not imported_as_raises: + if func.id not in raises_bindings: continue else: continue -- 2.55.0 From e5b9146bab403099ca4f3d34a762ca7b5b3f348b Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:06:44 -0400 Subject: [PATCH 09/11] test: import-shadow validation covers Import nodes; plain 'import pytest' allowed, aliased spoofs rejected --- tests/test_dao_children.py | 47 +++++++++++++++++++++++++------------- 1 file changed, 31 insertions(+), 16 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index a8dfe7c..548ecfb 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -666,28 +666,26 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: NotIntegrityError and fake.IntegrityError fail). """ # Names bound to pytest's raises: parse the FILE's actual - # ImportFrom nodes. A bare Name('raises') is trusted only - # when an alias binds it from the pytest module (with or - # without an asname), and only when no other import or - # assignment shadows it anywhere in this file. + # Import nodes. A bare Name('raises') is trusted only when + # an alias binds it from the pytest module; ANY other import + # (Import or ImportFrom) binding 'raises', 'pytest', or any + # raises alias from a non-pytest module fails closed, so a + # spoofed module or shadowed alias cannot defeat the check. raises_bindings: set[str] = set() for file_node in ast.walk(tree): - if ( - isinstance(file_node, ast.ImportFrom) - and file_node.module == "pytest" - ): - for alias in file_node.names: - if alias.name == "raises": - raises_bindings.add( - alias.asname or alias.name - ) - for file_node in ast.walk(tree): - # Any competing binding of the same names fails closed. if isinstance(file_node, ast.ImportFrom): + if file_node.module == "pytest": + for alias in file_node.names: + if alias.name == "raises": + raises_bindings.add( + alias.asname or alias.name + ) + # Any ImportFrom binding the guarded names from a + # non-pytest module fails closed. for alias in file_node.names: bound = alias.asname or alias.name if ( - bound in {"raises", "pytest"} + bound in raises_bindings | {"pytest"} and file_node.module != "pytest" ): raise AssertionError( @@ -695,6 +693,23 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: f"{file_node.module!r} defeats the " "pytest.raises probe validation" ) + elif isinstance(file_node, ast.Import): + # import fake_module as pytest (or as anything the + # raises machinery trusts) fails closed. A plain + # `import pytest` (alias.name == 'pytest', no asname) + # is the genuine module binding and is allowed. + for alias in file_node.names: + bound = alias.asname or alias.name + genuine_pytest = ( + alias.name == "pytest" and alias.asname is None + ) + if bound in raises_bindings | {"pytest"} and ( + not genuine_pytest + ): + raise AssertionError( + f"import shadowing {bound!r} defeats the " + "pytest.raises probe validation" + ) for sub in ast.walk(node): if not ( isinstance(sub, ast.With) -- 2.55.0 From b67dda86556db300b4eba99ac4f558637c4ac7a3 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:08:51 -0400 Subject: [PATCH 10/11] test: two-pass import validation so binding order cannot matter pytest raises aliases are collected across the whole file FIRST, then every competing import (wherever it appears, even before the genuine binding) is rejected - closing the reversed-order spoof --- tests/test_dao_children.py | 45 ++++++++++++++++++++------------------ 1 file changed, 24 insertions(+), 21 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 548ecfb..77e7697 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -666,38 +666,41 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: NotIntegrityError and fake.IntegrityError fail). """ # Names bound to pytest's raises: parse the FILE's actual - # Import nodes. A bare Name('raises') is trusted only when - # an alias binds it from the pytest module; ANY other import - # (Import or ImportFrom) binding 'raises', 'pytest', or any - # raises alias from a non-pytest module fails closed, so a - # spoofed module or shadowed alias cannot defeat the check. - raises_bindings: set[str] = set() + # Import nodes in TWO passes so order cannot matter — pass + # one collects every binding pytest's raises gets (whatever + # alias), pass two rejects ANY competing binding of those + # names (or of 'pytest' itself) from a non-pytest import, + # wherever it appears. A spoof imported BEFORE the genuine + # binding therefore still fails. + pytest_raises_aliases: set[str] = set() + for file_node in ast.walk(tree): + if ( + isinstance(file_node, ast.ImportFrom) + and file_node.module == "pytest" + ): + for alias in file_node.names: + if alias.name == "raises": + pytest_raises_aliases.add( + alias.asname or alias.name + ) + raises_bindings: set[str] = set(pytest_raises_aliases) for file_node in ast.walk(tree): if isinstance(file_node, ast.ImportFrom): if file_node.module == "pytest": - for alias in file_node.names: - if alias.name == "raises": - raises_bindings.add( - alias.asname or alias.name - ) - # Any ImportFrom binding the guarded names from a - # non-pytest module fails closed. + continue for alias in file_node.names: bound = alias.asname or alias.name - if ( - bound in raises_bindings | {"pytest"} - and file_node.module != "pytest" - ): + if bound in raises_bindings | {"pytest"}: raise AssertionError( f"shadowed import of {bound!r} from " f"{file_node.module!r} defeats the " "pytest.raises probe validation" ) elif isinstance(file_node, ast.Import): - # import fake_module as pytest (or as anything the - # raises machinery trusts) fails closed. A plain - # `import pytest` (alias.name == 'pytest', no asname) - # is the genuine module binding and is allowed. + # import fake_module as pytest (or as any raises + # alias) fails closed. A plain `import pytest` + # (alias.name == 'pytest', no asname) is the genuine + # module binding and is allowed. for alias in file_node.names: bound = alias.asname or alias.name genuine_pytest = ( -- 2.55.0 From a8d2b1af9dc5455f37e0eebd494dc692e260c0b0 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 09:10:48 -0400 Subject: [PATCH 11/11] test: collect database calls anywhere in the expression tree The probe guard now walks Call nodes (execute/fetch_one/fetch_all/ execute_many) rather than bare Await nodes, so calls wrapped inside asyncio.gather or helper awaitables are still classified and cannot hide a raw write from the sanctioned-function scan. --- tests/test_dao_children.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py index 77e7697..e3d78b0 100644 --- a/tests/test_dao_children.py +++ b/tests/test_dao_children.py @@ -640,18 +640,19 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: return rendered # Collect EVERY database SQL entry point (execute, fetch_one, - # fetch_all, execute_many): a bypass through fetch_one or - # execute_many must be caught the same as execute. - raw_statements: list[tuple[ast.Await, str]] = [] + # fetch_all, execute_many) wherever it appears in the awaited + # expression tree: a bypass through fetch_one/execute_many OR a + # call wrapped in another awaitable (asyncio.gather, a helper + # coroutine, a Task) must be caught the same as a bare await. + raw_statements: list[tuple[ast.AST, str]] = [] for sub in ast.walk(node): if ( - isinstance(sub, ast.Await) - and isinstance(sub.value, ast.Call) - and isinstance(sub.value.func, ast.Attribute) - and sub.value.func.attr + isinstance(sub, ast.Call) + and isinstance(sub.value if False else sub.func, ast.Attribute) + and sub.func.attr in {"execute", "fetch_one", "fetch_all", "execute_many"} ): - sql_arg = sub.value.args[0] if sub.value.args else None + sql_arg = sub.args[0] if sub.args else None rendered = ast.unparse(sql_arg) if sql_arg else "" raw_statements.append((sub, _unquote(rendered))) @@ -785,7 +786,7 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: # constraint insert — an admin_users insert inside the children # function (or vice versa) is a bypass attempt and fails here, # as does any second insert of the expected table. - for insert, rendered in raw_statements: + for stmt_node, rendered in raw_statements: if not any_table.search(rendered): continue assert _is_insert_into_table(rendered, table), ( @@ -796,8 +797,8 @@ def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None: # This is the sanctioned probe: require pytest.raises # with an exact IntegrityError matcher around it. inside = any( - block.lineno <= insert.lineno - and insert.end_lineno <= block.end_lineno + block.lineno <= stmt_node.lineno + and stmt_node.end_lineno <= block.end_lineno and matcher_ok for block, matcher_ok in _raises_blocks() ) -- 2.55.0