From 7025582e5ba99781197379415abbc9a8001a78a0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 31 Oct 2026 20:44:17 +0800 Subject: [PATCH] test: reproduce normal-wheel runtime dependency gap --- .github/workflows/ci.yml | 56 +++++++ scripts/check_normal_wheel_install.py | 226 ++++++++++++++++++++++++++ tests/test_normal_wheel_install.py | 154 ++++++++++++++++++ 3 files changed, 436 insertions(+) create mode 100644 scripts/check_normal_wheel_install.py create mode 100644 tests/test_normal_wheel_install.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 79b6b60..815fb9d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,3 +98,59 @@ jobs: run: | set -euo pipefail python -m build + + normal-wheel-baseline: + name: Normal wheel baseline (Python ${{ matrix.python-version }}) + runs-on: ubuntu-latest + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12"] + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Setup Python + uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + + - name: Build the normal wheel + run: | + set -euo pipefail + python -m pip install build + python -m build --wheel --outdir dist + sha256sum dist/*.whl + + - name: Install only the wheel and its declared dependencies + run: | + set -euo pipefail + python -m venv "${RUNNER_TEMP}/crypto-normal-wheel" + wheels=(dist/*.whl) + test "${#wheels[@]}" -eq 1 + test -f "${wheels[0]}" + "${RUNNER_TEMP}/crypto-normal-wheel/bin/python" -I -m pip install \ + --no-cache-dir --report "${RUNNER_TEMP}/normal-wheel-install.json" \ + "${wheels[0]}" + "${RUNNER_TEMP}/crypto-normal-wheel/bin/python" -I -m pip check + + - name: Reproduce the missing runtime dependency + run: | + set -euo pipefail + qpk_commit="$(python - <<'PY' + import re + import tomllib + from pathlib import Path + + dependencies = tomllib.loads(Path("pyproject.toml").read_text())["project"]["dependencies"] + prefix = "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@" + matches = [item.removeprefix(prefix) for item in dependencies if item.startswith(prefix)] + assert len(matches) == 1 and re.fullmatch(r"[0-9a-f]{40}", matches[0]), matches + print(matches[0]) + PY + )" + cd "${RUNNER_TEMP}" + "${RUNNER_TEMP}/crypto-normal-wheel/bin/python" -I \ + "${GITHUB_WORKSPACE}/scripts/check_normal_wheel_install.py" \ + --expected-qpk-commit "${qpk_commit}" --expect-missing pandas diff --git a/scripts/check_normal_wheel_install.py b/scripts/check_normal_wheel_install.py new file mode 100644 index 0000000..c4e926e --- /dev/null +++ b/scripts/check_normal_wheel_install.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""Verify a normal Crypto wheel installation without evaluating a strategy.""" + +from __future__ import annotations + +import argparse +import importlib +import importlib.metadata +import importlib.util +import json +from pathlib import Path +import re +import site +import sys +import sysconfig +from urllib.parse import urlsplit + +_NETWORK_EVENTS = frozenset({"socket.connect", "socket.bind", "socket.getaddrinfo"}) +_NUMERIC_MODULES = ("numpy", "pandas") +_REQUIRED_MODULES = ("crypto_strategies", "quant_platform_kit") +_QPK_REPOSITORY = ("github.com", "/QuantStrategyLab/QuantPlatformKit.git") + + +class VerificationError(RuntimeError): + """An installation or provenance requirement was not met.""" + + +def _within(path: Path, roots: tuple[Path, ...]) -> bool: + return any(path.is_relative_to(root) for root in roots) + + +def _library_roots() -> tuple[Path, ...]: + prefix = Path(sys.prefix).resolve() + if sys.prefix == sys.base_prefix or not sys.flags.isolated or site.ENABLE_USER_SITE: + raise VerificationError("an isolated virtualenv interpreter is required; invoke python -I") + config = prefix / "pyvenv.cfg" + if not config.is_file(): + raise VerificationError("virtualenv configuration is absent") + settings = {} + for line in config.read_text().splitlines(): + if "=" in line: + key, value = line.split("=", 1) + settings[key.strip().lower()] = value.strip().lower() + if settings.get("include-system-site-packages") != "false": + raise VerificationError("system site packages must be disabled") + paths = sysconfig.get_paths() + roots = tuple(dict.fromkeys(Path(paths[key]).resolve() for key in ("purelib", "platlib"))) + if not all(root.is_relative_to(prefix) for root in roots): + raise VerificationError("site packages must belong to the active virtualenv") + return roots + + +def _distribution(name: str, roots: tuple[Path, ...], expected_qpk_pin: str) -> dict | None: + try: + distribution = importlib.metadata.distribution(name) + except importlib.metadata.PackageNotFoundError: + return None + location = Path(distribution.locate_file("")).resolve() + if not _within(location, roots): + raise VerificationError(f"{name} distribution is outside virtualenv site packages") + raw = distribution.read_text("direct_url.json") + direct = json.loads(raw) if raw is not None else {} + if direct.get("dir_info", {}).get("editable"): + raise VerificationError(f"{name} is editable") + info = {"version": distribution.version, "location": str(location)} + if name == "crypto-strategies": + source = urlsplit(direct.get("url", "")) + if not source.path.lower().endswith(".whl") or "archive_info" not in direct: + raise VerificationError("Crypto must be installed from the built wheel") + info["wheel_filename"] = Path(source.path).name + if name == "quant-platform-kit": + source = urlsplit(direct.get("url", "")) + vcs = direct.get("vcs_info", {}) + if (source.hostname, source.path) != _QPK_REPOSITORY: + raise VerificationError("QPK distribution has an unexpected repository origin") + if vcs.get("vcs") != "git" or vcs.get("commit_id") != expected_qpk_pin: + raise VerificationError("QPK distribution does not match the expected Git pin") + info["vcs_commit"] = vcs["commit_id"] + return info + + +def _module_origin(name: str, roots: tuple[Path, ...], *, loaded: bool = False) -> str | None: + if loaded: + module = sys.modules.get(name) + origin = None if module is None else getattr(module, "__file__", None) + else: + specification = importlib.util.find_spec(name) + origin = None if specification is None else specification.origin + if origin is None: + return None + path = Path(origin).resolve() + if not _within(path, roots): + raise VerificationError(f"{name} module is outside virtualenv site packages") + return str(path) + + +def classify_import( + error: Exception | None, + numeric_presence: dict, + network_attempts: list[str], + expect_missing: str | None, +) -> tuple[dict, int]: + """Keep expected baseline evidence distinct from successful runtime import.""" + result = { + "runtime_import_succeeded": error is None, + "expected_gap_reproduced": False, + "missing_dependency": error.name if type(error) is ModuleNotFoundError else None, + } + if network_attempts: + return {**result, "status": "network_attempt_detected"}, 1 + all_present = all( + numeric_presence[name]["distribution_present"] and numeric_presence[name]["module_present"] + for name in _NUMERIC_MODULES + ) + both_absent = all( + not numeric_presence[name]["distribution_present"] and not numeric_presence[name]["module_present"] + for name in _NUMERIC_MODULES + ) + if expect_missing is not None: + if expect_missing != "pandas": + return {**result, "status": "invalid_baseline_expectation"}, 1 + if both_absent and type(error) is ModuleNotFoundError and error.name == "pandas": + return {**result, "status": "expected_gap_reproduced", "expected_gap_reproduced": True}, 0 + return {**result, "status": "expected_gap_not_reproduced"}, 1 + if error is not None: + return {**result, "status": "runtime_import_failed", "error_type": type(error).__name__}, 1 + if not all_present: + return {**result, "status": "numeric_distribution_metadata_missing"}, 1 + return {**result, "status": "runtime_import_succeeded"}, 0 + + +def verify(expected_qpk_pin: str, expect_missing: str | None = None) -> tuple[dict, int]: + report = { + "schema_version": "qsl.normal_wheel_import.v1", + "expected_qpk_pin": expected_qpk_pin, + "runtime_import_succeeded": False, + "expected_gap_reproduced": False, + "missing_dependency": None, + "network_attempts": [], + } + try: + if re.fullmatch(r"[0-9a-f]{40}", expected_qpk_pin) is None: + raise VerificationError("expected QPK pin must be a full lowercase Git SHA") + roots = _library_roots() + report["site_packages"] = [str(root) for root in roots] + if any(name in sys.modules for name in (*_REQUIRED_MODULES, *_NUMERIC_MODULES)): + raise VerificationError("target modules must not be preloaded") + + def network_guard(event, _arguments): + if event in _NETWORK_EVENTS: + report["network_attempts"].append(event) + raise VerificationError("network access attempted during the import probe") + + sys.addaudithook(network_guard) + report["distributions"] = {} + report["module_origins_before_import"] = {} + for distribution_name, module_name in ( + ("crypto-strategies", "crypto_strategies"), + ("quant-platform-kit", "quant_platform_kit"), + ): + info = _distribution(distribution_name, roots, expected_qpk_pin) + if info is None: + raise VerificationError(f"{distribution_name} is not installed") + report["distributions"][distribution_name] = info + origin = _module_origin(module_name, roots) + if origin is None: + raise VerificationError(f"{module_name} has no installed module") + report["module_origins_before_import"][module_name] = origin + + numeric = {} + for name in _NUMERIC_MODULES: + info = _distribution(name, roots, expected_qpk_pin) + origin = _module_origin(name, roots) + numeric[name] = { + "distribution_present": info is not None, + "module_present": origin is not None, + "version": None if info is None else info["version"], + "module_origin": origin, + } + report["numeric_presence_before_import"] = numeric + + error = None + try: + importlib.import_module("crypto_strategies.entrypoints") + except Exception as exc: + error = exc + report["loaded_module_origins"] = {} + for name in (*_REQUIRED_MODULES, *_NUMERIC_MODULES, "crypto_strategies.entrypoints"): + if name in sys.modules: + origin = _module_origin(name, roots, loaded=True) + if origin is None: + raise VerificationError(f"{name} has no loaded file origin") + report["loaded_module_origins"][name] = origin + if error is None: + required = {*_REQUIRED_MODULES, *_NUMERIC_MODULES, "crypto_strategies.entrypoints"} + if not required.issubset(report["loaded_module_origins"]): + raise VerificationError("successful import did not load the required installed modules") + result, code = classify_import(error, numeric, report["network_attempts"], expect_missing) + return {**report, **result}, code + except Exception as exc: + report["status"] = "verification_failed" + report["error_type"] = type(exc).__name__ + if isinstance(exc, VerificationError): + report["verification_error"] = str(exc) + return report, 1 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--expected-qpk-commit", required=True) + parser.add_argument( + "--expect-missing", choices=["pandas"], + help="Baseline measurement only: require missing Pandas with both numeric distributions absent.", + ) + parser.add_argument("--output", type=Path) + arguments = parser.parse_args(argv) + report, code = verify(arguments.expected_qpk_commit, arguments.expect_missing) + text = json.dumps(report, indent=2, sort_keys=True) + "\n" + if arguments.output is not None: + arguments.output.write_text(text) + print(text, end="") + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_normal_wheel_install.py b/tests/test_normal_wheel_install.py new file mode 100644 index 0000000..297236d --- /dev/null +++ b/tests/test_normal_wheel_install.py @@ -0,0 +1,154 @@ +"""Offline checks for provenance rejection and baseline-result classification.""" + +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +_CHECKER = Path(__file__).resolve().parents[1] / "scripts/check_normal_wheel_install.py" +_spec = importlib.util.spec_from_file_location("normal_wheel_checker", _CHECKER) +checker = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(checker) + + +def presence(*, numpy=False, pandas=False): + return { + name: {"distribution_present": installed, "module_present": installed} + for name, installed in (("numpy", numpy), ("pandas", pandas)) + } + + +class NormalWheelClassificationTests(unittest.TestCase): + def test_baseline_records_missing_import_without_claiming_runtime_success(self): + result, code = checker.classify_import( + ModuleNotFoundError("No module named 'pandas'", name="pandas"), + presence(), [], "pandas", + ) + self.assertEqual(code, 0) + self.assertEqual(result, { + "status": "expected_gap_reproduced", + "runtime_import_succeeded": False, + "expected_gap_reproduced": True, + "missing_dependency": "pandas", + }) + + def test_strict_default_rejects_the_same_missing_dependency(self): + result, code = checker.classify_import( + ModuleNotFoundError(name="pandas"), presence(), [], None, + ) + self.assertEqual(code, 1) + self.assertFalse(result["runtime_import_succeeded"]) + self.assertFalse(result["expected_gap_reproduced"]) + self.assertEqual(result["status"], "runtime_import_failed") + + def test_network_attempt_overrides_an_otherwise_expected_gap(self): + result, code = checker.classify_import( + ModuleNotFoundError(name="pandas"), presence(), ["socket.connect"], "pandas", + ) + self.assertEqual(code, 1) + self.assertEqual(result["status"], "network_attempt_detected") + self.assertFalse(result["expected_gap_reproduced"]) + + def test_other_failures_are_never_accepted_as_the_expected_gap(self): + class MissingSubclass(ModuleNotFoundError): + pass + + for error in ( + ModuleNotFoundError(name="numpy"), + ModuleNotFoundError(name="pandas._libs"), + ModuleNotFoundError(name="google"), + ModuleNotFoundError(), + ImportError("pandas"), + RuntimeError("pandas"), + MissingSubclass(name="pandas"), + ): + with self.subTest(error=type(error).__name__, missing=getattr(error, "name", None)): + result, code = checker.classify_import(error, presence(), [], "pandas") + self.assertEqual(code, 1) + self.assertFalse(result["expected_gap_reproduced"]) + + def test_baseline_rejects_any_numeric_distribution_or_module_already_present(self): + for module in ("numpy", "pandas"): + for field in ("distribution_present", "module_present"): + with self.subTest(module=module, field=field): + before = presence() + before[module][field] = True + result, code = checker.classify_import( + ModuleNotFoundError(name="pandas"), before, [], "pandas", + ) + self.assertEqual(code, 1) + self.assertFalse(result["expected_gap_reproduced"]) + + def test_strict_success_requires_distribution_and_module_metadata(self): + result, code = checker.classify_import(None, presence(numpy=True, pandas=True), [], None) + self.assertEqual(code, 0) + self.assertTrue(result["runtime_import_succeeded"]) + self.assertFalse(result["expected_gap_reproduced"]) + for module in ("numpy", "pandas"): + for field in ("distribution_present", "module_present"): + with self.subTest(module=module, field=field): + before = presence(numpy=True, pandas=True) + before[module][field] = False + result, code = checker.classify_import(None, before, [], None) + self.assertEqual(code, 1) + self.assertEqual(result["status"], "numeric_distribution_metadata_missing") + + def test_baseline_mode_rejects_an_import_that_now_succeeds(self): + result, code = checker.classify_import(None, presence(numpy=True, pandas=True), [], "pandas") + self.assertEqual(code, 1) + self.assertTrue(result["runtime_import_succeeded"]) + self.assertFalse(result["expected_gap_reproduced"]) + + +class NormalWheelProvenanceTests(unittest.TestCase): + def distribution(self, direct_url, location="/synthetic-venv/site-packages"): + return SimpleNamespace( + version="1.0.0", + locate_file=lambda _path: Path(location), + read_text=lambda _name: json.dumps(direct_url), + ) + + def test_qpk_requires_both_the_repository_and_exact_commit(self): + pin = "a" * 40 + direct = { + "url": "https://github.com/QuantStrategyLab/QuantPlatformKit.git", + "vcs_info": {"vcs": "git", "commit_id": pin}, + } + roots = (Path("/synthetic-venv/site-packages"),) + with patch.object(checker.importlib.metadata, "distribution", return_value=self.distribution(direct)): + self.assertEqual(checker._distribution("quant-platform-kit", roots, pin)["vcs_commit"], pin) + for changed in ( + {**direct, "url": "https://example.invalid/QuantPlatformKit.git"}, + {**direct, "vcs_info": {"vcs": "git", "commit_id": "b" * 40}}, + {**direct, "vcs_info": {"vcs": "git", "requested_revision": pin}}, + {**direct, "dir_info": {"editable": True}}, + ): + with self.subTest(direct_url=changed): + with patch.object(checker.importlib.metadata, "distribution", return_value=self.distribution(changed)): + with self.assertRaises(checker.VerificationError): + checker._distribution("quant-platform-kit", roots, pin) + + def test_crypto_requires_a_wheel_and_rejects_a_source_archive(self): + roots = (Path("/synthetic-venv/site-packages"),) + for url in ("file:///build/crypto_strategies-0.4.11.tar.gz", "file:///source"): + with self.subTest(url=url): + distribution = self.distribution({"url": url, "archive_info": {}}) + with patch.object(checker.importlib.metadata, "distribution", return_value=distribution): + with self.assertRaises(checker.VerificationError): + checker._distribution("crypto-strategies", roots, "a" * 40) + + def test_distribution_and_module_origins_cannot_escape_the_virtualenv(self): + roots = (Path("/synthetic-venv/site-packages"),) + distribution = self.distribution({}, location="/sibling-source") + with patch.object(checker.importlib.metadata, "distribution", return_value=distribution): + with self.assertRaises(checker.VerificationError): + checker._distribution("numpy", roots, "a" * 40) + with patch.object(checker.importlib.util, "find_spec", return_value=SimpleNamespace(origin="/sibling-source/pandas.py")): + with self.assertRaises(checker.VerificationError): + checker._module_origin("pandas", roots) + + +if __name__ == "__main__": + unittest.main()