From d284cee493142241156c02f728fec63f597cd0cd Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 3 Jul 2026 13:51:51 +0800 Subject: [PATCH] test: harden dependency pin guard Co-Authored-By: Codex --- .github/workflows/ci.yml | 1 - scripts/check_qpk_pin_consistency.py | 88 +++++++++++++++++++++++----- tests/test_dependency_pin_guard.py | 65 ++++++++++++++++++++ 3 files changed, 137 insertions(+), 17 deletions(-) create mode 100644 tests/test_dependency_pin_guard.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 602ec0f..5837d7c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -94,7 +94,6 @@ jobs: - name: Check QPK pin consistency run: python scripts/check_qpk_pin_consistency.py - continue-on-error: true - name: Run unit tests run: | diff --git a/scripts/check_qpk_pin_consistency.py b/scripts/check_qpk_pin_consistency.py index a3c90d9..cde69e7 100644 --- a/scripts/check_qpk_pin_consistency.py +++ b/scripts/check_qpk_pin_consistency.py @@ -1,31 +1,76 @@ #!/usr/bin/env python3 -"""Check that all QPK git references match the canonical QPK_PIN. +"""Check QuantStrategyLab direct git dependency pins. + +Guards two failure modes: +- QuantPlatformKit refs must match the canonical QPK_PIN on main. +- Any QuantStrategyLab git dependency appearing in multiple dependency files + must use one consistent ref across requirements/constraints/pyproject. + Usage: python scripts/check_qpk_pin_consistency.py [--fix] """ -import re, sys +from __future__ import annotations + +import re +import subprocess +import sys +import urllib.request from pathlib import Path QPK_PIN_URL = "https://raw.githubusercontent.com/QuantStrategyLab/QuantPlatformKit/main/QPK_PIN" QPK_REF_RE = re.compile(r"QuantPlatformKit\.git@([a-f0-9]{40})") +QSL_REF_RE = re.compile(r"github\.com/QuantStrategyLab/([A-Za-z0-9_.-]+)\.git@([A-Za-z0-9._/-]+)") +PINNED_FILE_GLOBS = ("**/requirements*.txt", "**/constraints*.txt", "**/pyproject.toml") -def fetch_pin() -> str: - import urllib.request - with urllib.request.urlopen(QPK_PIN_URL, timeout=10) as r: - sha = r.read().decode().strip().split()[0] - if len(sha) == 40: return sha + +def _extract_pin(raw: str) -> str: + sha = raw.strip().split()[0] + if len(sha) == 40: + return sha raise RuntimeError(f"Invalid QPK_PIN: {sha}") -def main(): + +def fetch_pin() -> str: + errors: list[str] = [] + try: + with urllib.request.urlopen(QPK_PIN_URL, timeout=10) as response: + return _extract_pin(response.read().decode()) + except Exception as exc: # pragma: no cover - fallback path depends on host certs + errors.append(f"urllib: {exc}") + + try: + output = subprocess.check_output(["curl", "-fsSL", QPK_PIN_URL], text=True, timeout=10) + return _extract_pin(output) + except Exception as exc: + errors.append(f"curl: {exc}") + + raise RuntimeError("Unable to fetch QPK_PIN: " + "; ".join(errors)) + + +def iter_pinned_files() -> list[Path]: + paths: dict[str, Path] = {} + for pattern in PINNED_FILE_GLOBS: + for path in Path.cwd().glob(pattern): + if "external" in path.parts: + continue + paths[str(path)] = path + return [paths[name] for name in sorted(paths)] + + +def main() -> int: fix = "--fix" in sys.argv target = fetch_pin() print(f"QPK_PIN: {target[:12]}...") errors = 0 - for path in sorted(Path.cwd().glob("**/requirements*.txt")) + sorted(Path.cwd().glob("**/constraints*.txt")) + sorted(Path.cwd().glob("**/pyproject.toml")): - if "external" in str(path): continue - content = path.read_text() + qsl_refs: dict[str, list[tuple[Path, int, str]]] = {} + + for path in iter_pinned_files(): + content = path.read_text(encoding="utf-8") updated = content - for m in QPK_REF_RE.finditer(content): - sha = m.group(1) + for line_no, line in enumerate(content.splitlines(), start=1): + for repo, ref in QSL_REF_RE.findall(line): + qsl_refs.setdefault(repo, []).append((path, line_no, ref)) + for match in QPK_REF_RE.finditer(content): + sha = match.group(1) if sha != target: errors += 1 print(f" ❌ {path}: QPK@{sha[:12]} (expected {target[:12]})") @@ -33,12 +78,23 @@ def main(): updated = updated.replace(f"QuantPlatformKit.git@{sha}", f"QuantPlatformKit.git@{target}") print(" → fixed") if fix and updated != content: - path.write_text(updated) + path.write_text(updated, encoding="utf-8") + + for repo, matches in sorted(qsl_refs.items()): + refs = sorted({ref for _, _, ref in matches}) + if len(refs) <= 1: + continue + errors += 1 + print(f" ❌ inconsistent QuantStrategyLab dependency pin for {repo}:") + for path, line_no, ref in matches: + print(f" {path}:{line_no}: {ref}") + if errors: - print(f"\n{errors} mismatch(es). Run with --fix to auto-fix.") + print(f"\n{errors} dependency pin mismatch(es). Run with --fix to auto-fix QPK refs only.") return 1 - print("✅ All QPK pins match.") + print("✅ QuantStrategyLab dependency pins are consistent.") return 0 + if __name__ == "__main__": raise SystemExit(main()) diff --git a/tests/test_dependency_pin_guard.py b/tests/test_dependency_pin_guard.py new file mode 100644 index 0000000..23d1079 --- /dev/null +++ b/tests/test_dependency_pin_guard.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import ModuleType + + +SCRIPT = Path("scripts/check_qpk_pin_consistency.py") +CI_WORKFLOW = Path(".github/workflows/ci.yml") + + +def _load_guard_module() -> ModuleType: + spec = importlib.util.spec_from_file_location("check_qpk_pin_consistency_for_test", SCRIPT) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_dependency_pin_guard_checks_constraints_and_all_qsl_git_refs() -> None: + script = SCRIPT.read_text(encoding="utf-8") + + assert '"**/constraints*.txt"' in script + assert "QSL_REF_RE" in script + assert "inconsistent QuantStrategyLab dependency pin" in script + + +def test_dependency_pin_guard_rejects_internal_qsl_git_ref_drift(tmp_path, monkeypatch, capsys) -> None: + module = _load_guard_module() + qpk_ref = "0" * 40 + old_strategy_ref = "1" * 40 + new_strategy_ref = "2" * 40 + (tmp_path / "requirements.txt").write_text( + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@" + qpk_ref + "\n" + "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@" + + old_strategy_ref + + "\n", + encoding="utf-8", + ) + (tmp_path / "constraints.txt").write_text( + "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@" + + new_strategy_ref + + "\n", + encoding="utf-8", + ) + + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(module, "fetch_pin", lambda: qpk_ref) + monkeypatch.setattr(sys, "argv", ["check_qpk_pin_consistency.py"]) + + assert module.main() == 1 + output = capsys.readouterr().out + assert "inconsistent QuantStrategyLab dependency pin for UsEquityStrategies" in output + + +def test_dependency_pin_guard_is_blocking_in_ci() -> None: + workflow = CI_WORKFLOW.read_text(encoding="utf-8") + step_start = workflow.index("name: Check QPK pin consistency") + next_step = workflow.find("\n - name:", step_start + 1) + step = workflow[step_start : next_step if next_step != -1 else len(workflow)] + + assert "python scripts/check_qpk_pin_consistency.py" in step + assert "continue-on-error" not in step