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]