From 434c1354285b327b5c8f5b1b0a850e6c4b85a881 Mon Sep 17 00:00:00 2001 From: keoyyo85 <335569976+keoyyo85@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:44:41 +0800 Subject: [PATCH] test(architecture): close the two coverage gaps review found in the owner censuses The chat action-kind census visited only ast.Assign, so a reintroduced copy annotated as LOCAL: set[str] = {...} passed the guard that exists to catch a second owner. It now reads annotated module-level assignments too, and a parametrised case proves a typed, renamed or reordered copy is recognised. The pending-reason guard checked that the import exists and that the value still compares equal. Rebinding the imported name below the import with an equal computed string keeps both checks green and still installs a second owner, and an identity assertion cannot see it either because CPython folds and interns such a string. The guard now rejects any later module-level assignment to the imported name, whatever shape that value has, with the counterexample pinned as a case. Signed-off-by: keoyyo85 <335569976+keoyyo85@users.noreply.github.com> --- .../test_chat_action_kind_owner.py | 53 +++++++++++++--- ...st_progress_review_pending_reason_owner.py | 60 +++++++++++++++++-- 2 files changed, 98 insertions(+), 15 deletions(-) diff --git a/tests/architecture/test_chat_action_kind_owner.py b/tests/architecture/test_chat_action_kind_owner.py index 8f94565eb9..c544dfff0e 100644 --- a/tests/architecture/test_chat_action_kind_owner.py +++ b/tests/architecture/test_chat_action_kind_owner.py @@ -5,6 +5,11 @@ ``loopx/chat_actions.py`` kept a byte-identical copy under a different name and gated dispatch on it, so a kind added to one side stayed invisible to the other and a proposal could be accepted by the service while the store rejected it. + +The census reads annotated module-level assignments as well as plain ones: a +``LOCAL: set[str] = {...}`` restates the same decision as a copy, so a scanner +that only visited ``ast.Assign`` would certify a fold while leaving a typed +second owner in the tree. """ from __future__ import annotations @@ -55,21 +60,36 @@ def set_literal(node: ast.AST) -> frozenset[str] | None: return frozenset(members) +def module_level_bindings(source: str) -> list[tuple[list[ast.expr], ast.expr | None]]: + """(targets, value) pairs for module-level assignments, typed or untyped.""" + bindings: list[tuple[list[ast.expr], ast.expr | None]] = [] + for statement in ast.parse(source).body: + if isinstance(statement, ast.Assign): + bindings.append((list(statement.targets), statement.value)) + elif isinstance(statement, ast.AnnAssign): + bindings.append(([statement.target], statement.value)) + return bindings + + +def vocabulary_assignments(source: str) -> list[str]: + """Names in this source that restate the owner's vocabulary as their own set.""" + held: list[str] = [] + for targets, value in module_level_bindings(source): + if value is None or set_literal(value) != KIND_VALUE: + continue + held.extend(target.id for target in targets if isinstance(target, ast.Name)) + return held + + def modules_holding_the_vocabulary() -> list[str]: offenders: list[str] = [] for path in sorted(REPO_ROOT.glob("loopx/**/*.py")): - relative = path.relative_to(REPO_ROOT).as_posix() try: - tree = ast.parse(path.read_text(encoding="utf-8")) + held = vocabulary_assignments(path.read_text(encoding="utf-8")) except (SyntaxError, UnicodeDecodeError): # pragma: no cover - guarded tree continue - for statement in tree.body: - if not isinstance(statement, ast.Assign): - continue - vocabulary = set_literal(statement.value) - if vocabulary == KIND_VALUE: - offenders.append(relative) - break + if held: + offenders.append(path.relative_to(REPO_ROOT).as_posix()) return offenders @@ -85,6 +105,21 @@ def test_no_module_outside_the_owner_holds_its_own_copy() -> None: assert modules_holding_the_vocabulary() == [OWNER_PATH] +@pytest.mark.parametrize( + "declaration", + ["SHADOW = {MEMBERS}", "SHADOW: set[str] = {MEMBERS}", "SHADOW: frozenset[str] = frozenset({MEMBERS})"], +) +def test_the_census_reads_a_reintroduced_copy_whatever_it_is_called_or_annotated( + declaration: str, +) -> None: + # Members are written in reverse so a reordered copy is proven to count too. + members = ", ".join(repr(value) for value in sorted(KIND_VALUE, reverse=True)) + + assert vocabulary_assignments(declaration.replace("{MEMBERS}", "{" + members + "}") + "\n") == [ + "SHADOW" + ] + + @pytest.mark.parametrize("action_kind", sorted(KIND_VALUE)) def test_every_declared_kind_is_dispatchable(action_kind: str) -> None: assert action_kind in chat_actions.SUPPORTED_ACTION_KINDS diff --git a/tests/architecture/test_progress_review_pending_reason_owner.py b/tests/architecture/test_progress_review_pending_reason_owner.py index 0e2089a8f2..bd2cb515f0 100644 --- a/tests/architecture/test_progress_review_pending_reason_owner.py +++ b/tests/architecture/test_progress_review_pending_reason_owner.py @@ -6,6 +6,14 @@ pending reason was the exception: each side declared ``"pending_evaluation"`` under its own name, so a rename on one side would leave the other silently comparing against a reason that no longer exists. + +A consumer must import the owner's name and then leave it alone. Checking only +that the import exists, or that the value still compares equal, is not enough: +rebinding the name below the import with a computed but equal string keeps the +import, keeps the value and still installs a second owner, and an identity check +cannot see it either because CPython folds and interns such a string. The guard +therefore reads the import line and rejects any later module-level assignment to +that name, whatever value shape it uses. """ from __future__ import annotations @@ -63,16 +71,39 @@ def modules_holding_the_reason() -> list[str]: return holders -def owner_import_levels(source: str) -> dict[str, int]: - """Local names bound by an import from the owner module, with their levels.""" +def _first_line(lines: dict[str, int], name: str, lineno: int) -> dict[str, int]: + lines[name] = min(lineno, lines.get(name, lineno)) + return lines + + +def owner_import_lines(source: str) -> dict[str, int]: + """Local names bound by an import from the owner module, with their lines.""" bindings: dict[str, int] = {} for node in ast.walk(ast.parse(source)): if isinstance(node, ast.ImportFrom) and (node.module or "").endswith(OWNER_MODULE): for alias in node.names: - bindings[alias.asname or alias.name] = node.level + bindings = _first_line(bindings, alias.asname or alias.name, node.lineno) return bindings +def module_assignment_lines(source: str) -> dict[str, int]: + """Module-level names assigned a value, whatever shape that value has.""" + assignments: dict[str, int] = {} + for statement in ast.parse(source).body: + if isinstance(statement, ast.Assign): + targets, value = list(statement.targets), statement.value + elif isinstance(statement, ast.AnnAssign): + targets, value = [statement.target], statement.value + else: + continue + if value is None: + continue + for target in targets: + if isinstance(target, ast.Name): + assignments = _first_line(assignments, target.id, statement.lineno) + return assignments + + def test_the_policy_owner_states_the_reason() -> None: assert progress_review_policy.PROGRESS_REVIEW_PENDING_REASON == REASON_VALUE @@ -80,22 +111,39 @@ def test_the_policy_owner_states_the_reason() -> None: @pytest.mark.parametrize( ("module", "local_name"), [ - (receipt, "PROGRESS_REVIEW_PENDING_REASON"), + (receipt, OWNER_NAME), (external_progress_review, "EXTERNAL_PROGRESS_REVIEW_PENDING_REASON"), ], ids=["capability-receipt", "external-review"], ) -def test_each_consumer_imports_the_owner_name_and_carries_its_value( +def test_each_consumer_imports_the_owner_name_and_never_rebinds_it( module: object, local_name: str ) -> None: source = Path(module.__file__).read_text(encoding="utf-8") # type: ignore[attr-defined] - assert owner_import_levels(source).get(local_name), ( + assert local_name in owner_import_lines(source), ( f"{local_name} must be imported from {OWNER_MODULE}, not re-declared locally" ) + assert local_name not in module_assignment_lines(source), ( + f"{local_name} is assigned again in this module; an equal local value is a" + " second owner even while the owner import stays in place" + ) assert getattr(module, local_name) == REASON_VALUE +def test_the_census_flags_a_rebind_that_keeps_the_import_and_the_value() -> None: + # The counterexample review raised: the import survives, the value still + # compares equal, and a literal-keyed census sees nothing because the copy is + # built by concatenation. + source = ( + f"from loopx.control_plane.work_items.{OWNER_MODULE} import {OWNER_NAME}\n" + f'{OWNER_NAME} = "pending_" + "evaluation"\n' + ) + + assert owner_import_lines(source) == {OWNER_NAME: 1} + assert module_assignment_lines(source) == {OWNER_NAME: 2} + + def test_no_module_outside_the_owner_holds_the_reason() -> None: assert modules_holding_the_reason() == [OWNER_PATH]