diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c38fcc6..edd89bb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -147,6 +147,12 @@ jobs: assert resolve_canonical_profile("russell_top50_leader_rotation") == "russell_top50_leader_rotation" PY + - name: Verify installed shared test imports + run: | + set -euo pipefail + uv run --no-sync python -m pytest -q -p test_support.pytest_provenance \ + --shared-import-mode installed-lock tests/test_shared_import_contract.py + - name: Verify Python dependencies run: uv pip check diff --git a/docs/shared-test-imports.md b/docs/shared-test-imports.md new file mode 100644 index 0000000..bdbd0f7 --- /dev/null +++ b/docs/shared-test-imports.md @@ -0,0 +1,58 @@ +# Shared dependency test imports + +The ordinary pytest command keeps its existing import order and test doubles. +It does not certify shared dependency provenance. Select the separate strict lane +explicitly after installing the unchanged `uv.lock` with `uv sync --frozen --extra test`. +Dependency installation is preparation; run tests without network or credential access. + +The installed-lock lane checks VCS URL, exact locked revision and version together +with every loaded shared module's actual file/spec origin, package search path, +distribution ownership and RECORD SHA-256. Editable installs and sibling shadows fail. +The existing sibling entries in pytest's `pythonpath` remain unchanged. + +```bash +uv run --no-sync python -m pytest -q -p test_support.pytest_provenance \ + --shared-import-mode installed-lock tests/test_shared_import_contract.py +``` + +The audited selection is exactly the three tests in +`tests/test_shared_import_contract.py`: safe pinned URI/profile parsers and one +explicit, restored test double. Checks run in that pytest process at session start, +after collection, before and after setup/call/teardown, and at successful completion. +Session start rejects any preloaded shared `sys.modules` entry as unverified: +current disk bytes cannot establish what an earlier module object executed, even +when its path/spec metadata agrees. No module is reloaded or evicted. Each boundary +also requires a standard Python source loader and compares its code with a fresh compile of the verified +source. Stale timestamp caches (including equal-size/mtime collisions) and +unchecked-hash bytecode cannot pass merely because the `.py` file is correct. +Later checks cover all shared modules present at each boundary, including parser implementations. No shared namespace is exempted. +A scoped double replaces one previously verified module, is counted separately, +cannot count toward real-source completion, and must restore the original identity. +An undeclared or leaked double fails the strict lane. The lane never imports broker +or cloud entry points, fetches remote artifacts or reads real credentials. + +Completion requires the exact full collection, all three calls passed, no failed or +skipped phases, final restored state and the complete required real-module inventory. +Collection-only, deselection, interrupted runs and failed checks cannot emit +`SHARED_IMPORT_PROVENANCE_COMPLETE`. Receipts contain public source revisions, +module names and declared origin classes, without absolute paths or environment dumps. +They are engineering observations, not authenticated review or deployment attestations. + +Pinned-checkout validation requires an explicit root, exact locked Git HEAD, a clean +worktree, tracked source bytes equal to the Git object, and origins inside that root. +An unexpected same-named root, dirty content, ignored/untracked source, unknown origin, +file/spec disagreement or symlink escape fails. Installed-lock mode rejects editable +sources even if revision metadata appears correct. + +Run synthetic negative controls with +`uv run --no-sync python -m pytest -q tests/test_import_provenance.py`. +They use disposable modules/metadata/Git repositories, never modify real dependencies, +and observe a child process successfully importing a sibling before the guard rejects +it with a nonzero exit. Repeat/reorder the strict selection to check cleanup. + +This is a bounded parser lane. It does not certify the ordinary whole suite or +transient modules loaded and removed between boundaries by arbitrary third-party hooks. +Existing test-specific path edits and mocks remain in place, so legacy whole-suite +source ambiguity remains characterized rather than eliminated. The test support is +excluded from the setuptools production package selection; pins, production source, +existing scanner/completion checks and coverage configuration are unchanged. diff --git a/test_support/__init__.py b/test_support/__init__.py new file mode 100644 index 0000000..297acbc --- /dev/null +++ b/test_support/__init__.py @@ -0,0 +1 @@ +"""Repository-local test support; excluded from production packages.""" diff --git a/test_support/provenance.py b/test_support/provenance.py new file mode 100644 index 0000000..7675e2e --- /dev/null +++ b/test_support/provenance.py @@ -0,0 +1,198 @@ +"""Test-only shared import checks. Receipts describe a bounded pytest process.""" +from __future__ import annotations + +import base64 +from contextlib import contextmanager +import hashlib +from importlib import metadata +from importlib.machinery import SourceFileLoader +import json +from pathlib import Path +import re +import subprocess +import sys +import sysconfig +import tomllib +from types import ModuleType +from urllib.parse import urlsplit, urlunsplit + +SPECS = { + "quant_platform_kit": ("quant-platform-kit", "QuantPlatformKit"), + "us_equity_strategies": ("us-equity-strategies", "UsEquityStrategies"), +} +REQUIRED = { + "quant_platform_kit", "quant_platform_kit.common.strategy_plugin_artifacts", + "us_equity_strategies", "us_equity_strategies.catalog", +} + + +class ProvenanceError(RuntimeError): + """Fixed reason codes only; never include a private path or metadata value.""" + + +def locked_sources(lockfile): + packages = tomllib.loads(Path(lockfile).read_text())["package"] + result = {} + for namespace, (name, repo) in SPECS.items(): + matches = [p for p in packages if p["name"] == name] + if len(matches) != 1: + raise ProvenanceError("ambiguous_lock") + package = matches[0] + parsed = urlsplit(package.get("source", {}).get("git", "")) + url = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) + if (url != f"https://github.com/QuantStrategyLab/{repo}.git" + or not re.fullmatch(r"[0-9a-f]{40}", parsed.fragment)): + raise ProvenanceError("invalid_lock") + result[namespace] = dict(name=name, url=url, revision=parsed.fragment, + version=package["version"]) + return result + + +def git(root, *args): + result = subprocess.run(["git", "-C", str(root), *args], capture_output=True) + if result.returncode: + raise ProvenanceError("checkout_unproven") + return result.stdout + + +class Audit: + def __init__(self, locked, mode, *, checkouts=None, distributions=metadata.distribution, + sites=None, source_modes=None): + if mode not in {"installed-lock", "pinned-checkout"}: + raise ProvenanceError("explicit_mode_required") + if set(locked) != set(SPECS): + raise ProvenanceError("incomplete_lock") + self.locked = locked + self.mode = mode + self.source_modes = dict.fromkeys(SPECS, mode) if source_modes is None else source_modes + if (set(self.source_modes) != set(SPECS) + or not set(self.source_modes.values()) <= {"installed-lock", "pinned-checkout"} + or (mode == "installed-lock" and set(self.source_modes.values()) != {mode})): + raise ProvenanceError("explicit_sources_required") + self.checkouts = {k: Path(v).resolve() for k, v in (checkouts or {}).items()} + if set(self.checkouts) != {k for k, v in self.source_modes.items() if v == "pinned-checkout"}: + raise ProvenanceError("explicit_checkouts_required") + self.distributions = distributions + self.sites = tuple(Path(p).resolve() for p in (sites if sites is not None else { + sysconfig.get_path("purelib"), sysconfig.get_path("platlib")})) + self.doubles = {} + self.double_observations = 0 + + def _distribution(self, namespace): + expected = self.locked[namespace] + dist = self.distributions(expected["name"]) + direct = json.loads(dist.read_text("direct_url.json") or "{}") + if (dist.version != expected["version"] or direct.get("dir_info", {}).get("editable") + or direct.get("url") != expected["url"] + or direct.get("vcs_info", {}).get("vcs") != "git" + or direct.get("vcs_info", {}).get("commit_id") != expected["revision"]): + raise ProvenanceError("installed_metadata_mismatch") + return dist + + def _checkout(self, namespace): + root = self.checkouts[namespace] + if (Path(git(root, "rev-parse", "--show-toplevel").decode().strip()).resolve() != root + or git(root, "rev-parse", "HEAD").decode().strip() + != self.locked[namespace]["revision"] + or git(root, "status", "--porcelain", "--untracked-files=all")): + raise ProvenanceError("checkout_not_clean_locked_root") + return root + + def _file(self, path, namespace, source): + resolved = path.resolve(strict=True) + if self.source_modes[namespace] == "installed-lock": + if not any(resolved.is_relative_to(root) for root in self.sites): + raise ProvenanceError("outside_distribution") + entries = [entry for entry in (source.files or ()) + if Path(source.locate_file(entry)).absolute() == path.absolute()] + if (len(entries) != 1 or entries[0].hash is None + or entries[0].hash.mode != "sha256"): + raise ProvenanceError("missing_record_hash") + digest = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()) + if digest.decode().rstrip("=") != entries[0].hash.value: + raise ProvenanceError("record_hash_mismatch") + package_root = Path(source.locate_file(namespace)).resolve() + else: + package_root = source / "src" / namespace + if not resolved.is_relative_to(package_root): + raise ProvenanceError("outside_declared_checkout") + relative = path.absolute().relative_to(source).as_posix() + # The Git object supplies the expected bytes, including for ignored files. + if git(source, "show", f"HEAD:{relative}") != path.read_bytes(): + raise ProvenanceError("checkout_content_mismatch") + if not resolved.is_relative_to(package_root): + raise ProvenanceError("symlink_escape") + return resolved + + def require_fresh_process(self, modules=None): + # Files on disk cannot prove what an earlier module object executed. + modules = sys.modules if modules is None else modules + if any(name.split(".")[0] in SPECS for name in modules): + raise ProvenanceError("preloaded_shared_modules_unverified") + + def audit(self, modules=None): + modules = dict(sys.modules if modules is None else modules) + sources = {namespace: (self._distribution(namespace) if self.source_modes[namespace] == "installed-lock" + else self._checkout(namespace)) for namespace in SPECS} + observed = set() + for name, module in modules.items(): + namespace = name.split(".")[0] + if namespace not in SPECS: + continue + if name in self.doubles and module is self.doubles[name]: + self.double_observations += 1 + continue + spec = getattr(module, "__spec__", None) + filename = getattr(module, "__file__", None) + origin = getattr(spec, "origin", None) + if (not isinstance(module, ModuleType) or module.__name__ != name + or getattr(spec, "name", None) != name or not filename or not origin): + raise ProvenanceError("unproven_module") + file = Path(filename) + if not file.is_absolute() or file.resolve() != Path(origin).resolve(): + raise ProvenanceError("origin_disagreement") + checked = self._file(file, namespace, sources[namespace]) + loader = getattr(spec, "loader", None) + if (type(loader) is not SourceFileLoader or file.suffix != ".py" + or loader.name != name or Path(loader.path).resolve() != checked): + raise ProvenanceError("unproven_source_loader") + # A valid source hash does not validate a timestamp or unchecked-hash + # .pyc. Compare the code Python would load with a fresh source compile. + expected_code = compile(file.read_bytes(), str(file), "exec", dont_inherit=True) + if loader.get_code(name) != expected_code: + raise ProvenanceError("bytecode_source_mismatch") + if hasattr(module, "__path__"): + paths = tuple(Path(p).resolve() for p in module.__path__) + locations = tuple(Path(p).resolve() for p in (spec.submodule_search_locations or ())) + if paths != (checked.parent,) or locations != paths: + raise ProvenanceError("package_search_path_mismatch") + observed.add(name) + return observed + + @contextmanager + def scoped_double(self, name, double): + # Only an exact, already verified module can be temporarily replaced. + if name not in self.audit() or name in self.doubles: + raise ProvenanceError("double_boundary_unproven") + original = sys.modules[name] + self.doubles[name] = double + sys.modules[name] = double + try: + self.audit() + yield double + finally: + leaked = sys.modules.get(name) is not double + sys.modules[name] = original + del self.doubles[name] + self.audit() + if leaked: + raise ProvenanceError("double_boundary_changed") + + def receipt(self): + observed = self.audit() + if self.doubles or not REQUIRED <= observed: + raise ProvenanceError("incomplete_real_inventory") + return {"mode": self.mode, "scope": "tests/test_shared_import_contract.py", + "origin_classes": self.source_modes, "real_modules": sorted(observed), + "revisions": {k: v["revision"] for k, v in self.locked.items()}, + "scoped_double_observations": self.double_observations} diff --git a/test_support/pytest_provenance.py b/test_support/pytest_provenance.py new file mode 100644 index 0000000..3d34724 --- /dev/null +++ b/test_support/pytest_provenance.py @@ -0,0 +1,108 @@ +"""Explicit strict lane; never auto-loaded by ordinary pytest runs.""" +import json +from pathlib import Path + +import pytest + +from .provenance import Audit, ProvenanceError, locked_sources + +MODES = ("installed-lock",) + +EXPECTED = {"test_pinned_uri_parser", "test_pinned_profile_parser", "test_scoped_double_restores"} + + +def pytest_addoption(parser): + group = parser.getgroup("shared import provenance") + group.addoption("--shared-import-mode", choices=MODES, required=True) + group.addoption("--qpk-checkout") + group.addoption("--ues-checkout") + + +def pytest_configure(config): + roots = {} + for namespace, option in (("quant_platform_kit", "qpk_checkout"), + ("us_equity_strategies", "ues_checkout")): + value = config.getoption(option) + if value: + roots[namespace] = value + config._shared_audit = Audit(locked_sources(config.rootpath / "uv.lock"), + config.getoption("shared_import_mode"), checkouts=roots) + config._shared_executed = set() + config._shared_collected = False + config._shared_failed = False + + +def checked(config): + try: + config._shared_audit.audit() + except (ProvenanceError, OSError, ValueError, ImportError) as exc: + config._shared_failed = True + reason = str(exc) if isinstance(exc, ProvenanceError) else type(exc).__name__ + raise pytest.UsageError("shared import provenance: " + reason) from None + + +def pytest_sessionstart(session): + try: + session.config._shared_audit.require_fresh_process() + except ProvenanceError: + session.config._shared_failed = True + raise pytest.UsageError("shared import provenance: preloaded_shared_modules_unverified") from None + checked(session.config) + + +def pytest_collection_finish(session): + config = session.config + expected_path = (config.rootpath / "tests/test_shared_import_contract.py").resolve() + if ({item.name for item in session.items} != EXPECTED + or len(session.items) != len(EXPECTED) + or any(Path(item.path).resolve() != expected_path for item in session.items)): + config._shared_failed = True + raise pytest.UsageError("shared import provenance: incomplete_audited_selection") + checked(config) # Same process, after test modules/conftests have been imported. + config._shared_collected = True + + +def boundary(item): + checked(item.config) + outcome = yield + checked(item.config) + if outcome.excinfo: + item.config._shared_failed = True + + +@pytest.hookimpl(hookwrapper=True, tryfirst=True) +def pytest_runtest_setup(item): + yield from boundary(item) + + +@pytest.hookimpl(hookwrapper=True, tryfirst=True) +def pytest_runtest_call(item): + yield from boundary(item) + + +@pytest.hookimpl(hookwrapper=True, tryfirst=True) +def pytest_runtest_teardown(item): + yield from boundary(item) + + +@pytest.hookimpl(hookwrapper=True) +def pytest_runtest_makereport(item, call): + report = (yield).get_result() + if report.failed or report.skipped: + item.config._shared_failed = True + if report.when == "call" and report.passed: + item.config._shared_executed.add(item.name) + + +def pytest_sessionfinish(session, exitstatus): + config = session.config + if (exitstatus or config._shared_failed or not config._shared_collected + or config._shared_executed != EXPECTED): + return + try: + receipt = config._shared_audit.receipt() + except (ProvenanceError, OSError, ValueError, ImportError): + session.exitstatus = 1 + return + reporter = config.pluginmanager.getplugin("terminalreporter") + reporter.write_line("SHARED_IMPORT_PROVENANCE_COMPLETE " + json.dumps(receipt, sort_keys=True)) diff --git a/tests/test_import_provenance.py b/tests/test_import_provenance.py new file mode 100644 index 0000000..b76478f --- /dev/null +++ b/tests/test_import_provenance.py @@ -0,0 +1,355 @@ +"""Disposable source/metadata fixtures; never mutate a real installed dependency.""" +import base64 +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import py_compile +import subprocess +import sys +from types import SimpleNamespace + +import pytest + +from test_support.provenance import Audit, ProvenanceError, SPECS, locked_sources + + +class Entry: + def __init__(self, path, content): + self.path = path + self.hash = SimpleNamespace(mode="sha256", value=base64.urlsafe_b64encode( + hashlib.sha256(content).digest()).decode().rstrip("=")) + + +class Distribution: + def __init__(self, site, namespace, expected): + self.site = site + self.version = expected["version"] + self.direct = {"url": expected["url"], "vcs_info": { + "vcs": "git", "commit_id": expected["revision"]}} + path = Path(namespace) / "__init__.py" + self.files = [Entry(path, (site / path).read_bytes())] + + def locate_file(self, entry): + return self.site / (entry.path if isinstance(entry, Entry) else entry) + + def read_text(self, name): + assert name == "direct_url.json" + return json.dumps(self.direct) + + +def module_at(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def fixture(tmp_path): + site = tmp_path / "site" + locked, distributions, modules = {}, {}, {} + for namespace, (name, repo) in SPECS.items(): + path = site / namespace / "__init__.py" + path.parent.mkdir(parents=True) + path.write_text('SOURCE = "installed"\n') + locked[namespace] = dict(name=name, url=f"https://github.com/QuantStrategyLab/{repo}.git", + revision="a" * 40, version="1.0") + distributions[name] = Distribution(site, namespace, locked[namespace]) + modules[namespace] = module_at(namespace, path) + audit = Audit(locked, "installed-lock", distributions=distributions.__getitem__, sites=[site]) + return SimpleNamespace(site=site, locked=locked, dists=distributions, modules=modules, + audit=audit, tmp=tmp_path) + + +def test_installed_metadata_and_module_origins_pass(fixture): + assert fixture.audit.audit(fixture.modules) == set(SPECS) + + +@pytest.mark.parametrize("damage", ["wrong_revision", "missing_revision", "missing_metadata", + "wrong_url", "wrong_version", "editable"]) +def test_metadata_is_required(fixture, damage): + dist = fixture.dists["quant-platform-kit"] + if damage == "wrong_revision": + dist.direct["vcs_info"]["commit_id"] = "b" * 40 + elif damage == "missing_revision": + del dist.direct["vcs_info"]["commit_id"] + elif damage == "missing_metadata": + dist.direct = {} + elif damage == "wrong_url": + dist.direct["url"] = "https://example.invalid/other.git" + elif damage == "wrong_version": + dist.version = "2.0" + else: + dist.direct["dir_info"] = {"editable": True} + with pytest.raises(ProvenanceError, match="installed_metadata_mismatch"): + fixture.audit.audit(fixture.modules) + + +def test_sibling_shadow_gap_and_rejection(fixture): + path = fixture.tmp / "sibling" / "quant_platform_kit" / "__init__.py" + path.parent.mkdir(parents=True) + path.write_text('SOURCE = "sibling"\n') + shadow = module_at("quant_platform_kit", path) + # Without a guard, Python successfully executes the wrong source despite correct dist metadata. + assert shadow.SOURCE == "sibling" + fixture.modules["quant_platform_kit"] = shadow + with pytest.raises(ProvenanceError, match="outside_distribution"): + fixture.audit.audit(fixture.modules) + + +@pytest.mark.parametrize("damage", ["file_spec", "no_spec", "namespace", "none", "search_path", + "missing_record", "changed_bytes", "symlink"]) +def test_unproven_origins_fail(fixture, damage): + module = fixture.modules["quant_platform_kit"] + if damage == "file_spec": + module.__spec__.origin = str(fixture.site / "other.py") + elif damage == "no_spec": + module.__spec__ = None + elif damage == "namespace": + module.__file__ = None + elif damage == "none": + fixture.modules["quant_platform_kit"] = None + elif damage == "search_path": + module.__path__.append(str(fixture.tmp / "sibling")) + elif damage == "missing_record": + fixture.dists["quant-platform-kit"].files = [] + elif damage == "changed_bytes": + Path(module.__file__).write_text("SOURCE = 'changed'\n") + else: + path = Path(module.__file__) + outside = fixture.tmp / "outside.py" + outside.write_bytes(path.read_bytes()) + path.unlink() # Disposable synthetic file only. + path.symlink_to(outside) + with pytest.raises((ProvenanceError, OSError)): + fixture.audit.audit(fixture.modules) + + +def test_stale_preloaded_module_cannot_evade(fixture, monkeypatch): + from types import ModuleType + for name in list(sys.modules): + if name.split(".")[0] in SPECS: + monkeypatch.delitem(sys.modules, name) + for name, module in fixture.modules.items(): + monkeypatch.setitem(sys.modules, name, module) + monkeypatch.setitem(sys.modules, "quant_platform_kit.stale", ModuleType("quant_platform_kit.stale")) + with pytest.raises(ProvenanceError, match="unproven_module"): + fixture.audit.audit() + + +def test_incomplete_inventory_cannot_complete(fixture, monkeypatch): + for name in list(sys.modules): + if name.split(".")[0] in SPECS: + monkeypatch.delitem(sys.modules, name) + for name, module in fixture.modules.items(): + monkeypatch.setitem(sys.modules, name, module) + with pytest.raises(ProvenanceError, match="incomplete_real_inventory"): + fixture.audit.receipt() + + +def test_scoped_and_leaked_doubles(fixture, monkeypatch): + from types import ModuleType + for name in list(sys.modules): + if name.split(".")[0] in SPECS: + monkeypatch.delitem(sys.modules, name) + for name, module in fixture.modules.items(): + monkeypatch.setitem(sys.modules, name, module) + name = "quant_platform_kit" + original = sys.modules[name] + double = ModuleType(name) + with fixture.audit.scoped_double(name, double): + assert name not in fixture.audit.audit() + with pytest.raises(ProvenanceError, match="incomplete_real_inventory"): + fixture.audit.receipt() + assert sys.modules[name] is original + assert name in fixture.audit.audit() + monkeypatch.setitem(sys.modules, name, double) + with pytest.raises(ProvenanceError, match="unproven_module"): + fixture.audit.audit() + + +def run_git(root, *args): + return subprocess.run(["git", "-C", str(root), *args], check=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE).stdout + + +@pytest.fixture +def checkout_fixture(fixture): + roots = {} + for namespace in SPECS: + root = fixture.tmp / namespace + path = root / "src" / namespace / "__init__.py" + path.parent.mkdir(parents=True) + path.write_text("SOURCE = 'checkout'\n") + (root / ".gitignore").write_text("__pycache__/\n") + run_git(root, "init", "-q") + run_git(root, "add", ".") + run_git(root, "-c", "user.name=Synthetic", "-c", "user.email=example@example.invalid", + "commit", "-qm", "test: freeze synthetic checkout") + fixture.locked[namespace]["revision"] = run_git(root, "rev-parse", "HEAD").decode().strip() + fixture.modules[namespace] = module_at(namespace, path) + roots[namespace] = root + fixture.roots = roots + fixture.audit = Audit(fixture.locked, "pinned-checkout", checkouts=roots) + return fixture + + +def test_exact_clean_checkout_passes_declared_mode_only(checkout_fixture): + f = checkout_fixture + assert f.audit.audit(f.modules) == set(SPECS) + installed = Audit(f.locked, "installed-lock", distributions=f.dists.__getitem__, sites=[f.site]) + with pytest.raises(ProvenanceError): + installed.audit(f.modules) + + +@pytest.mark.parametrize("damage", ["dirty", "untracked", "wrong_revision", "same_name_root", + "symlink_escape", "ignored_source"]) +def test_checkout_rejections(checkout_fixture, damage): + f = checkout_fixture + root = f.roots["quant_platform_kit"] + module = f.modules["quant_platform_kit"] + if damage == "dirty": + Path(module.__file__).write_text("SOURCE = 'dirty'\n") + elif damage == "untracked": + (root / "unexpected.py").write_text("# synthetic\n") + elif damage == "wrong_revision": + f.locked["quant_platform_kit"]["revision"] = "0" * 40 + elif damage == "same_name_root": + path = f.tmp / "unexpected" / root.name / "__init__.py" + path.parent.mkdir(parents=True) + path.write_text("SOURCE = 'unexpected'\n") + f.modules["quant_platform_kit"] = module_at("quant_platform_kit", path) + elif damage == "symlink_escape": + outside = f.tmp / "outside.py" + outside.write_bytes(Path(module.__file__).read_bytes()) + Path(module.__file__).unlink() + Path(module.__file__).symlink_to(outside) + else: + (root / "src" / "quant_platform_kit" / "__pycache__").mkdir(exist_ok=True) + path = root / "src" / "quant_platform_kit" / "__pycache__" / "hidden.py" + path.write_text("SOURCE = 'untracked ignored'\n") + f.modules["quant_platform_kit.hidden"] = module_at("quant_platform_kit.hidden", path) + with pytest.raises(ProvenanceError): + f.audit.audit(f.modules) + + +def test_no_implicit_mode_or_checkout(fixture): + with pytest.raises(ProvenanceError, match="explicit_mode"): + Audit(fixture.locked, None) + with pytest.raises(ProvenanceError, match="explicit_checkouts"): + Audit(fixture.locked, "pinned-checkout") + with pytest.raises(ProvenanceError, match="incomplete_lock"): + Audit({}, "installed-lock") + + +def test_lock_missing_or_moving_revision_fails(tmp_path): + lock = tmp_path / "uv.lock" + lock.write_text('package = []\n') + with pytest.raises(ProvenanceError, match="ambiguous_lock"): + locked_sources(lock) + content = [] + for namespace, (name, repo) in SPECS.items(): + content.append(f'[[package]]\nname = "{name}"\nversion = "1"\n' + f'source = {{git = "https://github.com/QuantStrategyLab/{repo}.git#main"}}\n') + lock.write_text("".join(content)) + with pytest.raises(ProvenanceError, match="invalid_lock"): + locked_sources(lock) + + +def test_subprocess_shadow_is_observed_failure(fixture): + # Execute the failing guard in a real child process; do not invent failed CI. + shadow = fixture.tmp / "sibling" / "quant_platform_kit" / "__init__.py" + shadow.parent.mkdir(parents=True) + shadow.write_text("SOURCE = 'sibling'\n") + helper = Path(__file__).resolve() + code = ("import runpy, sys; from pathlib import Path; " + f"h=runpy.run_path({str(helper)!r}); " + f"module=h['module_at']('quant_platform_kit', Path({str(shadow)!r})); " + "print('UNGUARDED_SIBLING_IMPORTED', flush=True); " + f"locked={fixture.locked!r}; site=Path({str(fixture.site)!r}); " + "dists={v['name']:h['Distribution'](site,k,v) for k,v in locked.items()}; " + "h['Audit'](locked,'installed-lock',distributions=dists.__getitem__,sites=[site])" + ".audit({'quant_platform_kit':module}); print('SHARED_IMPORT_PROVENANCE_COMPLETE')") + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) + assert result.returncode != 0 + assert "UNGUARDED_SIBLING_IMPORTED" in result.stdout + assert "outside_distribution" in result.stderr + assert "SHARED_IMPORT_PROVENANCE_COMPLETE" not in result.stdout + + + +@pytest.mark.parametrize("state", ["not_collected", "not_executed", "collect_only", "failed", "exit_error"]) +def test_incomplete_pytest_execution_cannot_emit_completion(state): + from test_support import pytest_provenance as plugin + lines = [] + config = SimpleNamespace(_shared_failed=state == "failed", + _shared_collected=state != "not_collected", + _shared_executed=set() if state in {"not_executed", "collect_only"} + else plugin.EXPECTED, + pluginmanager=SimpleNamespace(getplugin=lambda name: + SimpleNamespace(write_line=lines.append))) + session = SimpleNamespace(config=config, exitstatus=0) + plugin.pytest_sessionfinish(session, 1 if state == "exit_error" else 0) + assert not lines + + +def test_incomplete_pytest_collection_is_rejected(tmp_path): + from test_support import pytest_provenance as plugin + config = SimpleNamespace(rootpath=tmp_path, _shared_failed=False) + item = SimpleNamespace(name="test_pinned_uri_parser", + path=tmp_path / "tests/test_shared_import_contract.py") + with pytest.raises(pytest.UsageError, match="incomplete_audited_selection"): + plugin.pytest_collection_finish(SimpleNamespace(config=config, items=[item])) + assert config._shared_failed + + + +def test_same_path_stale_executed_content_is_unverified(fixture): + name = "quant_platform_kit" + path = fixture.site / name / "__init__.py" + expected_bytes = path.read_bytes() + path.write_text("SOURCE = 'stale-preloaded'\n") + stale = module_at(name, path) + path.write_bytes(expected_bytes) + assert stale.SOURCE == "stale-preloaded" + assert stale.__file__ == fixture.modules[name].__file__ + with pytest.raises(ProvenanceError, match="preloaded_shared_modules_unverified"): + fixture.audit.require_fresh_process({name: stale}) + fixture.audit.require_fresh_process({"unrelated_module": object()}) + + +@pytest.mark.parametrize("fixture_name", ["fixture", "checkout_fixture"]) +@pytest.mark.parametrize("namespace", SPECS) +@pytest.mark.parametrize("cache_mode", ["TIMESTAMP", "UNCHECKED_HASH"]) +def test_stale_bytecode_cannot_pass_source_audit(request, fixture_name, namespace, cache_mode): + f = request.getfixturevalue(fixture_name) + path = Path(f.modules[namespace].__file__) + expected_bytes = path.read_bytes() + # Equal size and restored mtime make timestamp caches appear current. + path.write_bytes(b"STALE = True\n".ljust(len(expected_bytes), b" ")) + stale_stat = path.stat() + cached = py_compile.compile(path, doraise=True, + invalidation_mode=getattr(py_compile.PycInvalidationMode, cache_mode)) + path.write_bytes(expected_bytes) + os.utime(path, ns=(stale_stat.st_atime_ns, stale_stat.st_mtime_ns)) + assert Path(cached).is_file() + # A new module object executes the stale cache even though source is pristine. + stale = module_at(namespace, path) + assert stale.STALE is True + assert stale.__file__ == str(path) + f.modules[namespace] = stale + with pytest.raises(ProvenanceError, match="bytecode_source_mismatch"): + f.audit.audit(f.modules) + + +@pytest.mark.parametrize("fixture_name", ["fixture", "checkout_fixture"]) +@pytest.mark.parametrize("cache_mode", list(py_compile.PycInvalidationMode)) +def test_matching_bytecode_passes_source_audit(request, fixture_name, cache_mode): + f = request.getfixturevalue(fixture_name) + for namespace, module in f.modules.items(): + path = Path(module.__file__) + py_compile.compile(path, doraise=True, invalidation_mode=cache_mode) + f.modules[namespace] = module_at(namespace, path) + assert f.audit.audit(f.modules) == set(SPECS) diff --git a/tests/test_shared_import_contract.py b/tests/test_shared_import_contract.py new file mode 100644 index 0000000..2b0fc11 --- /dev/null +++ b/tests/test_shared_import_contract.py @@ -0,0 +1,28 @@ +"""Safe real pinned parsers. No broker/cloud entry point or remote object read.""" +import sys +from types import ModuleType + +from quant_platform_kit.common.strategy_plugin_artifacts import parse_cloud_uri +from us_equity_strategies import resolve_canonical_profile + + +def test_pinned_uri_parser(): + assert parse_cloud_uri("gs://example-bucket/example.json") == ("example-bucket", "example.json") + + +def test_pinned_profile_parser(): + assert resolve_canonical_profile("russell_top50_leader_rotation") == "russell_top50_leader_rotation" + + +def test_scoped_double_restores(request): + # Ordinary tests retain their default import behavior; strict evidence requires the plugin. + audit = getattr(request.config, "_shared_audit", None) + if audit is None: + return + name = "quant_platform_kit.common.strategy_plugin_artifacts" + original = sys.modules[name] + with audit.scoped_double(name, ModuleType(name)) as double: + assert sys.modules[name] is double + assert name not in audit.audit() # A double is never certified as real source. + assert sys.modules[name] is original + assert name in audit.audit()