From de82bbc92c68e5acb97a016cb098b3f02247b86b Mon Sep 17 00:00:00 2001 From: Anthony James Padavano Date: Mon, 26 Oct 2026 23:23:35 -0400 Subject: [PATCH 1/2] fix(drift): reject non-finite and out-of-range thresholds (fixes #21) --- scripts/__main__.py | 20 ++++++++- scripts/scan_drift.py | 22 ++++++++- tests/test_cli_entrypoint.py | 87 ++++++++++++++++++++++++++++++++++++ tests/test_scripts.py | 79 ++++++++++++++++++++++++++++++++ 4 files changed, 205 insertions(+), 3 deletions(-) diff --git a/scripts/__main__.py b/scripts/__main__.py index 1295bdb..67655ab 100644 --- a/scripts/__main__.py +++ b/scripts/__main__.py @@ -84,6 +84,24 @@ def cmd_risk(args: argparse.Namespace) -> int: 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 @@ -135,7 +153,7 @@ def main(argv: list[str] | None = None) -> int: 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 diff --git a/scripts/scan_drift.py b/scripts/scan_drift.py index bd4ea22..52e1288 100644 --- a/scripts/scan_drift.py +++ b/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() diff --git a/tests/test_cli_entrypoint.py b/tests/test_cli_entrypoint.py index dfcb904..3c2f32d 100644 --- a/tests/test_cli_entrypoint.py +++ b/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() diff --git a/tests/test_scripts.py b/tests/test_scripts.py index e402a6c..94fa1e0 100644 --- a/tests/test_scripts.py +++ b/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"): + out_path = tempfile.mktemp(suffix=".json") + # Pre-existing report must not be overwritten by a rejected run. + with open(out_path, "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 ────────────────────────────────────────── From a5de7ee889c0230141ffe790232626ae65dec9e7 Mon Sep 17 00:00:00 2001 From: Anthony James Padavano Date: Mon, 26 Oct 2026 23:31:06 -0400 Subject: [PATCH 2/2] fix(tests): use mkstemp, not deprecated mktemp (CodeQL py/insecure-temporary-file on PR #22) --- tests/test_scripts.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_scripts.py b/tests/test_scripts.py index 94fa1e0..ea2e588 100644 --- a/tests/test_scripts.py +++ b/tests/test_scripts.py @@ -291,9 +291,9 @@ def test_drift_threshold_validation_rejects_invalid(self): prev_path, cur_path = self._drift_graph_paths() try: for value in ("nan", "inf", "-inf", "-0.1", "1.1"): - out_path = tempfile.mktemp(suffix=".json") + fd_out, out_path = tempfile.mkstemp(suffix=".json") # Pre-existing report must not be overwritten by a rejected run. - with open(out_path, "w") as f: + 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,