fix(drift): reject non-finite and out-of-range thresholds (fixes #21)

scripts/__main__.py

@@ -6,14 +6,14 @@
     python -m scripts drift --current graph.json --previous prev.json --out drift.json
     python -m scripts sbom (placeholder — wraps gen_sbom.sh)
 """
 from __future__ import annotations
 
 import argparse
 import json
 import sys
 import os
 import math
 import time
 
 
 def cmd_analyze(args: argparse.Namespace) -> int:
@@ -60,7 +60,7 @@
 
 def cmd_risk(args: argparse.Namespace) -> int:
     """Consolidate risk from multiple sources."""
     from risk_update import load, hotspot_sev
 
     hotspots = load(args.hotspots) if args.hotspots else None
     security = load(args.security) if args.security else None
@@ -84,15 +84,33 @@
     return 0
 
 
+def drift_threshold(value: str) -> float:
+    """Argparse type for the drift threshold: finite, 0 <= t <= 1.
+
+    Mirrors scan_drift.threshold_type (kept as a local copy because this
+    module is also loaded standalone in tests, where the sibling script may
+    not be importable). Runs during parsing, before any input is read.
+    """
+    try:
+        threshold = float(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"threshold must be a number, got {value!r}")
+    if not math.isfinite(threshold) or not 0 <= threshold <= 1:
+        raise argparse.ArgumentTypeError(
+            f"threshold must be a finite value in the range 0 to 1, got {value!r}"
+        )
+    return threshold
+
+
 def cmd_drift(args: argparse.Namespace) -> int:
     """Compare dependency graphs for architecture drift."""
     from scan_drift import load, edge_key
 
     cur = load(args.current)
     prev = load(args.previous)
 
     prev_edges = set(edge_key(e) for e in prev.get("edges", []))
     cur_edges = set(edge_key(e) for e in cur.get("edges", []))
     added = cur_edges - prev_edges
     removed = prev_edges - cur_edges
     prev_count = len(prev_edges) or 1
@@ -135,7 +153,7 @@
     p_drift = sub.add_parser("drift", help="Check architecture drift")
     p_drift.add_argument("--current", required=True, help="Current graph JSON")
     p_drift.add_argument("--previous", required=True, help="Previous graph JSON")
-    p_drift.add_argument("--threshold", type=float, default=0.1)
+    p_drift.add_argument("--threshold", type=drift_threshold, default=0.1)
     p_drift.add_argument("--out", help="Output file")
 
     # sbom

scripts/scan_drift.py

@@ -46,7 +46,7 @@
   2 if drift >= threshold (trigger pipeline action)
 
 """
-import json, argparse, sys, hashlib
+import json, argparse, sys, hashlib, math
 from collections import defaultdict
 
 def load(path):
@@ -56,11 +56,29 @@ def load(path):
 def edge_key(e):
     return (e["from"], e["to"], e.get("type",""))
 
+def threshold_type(value):
+    """Argparse type: accept only a finite threshold in the range 0 <= t <= 1.
+
+    Runs during argument parsing, so an invalid threshold is rejected before
+    any input is read or any report is written. Without this, NaN makes every
+    churn comparison false and a real drift breach is reported as no breach.
+    """
+    try:
+        threshold = float(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"threshold must be a number, got {value!r}")
+    if not math.isfinite(threshold) or not 0 <= threshold <= 1:
+        raise argparse.ArgumentTypeError(
+            f"threshold must be a finite value in the range 0 to 1, got {value!r}"
+        )
+    return threshold
+
+
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--current", required=True)
     ap.add_argument("--previous", required=True)
-    ap.add_argument("--threshold", type=float, default=0.1)
+    ap.add_argument("--threshold", type=threshold_type, default=0.1)
     ap.add_argument("--out", required=True)
     ap.add_argument("--mode", choices=["deps","services"], default="deps")
     args = ap.parse_args()

tests/test_cli_entrypoint.py

@@ -5,6 +5,8 @@
 import json
 from pathlib import Path
 
+import pytest
+
 
 def load_cli_module():
     cli_path = Path(__file__).resolve().parents[1] / "scripts" / "__main__.py"
@@ -152,6 +154,91 @@ def test_drift_command_returns_zero_when_churn_is_below_threshold(tmp_path):
     assert rc == 0
 
 
+def test_drift_command_rejects_invalid_thresholds(tmp_path):
+    cli = load_cli_module()
+    previous = write_json(
+        tmp_path / "previous.json",
+        {"edges": [{"from": "A", "to": "B", "type": "import"}]},
+    )
+    current = write_json(
+        tmp_path / "current.json",
+        {
+            "edges": [
+                {"from": "A", "to": "B", "type": "import"},
+                {"from": "A", "to": "C", "type": "import"},
+            ]
+        },
+    )
+
+    for value in ("nan", "inf", "-inf", "-0.1", "1.1"):
+        with pytest.raises(SystemExit) as exc_info:
+            cli.main(
+                [
+                    "drift",
+                    "--current",
+                    str(current),
+                    "--previous",
+                    str(previous),
+                    f"--threshold={value}",
+                ]
+            )
+        assert exc_info.value.code == 2
+
+
+def test_drift_command_threshold_boundaries(tmp_path):
+    cli = load_cli_module()
+    previous = write_json(
+        tmp_path / "previous.json",
+        {"edges": [{"from": "A", "to": "B", "type": "import"}]},
+    )
+    current = write_json(
+        tmp_path / "current.json",
+        {
+            "edges": [
+                {"from": "A", "to": "B", "type": "import"},
+                {"from": "A", "to": "C", "type": "import"},
+            ]
+        },
+    )
+
+    # Churn is 1.0, so both boundaries breach under the inclusive rule.
+    for value in ("0", "1"):
+        assert (
+            cli.main(
+                [
+                    "drift",
+                    "--current",
+                    str(current),
+                    "--previous",
+                    str(previous),
+                    "--threshold",
+                    value,
+                ]
+            )
+            == 2
+        )
+
+    # No churn at all stays below boundary threshold 1.
+    same = write_json(
+        tmp_path / "same.json",
+        {"edges": [{"from": "A", "to": "B", "type": "import"}]},
+    )
+    assert (
+        cli.main(
+            [
+                "drift",
+                "--current",
+                str(same),
+                "--previous",
+                str(previous),
+                "--threshold",
+                "1",
+            ]
+        )
+        == 0
+    )
+
+
 def test_sbom_and_missing_command_return_documented_exit_codes(capsys):
     cli = load_cli_module()
 

tests/test_scripts.py

@@ -261,6 +261,85 @@ def test_drift_detection(self):
         os.unlink(cur_path)
         os.unlink(out_path)
 
+    def _drift_graph_paths(self):
+        """Prev/cur graph pair with churn 1.0 (1 added edge over 1 previous)."""
+        prev = {
+            "nodes": [{"id": "A"}, {"id": "B"}],
+            "edges": [{"from": "A", "to": "B", "type": "import"}],
+        }
+        cur = {
+            "nodes": [{"id": "A"}, {"id": "B"}, {"id": "C"}],
+            "edges": [
+                {"from": "A", "to": "B", "type": "import"},
+                {"from": "A", "to": "C", "type": "import"},
+            ],
+        }
+        fd_prev, prev_path = tempfile.mkstemp(suffix=".json")
+        fd_cur, cur_path = tempfile.mkstemp(suffix=".json")
+        with os.fdopen(fd_prev, "w") as f:
+            json.dump(prev, f)
+        with os.fdopen(fd_cur, "w") as f:
+            json.dump(cur, f)
+        return prev_path, cur_path
+
+    def test_drift_threshold_validation_rejects_invalid(self):
+        """Invalid thresholds fail at parse time, before any report is written."""
+        import sys
+
+        from scan_drift import main as drift_main
+
+        prev_path, cur_path = self._drift_graph_paths()
+        try:
+            for value in ("nan", "inf", "-inf", "-0.1", "1.1"):
+                fd_out, out_path = tempfile.mkstemp(suffix=".json")
+                # Pre-existing report must not be overwritten by a rejected run.
+                with os.fdopen(fd_out, "w") as f:
+                    f.write("sentinel")
+                old_argv = sys.argv
+                sys.argv = ["scan_drift", "--current", cur_path, "--previous", prev_path,
+                            f"--threshold={value}", "--out", out_path]
+                try:
+                    with pytest.raises(SystemExit) as exc_info:
+                        drift_main()
+                finally:
+                    sys.argv = old_argv
+                assert exc_info.value.code == 2
+                with open(out_path) as f:
+                    assert f.read() == "sentinel"
+                os.unlink(out_path)
+        finally:
+            os.unlink(prev_path)
+            os.unlink(cur_path)
+
+    def test_drift_threshold_boundaries_remain_valid(self):
+        """Boundary thresholds 0 and 1 keep the inclusive breach rule."""
+        import sys
+
+        from scan_drift import main as drift_main
+
+        prev_path, cur_path = self._drift_graph_paths()
+        try:
+            for value in ("0", "1"):
+                fd_out, out_path = tempfile.mkstemp(suffix=".json")
+                os.close(fd_out)
+                old_argv = sys.argv
+                sys.argv = ["scan_drift", "--current", cur_path, "--previous", prev_path,
+                            "--threshold", value, "--out", out_path]
+                try:
+                    with pytest.raises(SystemExit) as exc_info:
+                        drift_main()
+                finally:
+                    sys.argv = old_argv
+                # Churn is 1.0, so both boundaries breach under >= comparison.
+                assert exc_info.value.code == 2
+                with open(out_path) as f:
+                    report = json.load(f)
+                assert report["summary"]["breach"] is True
+                os.unlink(out_path)
+        finally:
+            os.unlink(prev_path)
+            os.unlink(cur_path)
+
 
 # ─── ownership_diff tests ──────────────────────────────────────────