mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
chore: remove dead modules (judge, execution, calibrate, stage_processors, slot_contract)
The reachability-first audit (previous commit) found these five with no import path from any entry point: - model/judge.py: replaced by model/pipeline.py, which never imported it. - model/execution.py: the MGD Req 10 execution seam, never built or called. - model/calibrate.py: no CLI, no importer; learn/calibrate.py is the live one. - stage_processors.py + slot_contract.py: kept only so test_gate could pin them; the scan runs findings.quick_scan_findings directly. Their tests go with them, test_gate's zero-dependency list drops the two entries, the audit lists drop their rows, and the manifest is regenerated: 150 modules, 145 load_bearing, 5 standalone, 0 orphaned. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
b7f1f3d05b
commit
3091ed42ea
@@ -134,8 +134,6 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = (
|
||||
"openultrasast.ruleset.store",
|
||||
"openultrasast.improve",
|
||||
"openultrasast.improve.evolve",
|
||||
"openultrasast.slot_contract",
|
||||
"openultrasast.stage_processors",
|
||||
"openultrasast.fusion",
|
||||
"openultrasast.skills",
|
||||
"openultrasast.redaction",
|
||||
|
||||
@@ -1,97 +0,0 @@
|
||||
"""model-grounded-detection task 5.1: the corpus calibrates the MODEL, not an LLM vote (Req 9).
|
||||
|
||||
The inversion this feature rests on. Previously a pair the detector missed was a detector failure to be
|
||||
averaged away over K runs. Here a pair the model cannot split is a *named gap in the model*, with the reason
|
||||
recorded, because the model is deterministic: it will miss that pair the same way every time, and the only
|
||||
thing that changes it is better modelling.
|
||||
|
||||
`covered` therefore means the model reached a strictly higher rung on the vulnerable side than on its fix --
|
||||
not that it flagged something. Flagging both sides equally distinguishes nothing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def _outcome(name, family="injection", slice_name="vibe-py", vuln="model_entailed", fixed="none", reason=""): # type: ignore[no-untyped-def]
|
||||
from openultrasast.model.calibrate import PairOutcome
|
||||
|
||||
return PairOutcome(pair=name, family=family, slice=slice_name, vuln_rung=vuln, fixed_rung=fixed, reason=reason)
|
||||
|
||||
|
||||
def test_a_pair_the_model_splits_is_covered() -> None:
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
report = calibrate([_outcome("a")])
|
||||
assert report.covered == ("a",)
|
||||
assert report.gaps == ()
|
||||
|
||||
|
||||
def test_a_pair_the_model_cannot_split_is_a_named_gap_not_a_miss() -> None:
|
||||
"""Req 9.2: the reason travels with the gap, because a deterministic model fails the same way every time."""
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
report = calibrate([_outcome("b", vuln="none", fixed="none", reason="sink_not_in_table")])
|
||||
assert report.gaps and report.gaps[0].pair == "b"
|
||||
assert report.gaps[0].reason == "sink_not_in_table"
|
||||
assert report.covered == ()
|
||||
|
||||
|
||||
def test_flagging_both_sides_equally_is_a_gap_not_a_success() -> None:
|
||||
"""The model saw something on the fix too, so it distinguished nothing — that is the whole metric."""
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
report = calibrate([_outcome("c", vuln="model_entailed", fixed="model_entailed")])
|
||||
assert report.covered == ()
|
||||
assert report.gaps[0].reason, "an unsplit pair must carry a reason even when none was supplied"
|
||||
|
||||
|
||||
def test_gaps_are_grouped_by_family_with_their_reasons() -> None:
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
report = calibrate(
|
||||
[
|
||||
_outcome("a", family="path", vuln="none", reason="sink_not_in_table"),
|
||||
_outcome("b", family="path", vuln="none", reason="sink_not_in_table"),
|
||||
_outcome("c", family="injection", vuln="none", reason="source_unmodelled"),
|
||||
_outcome("d", family="injection"),
|
||||
]
|
||||
)
|
||||
by_family = report.gaps_by_family()
|
||||
assert set(by_family) == {"path", "injection"}
|
||||
assert by_family["path"]["pairs"] == 2
|
||||
assert by_family["path"]["reasons"] == {"sink_not_in_table": 2}
|
||||
assert by_family["injection"]["pairs"] == 1
|
||||
|
||||
|
||||
def test_the_report_is_per_slice_and_names_the_deciding_one() -> None:
|
||||
"""Req 9.3/11.1: the real-world slice decides; a development slice must not stand in for the corpus."""
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
report = calibrate(
|
||||
[
|
||||
_outcome("a", slice_name="vibe-py"),
|
||||
_outcome("b", slice_name="vfc-js", vuln="none", reason="sink_not_in_table"),
|
||||
]
|
||||
)
|
||||
assert set(report.per_slice) == {"vibe-py", "vfc-js"}
|
||||
assert report.per_slice["vibe-py"]["covered"] == 1
|
||||
assert report.per_slice["vfc-js"]["gaps"] == 1
|
||||
|
||||
|
||||
def test_the_report_refuses_a_bare_aggregate() -> None:
|
||||
from openultrasast.model.calibrate import calibrate
|
||||
|
||||
text = calibrate([_outcome("a"), _outcome("b", slice_name="vfc-js", vuln="none")]).to_markdown()
|
||||
assert "vibe-py" in text and "vfc-js" in text, "no headline without its per-slice rows"
|
||||
assert "coverage" in text.lower()
|
||||
|
||||
|
||||
def test_a_cve_recipe_yields_the_parent_and_fix_commits() -> None:
|
||||
"""Req 9.4: the harvest seam. The fix diff is the oracle; nothing is hand-labelled."""
|
||||
from openultrasast.model.calibrate import harvest_recipe
|
||||
|
||||
recipe = harvest_recipe({"repo": "org/proj", "fix_commit": "abc123", "relpath": "app.py", "function": "run"})
|
||||
assert recipe.repo == "org/proj"
|
||||
assert recipe.fix_commit == "abc123"
|
||||
assert recipe.parent_ref == "abc123^", "the vulnerable side is the fix commit's parent, by construction"
|
||||
assert recipe.relpath == "app.py" and recipe.function == "run"
|
||||
@@ -1,116 +0,0 @@
|
||||
"""model-grounded-detection task 6.1: the execution tier is deferred, adopted, and narrow (Req 10).
|
||||
|
||||
The rung this project does NOT build. Clearwing already runs sandboxes, container pools, sanitizer images and
|
||||
PoC stability, and this session's own history says reimplementing that class of machinery is where we go
|
||||
wrong. So `execution_confirmed` exists as a rung, the seam calls an adopted tool, and the eligibility rule is
|
||||
deliberately narrow: only a finding the static model could not decide, in a family a static model genuinely
|
||||
cannot arbitrate.
|
||||
|
||||
Everything here runs with Clearwing absent, which is the normal case.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _finding(family="memory", rung="suspicion"): # type: ignore[no-untyped-def]
|
||||
from openultrasast.model.execution import Candidate
|
||||
|
||||
return Candidate(finding_id="f1", family=family, rung=rung, path="a.c", line=10)
|
||||
|
||||
|
||||
def test_only_a_family_a_static_model_cannot_arbitrate_is_eligible() -> None:
|
||||
"""Req 10.3: no web/logic class depends on this tier. They have taint, dominance and constant abstraction."""
|
||||
from openultrasast.model.execution import eligible
|
||||
|
||||
assert eligible(_finding(family="memory")) is True
|
||||
for family in ("injection", "access_control", "path", "config_secrets", "output_encoding"):
|
||||
assert eligible(_finding(family=family)) is False, f"{family} must be arbitrated statically"
|
||||
|
||||
|
||||
def test_a_finding_the_model_already_decided_is_not_escalated() -> None:
|
||||
"""Req 10.1: the tier is reached only where the model could not decide -- never to double-check itself."""
|
||||
from openultrasast.model.execution import eligible
|
||||
|
||||
assert eligible(_finding(rung="suspicion")) is True
|
||||
assert eligible(_finding(rung="model_entailed")) is False
|
||||
assert eligible(_finding(rung="model_corroborated")) is False
|
||||
|
||||
|
||||
def test_without_clearwing_the_rung_is_never_reached_and_the_reason_is_recorded(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Req 11.3: an absent capability degrades to a recorded reason, never a traceback and never a verdict."""
|
||||
from openultrasast.model.execution import confirm, resolve_execution_backend
|
||||
|
||||
monkeypatch.setenv("OPENULTRASAST_CLEARWING", "off")
|
||||
backend = resolve_execution_backend()
|
||||
assert backend.available() is False
|
||||
outcome = confirm(_finding(), backend=backend)
|
||||
assert outcome.rung is None, "no engine, no rung"
|
||||
assert "clearwing_unavailable" in outcome.reason
|
||||
|
||||
|
||||
def test_an_ineligible_finding_is_not_sent_to_the_sandbox_even_when_one_exists() -> None:
|
||||
from openultrasast.model.execution import confirm
|
||||
|
||||
class _Backend:
|
||||
def __init__(self) -> None:
|
||||
self.calls = 0
|
||||
|
||||
def available(self) -> bool:
|
||||
return True
|
||||
|
||||
def reproduce(self, candidate: object) -> bool:
|
||||
self.calls += 1
|
||||
return True
|
||||
|
||||
backend = _Backend()
|
||||
outcome = confirm(_finding(family="injection"), backend=backend)
|
||||
assert outcome.rung is None and backend.calls == 0
|
||||
assert "not_eligible" in outcome.reason
|
||||
|
||||
|
||||
def test_a_reproduced_candidate_reaches_execution_confirmed() -> None:
|
||||
from openultrasast.model.execution import confirm
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
class _Backend:
|
||||
def available(self) -> bool:
|
||||
return True
|
||||
|
||||
def reproduce(self, candidate: object) -> bool:
|
||||
return True
|
||||
|
||||
assert confirm(_finding(), backend=_Backend()).rung is Rung.EXECUTION_CONFIRMED
|
||||
|
||||
|
||||
def test_a_candidate_that_does_not_reproduce_stays_where_it_was() -> None:
|
||||
"""Not reproducing is not a refutation: the sandbox failing to trigger a bug does not prove its absence."""
|
||||
from openultrasast.model.execution import confirm
|
||||
|
||||
class _Backend:
|
||||
def available(self) -> bool:
|
||||
return True
|
||||
|
||||
def reproduce(self, candidate: object) -> bool:
|
||||
return False
|
||||
|
||||
outcome = confirm(_finding(), backend=_Backend())
|
||||
assert outcome.rung is None
|
||||
assert "not_reproduced" in outcome.reason and "refut" not in outcome.reason.lower()
|
||||
|
||||
|
||||
def test_the_seam_reimplements_no_sandbox_machinery() -> None:
|
||||
"""Req 10.2, asserted structurally. The one thing this module must not grow is a sandbox."""
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
source = Path("src/openultrasast/model/execution.py").read_text()
|
||||
tree = ast.parse(source)
|
||||
imported = {n.module.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module}
|
||||
imported |= {a.name.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.Import) for a in n.names}
|
||||
for banned in ("docker", "podman", "subprocess", "container", "sandbox"):
|
||||
assert banned not in imported, f"the execution tier must adopt, not build: {banned}"
|
||||
names = {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)}
|
||||
for banned in ("build_image", "run_container", "compile_target", "apply_patch"):
|
||||
assert banned not in names, f"{banned} belongs to the adopted tool, not here"
|
||||
@@ -1,163 +0,0 @@
|
||||
"""model-grounded-detection task 3.2: the LLM proposes, the model disposes (Req 8).
|
||||
|
||||
The judge is where the prior architecture's whole premise is inverted. There is no K-run averaging, no noise
|
||||
floor and no acceptance rule; there is one bounded, typed question per candidate, and the answer is CHECKED
|
||||
against the CPG rather than voted on. The three outcomes are exhaustive:
|
||||
|
||||
the model entails it -> report it, and never call the model at all
|
||||
the model contradicts it -> drop the claim, and record the contradiction
|
||||
the model cannot decide -> report `suspicion`, honestly
|
||||
|
||||
Every test here runs with no network and a scripted client.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _spec(): # type: ignore[no-untyped-def]
|
||||
from openultrasast.model.specs import TaintSpec
|
||||
|
||||
return TaintSpec(family="injection", language="python", sources=("request.args",), sinks=("os.system",), sanitizers=("escape",))
|
||||
|
||||
|
||||
def _cpg(rows): # type: ignore[no-untyped-def]
|
||||
from openultrasast.cpg.backend import CpgResult
|
||||
|
||||
return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: rows)
|
||||
|
||||
|
||||
ENTAILING_ROW = [
|
||||
{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": False, "length": 3}
|
||||
]
|
||||
|
||||
|
||||
class _Client:
|
||||
"""A scripted chat client that records whether it was called at all."""
|
||||
|
||||
def __init__(self, answer: str = '{"vulnerable": true, "family": "injection"}') -> None:
|
||||
self.calls: list[object] = []
|
||||
self._answer = answer
|
||||
|
||||
def complete(self, **kwargs: object): # type: ignore[no-untyped-def]
|
||||
from openultrasast.tool_hunter import ChatResponse
|
||||
|
||||
self.calls.append(kwargs)
|
||||
return ChatResponse(content=self._answer, tool_calls=[], reasoning=None)
|
||||
|
||||
|
||||
def test_an_entailed_candidate_costs_no_model_call() -> None:
|
||||
"""The point of the arbiter: where the graph decides, the LLM is not consulted at all."""
|
||||
from openultrasast.model.judge import judge
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
client = _Client()
|
||||
answer = judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=client, model="m")
|
||||
assert answer.rung is Rung.ENTAILED
|
||||
assert client.calls == [], "an entailed candidate must not reach the model"
|
||||
|
||||
|
||||
def test_a_claim_the_model_contradicts_is_dropped_with_its_reason() -> None:
|
||||
"""Req 8.2. The LLM says vulnerable; the CPG shows no flow reaches the sink. The graph wins."""
|
||||
from openultrasast.model.judge import judge
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
client = _Client('{"vulnerable": true, "family": "injection"}')
|
||||
answer = judge(_cpg([]), _spec(), function="run", client=client, model="m")
|
||||
assert answer.rung is Rung.SUSPICION
|
||||
assert answer.contradiction, "a dropped claim must say why it was dropped"
|
||||
assert "no flow" in answer.contradiction.lower() or "not reach" in answer.contradiction.lower()
|
||||
|
||||
|
||||
def test_a_claim_the_model_supports_advances_to_corroborated() -> None:
|
||||
"""Req 8.3: a sanitized path is a real flow whose sufficiency only the LLM can judge."""
|
||||
from openultrasast.model.judge import judge
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}]
|
||||
client = _Client('{"vulnerable": true, "family": "injection"}')
|
||||
answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m")
|
||||
assert answer.rung is Rung.CORROBORATED
|
||||
assert len(client.calls) == 1, "exactly one bounded question, never a loop"
|
||||
|
||||
|
||||
def test_the_llm_declining_leaves_a_corroborated_flow_at_suspicion() -> None:
|
||||
"""The model saw a flow but the LLM says the sanitizer suffices; the claim does not advance."""
|
||||
from openultrasast.model.judge import judge
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}]
|
||||
client = _Client('{"vulnerable": false, "family": "injection"}')
|
||||
answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m")
|
||||
assert answer.rung is Rung.SUSPICION
|
||||
|
||||
|
||||
def test_without_a_client_the_verdict_still_stands_on_its_own() -> None:
|
||||
"""Req 11.3: the endpoint is optional. The model layer does not need it to entail."""
|
||||
from openultrasast.model.judge import judge
|
||||
from openultrasast.model.ladder import Rung
|
||||
|
||||
assert judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=None, model="").rung is Rung.ENTAILED
|
||||
sanitized = [
|
||||
{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}
|
||||
]
|
||||
answer = judge(_cpg(sanitized), _spec(), function="run", client=None, model="")
|
||||
assert answer.rung is Rung.SUSPICION and "endpoint" in answer.contradiction.lower()
|
||||
|
||||
|
||||
def test_the_question_is_bounded_typed_and_offers_no_tools() -> None:
|
||||
"""Req 8.1: the LLM answers one question about a site it was handed; it never searches for the site."""
|
||||
from openultrasast.model.judge import judge
|
||||
|
||||
rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['c']", "sanitized": True, "length": 3}]
|
||||
client = _Client()
|
||||
judge(_cpg(rows), _spec(), function="run", client=client, model="m")
|
||||
call = client.calls[0]
|
||||
assert call["tools"] == [], "no tools: the candidate is given, not searched for"
|
||||
assert call["json_object"] is True, "a typed answer, not prose"
|
||||
text = " ".join(str(m.get("content", "")) for m in call["messages"]) # type: ignore[union-attr]
|
||||
assert "os.system(cmd)" in text and "request.args['c']" in text, "the question carries the model's own evidence"
|
||||
|
||||
|
||||
def test_the_prompt_is_redacted() -> None:
|
||||
"""Req 11.4: every prompt passes through redaction."""
|
||||
from openultrasast.model.judge import judge
|
||||
|
||||
rows = [
|
||||
{
|
||||
"sink": "connect(pw)",
|
||||
"sinkLine": "4",
|
||||
"sinkMethod": "run",
|
||||
"source": "AKIA" + "IOSFODNN7" + "EXAMPLE",
|
||||
"sanitized": True,
|
||||
"length": 2,
|
||||
}
|
||||
]
|
||||
client = _Client()
|
||||
judge(_cpg(rows), _spec(), function="run", client=client, model="m")
|
||||
text = " ".join(str(m.get("content", "")) for m in client.calls[0]["messages"]) # type: ignore[index,union-attr]
|
||||
assert "AKIAIOSFODNN7EXAMPLE" not in text
|
||||
|
||||
|
||||
def test_there_is_no_k_run_averaging_anywhere_in_the_judge() -> None:
|
||||
"""Req 8.5, asserted structurally: the noise architecture must not grow back."""
|
||||
import ast
|
||||
|
||||
source = Path("src/openultrasast/model/judge.py").read_text()
|
||||
tree = ast.parse(source)
|
||||
|
||||
# Check CODE, not prose: the docstring is allowed to name the machinery it replaced.
|
||||
names = (
|
||||
{node.id for node in ast.walk(tree) if isinstance(node, ast.Name)}
|
||||
| {node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute)}
|
||||
| {node.name for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)}
|
||||
)
|
||||
for banned in ("k_runs", "majority", "vote", "votes", "average", "mean", "acceptance", "floor", "budget"):
|
||||
assert banned not in names, f"{banned} has no place in the judge"
|
||||
|
||||
# No loop over runs anywhere in the module.
|
||||
assert not [n for n in ast.walk(tree) if isinstance(n, ast.For | ast.While)], "the judge asks once"
|
||||
|
||||
imported = {n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module}
|
||||
assert not any("rounds" in m or "acceptance" in m for m in imported)
|
||||
@@ -11,7 +11,6 @@ SRC = Path("src/openultrasast")
|
||||
MANIFEST = Path("benchmarks/measurements/2026-09-08-module-audit.json")
|
||||
|
||||
|
||||
@pytest.mark.xfail(strict=True, reason="five dead modules the fixed audit found; removed in the next commit")
|
||||
def test_no_module_in_the_tree_is_orphaned() -> None:
|
||||
from openultrasast.model.audit import audit
|
||||
|
||||
@@ -37,6 +36,8 @@ def test_the_named_subsystems_each_carry_an_explicit_decision() -> None:
|
||||
assert rows[name].reason, f"{name} has no stated reason"
|
||||
# The removed agentic plane's modules are gone from the tree, not carried as rows (harnessx-removal Req 2.1).
|
||||
assert not {"harness_ext", "hunter_harness", "verify_judge"} & set(rows)
|
||||
# The five the reachability-first audit found orphaned are gone too (legacy cleanup 2026-10-02).
|
||||
assert not {"model.judge", "model.execution", "model.calibrate", "stage_processors", "slot_contract"} & set(rows)
|
||||
|
||||
|
||||
def test_the_manifest_matches_the_tree() -> None:
|
||||
@@ -47,7 +48,7 @@ def test_the_manifest_matches_the_tree() -> None:
|
||||
committed = json.loads(MANIFEST.read_text())
|
||||
current = to_dict(audit(SRC))
|
||||
assert committed["counts"] == current["counts"], "the tree drifted from its committed audit"
|
||||
assert committed["orphaned"] == current["orphaned"]
|
||||
assert committed["orphaned"] == current["orphaned"] == []
|
||||
|
||||
|
||||
def _tree(root: Path, files: dict[str, str]) -> Path:
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
"""Model-disabled, slot-contracted deterministic stages (Phase 3 task 6.3).
|
||||
|
||||
Proves: deterministic stages carry work in declared slots, a slot-contract violation
|
||||
is raised on undeclared slot access, the pipeline is model-disabled, the slot pipeline
|
||||
reproduces the deterministic quick-scan path byte-for-byte, and lifecycle hooks fire.
|
||||
"""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast.gate import _quick_scan
|
||||
from openultrasast.slot_contract import SlotContractError, SlotContractMixin, SlotPipeline
|
||||
from openultrasast.stage_processors import (
|
||||
build_deterministic_pipeline,
|
||||
deterministic_processors,
|
||||
run_quick_pipeline,
|
||||
)
|
||||
|
||||
|
||||
class _GoodStage(SlotContractMixin):
|
||||
name = "good"
|
||||
reads_slots = ("a",)
|
||||
writes_slots = ("b",)
|
||||
|
||||
def run(self, slots: Mapping[str, Any]) -> dict[str, Any]:
|
||||
return {"b": slots["a"] + 1}
|
||||
|
||||
|
||||
class _UndeclaredWriteStage(SlotContractMixin):
|
||||
name = "bad_write"
|
||||
reads_slots = ()
|
||||
writes_slots = ("declared",)
|
||||
|
||||
def run(self, slots: Mapping[str, Any]) -> dict[str, Any]:
|
||||
return {"undeclared": 1}
|
||||
|
||||
|
||||
class _MissingReadStage(SlotContractMixin):
|
||||
name = "bad_read"
|
||||
reads_slots = ("required",)
|
||||
writes_slots = ()
|
||||
|
||||
def run(self, slots: Mapping[str, Any]) -> dict[str, Any]:
|
||||
return {}
|
||||
|
||||
|
||||
# ---- slot contract enforcement (1.5) ----------------------------------------
|
||||
|
||||
|
||||
def test_declared_slots_pass() -> None:
|
||||
assert _GoodStage().execute({"a": 1}) == {"b": 2}
|
||||
|
||||
|
||||
def test_undeclared_write_is_rejected() -> None:
|
||||
with pytest.raises(SlotContractError, match="undeclared write slots: undeclared"):
|
||||
_UndeclaredWriteStage().execute({})
|
||||
|
||||
|
||||
def test_missing_read_is_rejected() -> None:
|
||||
with pytest.raises(SlotContractError, match="missing read slots: required"):
|
||||
_MissingReadStage().execute({})
|
||||
|
||||
|
||||
# ---- model-disabled invariant (1.6) -----------------------------------------
|
||||
|
||||
|
||||
def test_all_deterministic_stages_are_model_disabled() -> None:
|
||||
assert all(proc.skip_model for proc in deterministic_processors())
|
||||
|
||||
|
||||
def test_pipeline_rejects_a_model_enabled_stage() -> None:
|
||||
enabled = _GoodStage()
|
||||
enabled.skip_model = False
|
||||
with pytest.raises(SlotContractError, match="model-disabled"):
|
||||
SlotPipeline([enabled])
|
||||
|
||||
|
||||
# ---- parity with the deterministic quick-scan path --------------------------
|
||||
|
||||
|
||||
def test_pipeline_findings_match_quick_scan_baseline() -> None:
|
||||
target = Path("benchmarks/fixtures/python-vulnerable")
|
||||
slots = run_quick_pipeline(target)
|
||||
assert slots["reported_findings"] == _quick_scan(target)
|
||||
assert "score" in slots # the score slot is produced too
|
||||
|
||||
|
||||
# ---- lifecycle hooks fire per stage -----------------------------------------
|
||||
|
||||
|
||||
class _Recorder:
|
||||
def __init__(self) -> None:
|
||||
self.events: list[tuple[str, str | None]] = []
|
||||
|
||||
def emit(self, event_type: Any, stage: str | None, payload: dict[str, Any]) -> None:
|
||||
self.events.append((event_type, stage))
|
||||
|
||||
|
||||
def test_pipeline_emits_lifecycle_hooks_per_stage() -> None:
|
||||
recorder = _Recorder()
|
||||
run_quick_pipeline(Path("benchmarks/fixtures/python-vulnerable"), runtime=recorder)
|
||||
names = [proc.name for proc in build_deterministic_pipeline().processors]
|
||||
for name in names:
|
||||
assert ("stage_start", name) in recorder.events
|
||||
assert ("stage_end", name) in recorder.events
|
||||
Reference in New Issue
Block a user