105 lines
4.5 KiB
Python
105 lines
4.5 KiB
Python
"""S5.2 — Varsling: load-bearing detach seams for ``notify.py``. Each test goes RED when the
|
|
guarded mechanism is removed; a bare happy-path pass would not catch the regression.
|
|
|
|
Seams pinned here:
|
|
- **MAF-free registration** — the direct-import guard (``test_okf_is_maf_free``) actually scans
|
|
``notify.py`` (otherwise the MAF-free claim is green-but-dead);
|
|
- **transitive import-graph MAF-freedom** — a TYPE_CHECKING-aware static BFS over notify's
|
|
first-party RUNTIME import edges (model: the S51 hitl probe, forbidden roots
|
|
``{agent_framework, mcp}`` ONLY — notify legitimately imports ``urllib`` for the webhook seam).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
_SRC_DIR = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser"
|
|
_MAF_ROOTS = {"agent_framework", "mcp"}
|
|
|
|
|
|
def test_notify_registered_maf_free() -> None:
|
|
"""Meta: notify.py is registered in the MAF-free guard list, so ``test_okf_is_maf_free``
|
|
actually scans it — otherwise the MAF-free claim would be green-but-dead (never checked)."""
|
|
from tests.test_okf import _MAF_FREE_MODULES
|
|
|
|
assert "notify.py" in _MAF_FREE_MODULES
|
|
|
|
|
|
def _runtime_import_nodes(tree: ast.Module) -> list[ast.Import | ast.ImportFrom]:
|
|
"""Every ``Import``/``ImportFrom`` node OUTSIDE an ``if TYPE_CHECKING:`` body. notify's
|
|
``Verdict`` import is type-only (TYPE_CHECKING) and must NOT count as a runtime edge —
|
|
the probe walks the runtime graph, exactly what D7 portability is about."""
|
|
skip: set[int] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.If):
|
|
test = node.test
|
|
if (isinstance(test, ast.Name) and test.id == "TYPE_CHECKING") or (
|
|
isinstance(test, ast.Attribute) and test.attr == "TYPE_CHECKING"
|
|
):
|
|
for child in node.body:
|
|
for sub in ast.walk(child):
|
|
skip.add(id(sub))
|
|
return [
|
|
node
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, (ast.Import, ast.ImportFrom)) and id(node) not in skip
|
|
]
|
|
|
|
|
|
def _first_party_submodule_edges(tree: ast.Module) -> set[str]:
|
|
"""``portfolio_optimiser.<name>`` RUNTIME submodule import edges — modeled on
|
|
``test_hitl_loadbearing._first_party_submodule_edges`` (same package-top/``__init__``
|
|
discount), restricted to non-TYPE_CHECKING nodes via ``_runtime_import_nodes``."""
|
|
edges: set[str] = set()
|
|
for node in _runtime_import_nodes(tree):
|
|
if isinstance(node, ast.ImportFrom):
|
|
parts = (node.module or "").split(".")
|
|
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
|
|
edges.add(parts[1])
|
|
else:
|
|
for alias in node.names:
|
|
parts = alias.name.split(".")
|
|
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
|
|
edges.add(parts[1])
|
|
return edges
|
|
|
|
|
|
def _imports_maf(tree: ast.Module) -> bool:
|
|
for node in _runtime_import_nodes(tree):
|
|
if isinstance(node, ast.Import):
|
|
if any(a.name.split(".")[0] in _MAF_ROOTS for a in node.names):
|
|
return True
|
|
elif (node.module or "").split(".")[0] in _MAF_ROOTS:
|
|
return True
|
|
return False
|
|
|
|
|
|
def test_notify_transitive_import_graph_is_maf_free() -> None:
|
|
"""LOAD-BEARING: no module reachable from ``notify.py``'s OWN first-party RUNTIME import edges
|
|
imports ``agent_framework``/``mcp``. Static AST BFS (never traversing ``__init__.py``;
|
|
TYPE_CHECKING edges discounted — the ``Verdict`` annotation is type-only by design). Detach
|
|
point (documented): promote the ``verdicts`` import to runtime (e.g. ``from
|
|
portfolio_optimiser.verdicts import verdict_to_dict``) → the walk reaches ``verdicts.py``
|
|
(``agent_framework`` at :29) → RED. A clean notify → GREEN."""
|
|
seen: set[str] = set()
|
|
queue = list(
|
|
_first_party_submodule_edges(ast.parse((_SRC_DIR / "notify.py").read_text("utf-8")))
|
|
)
|
|
maf_bearing: list[str] = []
|
|
while queue:
|
|
mod = queue.pop()
|
|
if mod in seen:
|
|
continue
|
|
seen.add(mod)
|
|
mod_file = _SRC_DIR / f"{mod}.py"
|
|
if not mod_file.is_file():
|
|
continue
|
|
tree = ast.parse(mod_file.read_text("utf-8"))
|
|
if _imports_maf(tree):
|
|
maf_bearing.append(mod)
|
|
queue.extend(_first_party_submodule_edges(tree) - seen)
|
|
assert maf_bearing == [], (
|
|
f"notify's transitive first-party import graph reaches MAF-bearing module(s): "
|
|
f"{sorted(maf_bearing)}"
|
|
)
|