diff --git a/.github/scripts/check_critical_coverage.py b/.github/scripts/check_critical_coverage.py new file mode 100644 index 0000000..e9b555f --- /dev/null +++ b/.github/scripts/check_critical_coverage.py @@ -0,0 +1,92 @@ +"""Fail closed when a critical module loses measured line/branch coverage.""" + +from __future__ import annotations + +import io +import sys +from pathlib import Path + +from coverage import Coverage, CoverageData +from coverage.exceptions import CoverageException + +# These are independent floors, not an average that another module can offset. +# Keep the original three floors unchanged when extending the risk inventory. +CRITICAL_FLOORS = { + "src/pt_snap_cli/core/snapshot_import_backend.py": 90, + "src/pt_snap_cli/snapshot/simulate/simulated_caching_allocator.py": 88, + "src/pt_snap_cli/snapshot/simulate/snapshot_mutator.py": 84, + "src/pt_snap_cli/core/import_metadata_contract.py": 100, + "src/pt_snap_cli/core/import_metadata.py": 90, + "src/pt_snap_cli/core/split_service.py": 90, + "src/pt_snap_cli/core/focus_service.py": 90, + "src/pt_snap_cli/core/skill_service.py": 90, + "src/pt_snap_cli/core/snapshot_cache.py": 90, + "src/pt_snap_cli/snapshot/representation.py": 95, +} + + +class CoverageGuardError(ValueError): + """Coverage evidence is absent, invalid, or below an independent floor.""" + + +def check_coverage(root: Path) -> None: + root = root.resolve() + if not callable(getattr(CoverageData, "close", None)): + raise CoverageGuardError("Critical coverage requires coverage>=7.10.3; install .[dev]") + data_file = root / ".coverage" + if not data_file.is_file(): + raise CoverageGuardError(f"Missing coverage data: {data_file}") + data: CoverageData | None = None + try: + coverage = Coverage(data_file=str(data_file), config_file=str(root / "pyproject.toml")) + # Own the data before load(): a corrupt SQLite file can fail inside + # coverage.py's connection __enter__, before its normal cleanup runs. + data = coverage.get_data() + coverage.load() + if not data.has_arcs(): + raise CoverageGuardError("Critical coverage requires branch data, not line-only data") + measured = {Path(name).resolve() for name in data.measured_files()} + failures = [] + for relative, floor in CRITICAL_FLOORS.items(): + source = root / relative + if not source.is_file(): + failures.append(f"{relative}: missing critical source file") + continue + if source not in measured: + failures.append(f"{relative}: missing measured coverage data") + continue + if not coverage.analysis2(str(source))[1]: + failures.append(f"{relative}: no executable statements") + continue + # report() returns the unrounded line+branch percentage. Never compare + # the rounded terminal percentage, which could hide a regression. + percent = coverage.report( + morfs=[str(source)], + file=io.StringIO(), + ignore_errors=False, + skip_empty=False, + skip_covered=False, + ) + print(f"{relative}: {percent:.4f}% (required {floor}%)") + if percent < floor: + failures.append(f"{relative}: {percent:.4f}% is below {floor}%") + if failures: + raise CoverageGuardError("\n".join(failures)) + except (CoverageException, OSError) as error: + raise CoverageGuardError(f"Cannot validate critical coverage: {error}") from error + finally: + if data is not None: + data.close() + + +def main() -> int: + try: + check_coverage(Path(__file__).resolve().parents[2]) + except CoverageGuardError as error: + print(f"Critical coverage gate failed:\n{error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/AGENTS.md b/.github/workflows/AGENTS.md index 91fb4c0..9323fc9 100644 --- a/.github/workflows/AGENTS.md +++ b/.github/workflows/AGENTS.md @@ -17,13 +17,41 @@ Concurrency is keyed by GitHub ref and superseded runs are cancelled. | Compatibility matrix | Python 3.10–3.12; fixture provenance first, then `pytest -m "not slow and not nightly"` | | Full PR test tier | Python 3.13; `pytest -m "not nightly"` with branch coverage, preserving the existing slow tests | | Overall coverage | `pyproject.toml` fail-under 90 with branches enabled | -| Critical runtime coverage | Import backend 90; simulated allocator 88; snapshot mutator 84, checked independently | +| Critical runtime coverage | Ten independent line+branch floors, enforced by the fail-closed checker below | | Python 3.13 packaging | Upload coverage, build wheel, run offline clean-wheel skill/CLI acceptance including both sharded formats | | Nightly/manual dataset acceptance | Python 3.13; provenance, wheel build and audit, then `nightly` tests including the required installed-wheel CLI/API test | Million-event deep-offset pagination tests are marked `slow`; their small pagination contracts still run across the compatibility matrix. +### Independent critical-file coverage + +After full Python 3.13 coverage, `check_critical_coverage.py` enforces this complete +inventory. A high average cannot compensate for a single file below its floor. + +| Runtime module (`src/pt_snap_cli/`) | Minimum line + branch coverage | +| --- | --- | +| `core/snapshot_import_backend.py` | 90% | +| `snapshot/simulate/simulated_caching_allocator.py` | 88% | +| `snapshot/simulate/snapshot_mutator.py` | 84% | +| `core/import_metadata_contract.py` | 100% | +| `core/import_metadata.py` | 90% | +| `core/split_service.py` | 90% | +| `core/focus_service.py` | 90% | +| `core/skill_service.py` | 90% | +| `core/snapshot_cache.py` | 90% | +| `snapshot/representation.py` | 95% | + +The checker requires `.coverage`, branch data, every listed source file, measured +coverage for each exact path, and executable statements. Missing/corrupt evidence +fails; percentages are compared before display rounding. It has no selector or +floor overrides. The overall 90% gate remains separate and unchanged. CI contract +mutation tests reject an omitted/conditional checker, alternate coverage commands, +ignored failures, reordered execution, and narrowed source/omit/exclusion settings; +`tests/test_critical_coverage_gate.py` separately protects the exact inventory and +exercises real coverage-data success/failure cases. Raise floors only with actual +CI evidence; do not delete code, add exclusions, or weaken assertions to meet them. + `tests/test_ci_tiers.py` checks parsed commands, matrix rows, marker truth tables, conditions, independent file floors and gate ordering, not step display names or complete shell text. `tests/_ci_contract.py` intentionally supports only a bounded diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index e65bbe7..6350d75 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -139,10 +139,7 @@ jobs: - name: Guard critical runtime coverage if: matrix.python-version == '3.13' # Separate floors prevent strong modules from masking a weak boundary. - run: | - python -m coverage report --include='*/core/snapshot_import_backend.py' --fail-under=90 - python -m coverage report --include='*/simulate/simulated_caching_allocator.py' --fail-under=88 - python -m coverage report --include='*/simulate/snapshot_mutator.py' --fail-under=84 + run: python .github/scripts/check_critical_coverage.py - name: Upload coverage report uses: actions/upload-artifact@v4 diff --git a/pyproject.toml b/pyproject.toml index 5e777e2..ca4e0dd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,8 @@ Issues = "https://github.com/codeasier/pt-snap-cli/issues" dev = [ "pytest>=7.0.0", "pytest-cov>=4.0.0", + # The coverage gate owns/explicitly closes CoverageData, including bad SQLite input. + "coverage[toml]>=7.10.3", "hypothesis>=6.100.0,<7", "playwright>=1.40.0", "black>=23.0.0", diff --git a/tests/core/test_critical_coverage_cache.py b/tests/core/test_critical_coverage_cache.py new file mode 100644 index 0000000..3411a30 --- /dev/null +++ b/tests/core/test_critical_coverage_cache.py @@ -0,0 +1,285 @@ +"""Reject malformed cache contracts without repairing or publishing any artifacts.""" + +import copy +import errno +import json +import os +import sqlite3 +from contextlib import closing +from pathlib import Path + +import pytest + +from pt_snap_cli.core import snapshot_cache +from pt_snap_cli.core.snapshot_cache import ( + member_digest, + seal_manifest, + source_hashes, + validate_cache, +) + + +@pytest.fixture +def cache_artifact(tmp_path: Path) -> tuple[Path, dict, str]: + source = tmp_path / "source.pkl" + # Validation hashes but never deserializes this test-owned source. + source.write_bytes(b"test-owned source, deliberately not a pickle") + cache_hash = source_hashes(source)[1] + with closing(sqlite3.connect(tmp_path / "slice.db")) as conn, conn: + conn.executescript( + "CREATE TABLE trace_entry_0(id INTEGER PRIMARY KEY, action INTEGER, " + "address INTEGER, size INTEGER, stream INTEGER, allocated INTEGER, " + "active INTEGER, reserved INTEGER, callstack TEXT);" + "CREATE TABLE block_0(id INTEGER PRIMARY KEY, address INTEGER, size INTEGER, " + "requestedSize INTEGER, state INTEGER, allocEventId INTEGER, freeEventId INTEGER);" + 'CREATE TABLE dictionary("table" TEXT, "column" TEXT, "key" TEXT, "value" TEXT);' + "INSERT INTO trace_entry_0 VALUES(0,4,4096,16,0,16,16,64,'test frame');" + "INSERT INTO block_0 VALUES(0,4096,16,16,1,0,-1);" + "INSERT INTO dictionary VALUES('trace_entry_0','action','4','alloc');" + "INSERT INTO dictionary VALUES('block_0','state','1','active_allocated');" + ) + manifest = { + "schemaVersion": 1, + "status": "complete", + "sourceFile": str(source), + "cacheHash": cache_hash, + "eventsPerSlice": 1, + "devices": { + "0": { + "eventCount": 1, + "sliceCount": 1, + "readySlices": [0], + "slices": [ + { + "index": 0, + "startEventId": 0, + "endEventId": 0, + "file": "slice.db", + "ready": True, + } + ], + } + }, + } + seal_manifest(tmp_path, manifest) + return tmp_path, manifest, cache_hash + + +def _tree_state(directory: Path) -> dict[str, object]: + return { + str(path.relative_to(directory)): ( + ("symlink", str(path.readlink())) + if path.is_symlink() + else ("file", path.read_bytes()) if path.is_file() else ("directory",) + ) + for path in directory.rglob("*") + } + + +def _assert_rejected(artifact: tuple[Path, dict, str], message: str) -> None: + directory, manifest, cache_hash = artifact + (directory / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + before = _tree_state(directory) + with pytest.raises(ValueError, match=message): + validate_cache(directory, cache_hash) + assert _tree_state(directory) == before + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("schemaVersion", 2, "Unsupported snapshot manifest version"), + ("sourceFile", "", "Snapshot is incomplete"), + ("sourceFile", 7, "Snapshot is incomplete"), + ("eventsPerSlice", 0, "Invalid nonnegative integer"), + ("eventsPerSlice", 2**63, "Invalid nonnegative integer"), + ("devices", {}, "Missing snapshot devices"), + ("devices", [], "Missing snapshot devices"), + ("devices", {"01": {}}, "Invalid device ID"), + ("devices", {"0": []}, "Invalid device inventory"), + ("ptSnap", [], "Invalid snapshot extension"), + ("ptSnap", {"identity": []}, "Invalid producer identity"), + ("ptSnap", {"memberHashKind": "future-v2"}, "Unsupported member hash contract"), + ("ptSnap", {"members": []}, "Invalid legacy snapshot member inventory"), + ("ptSnap", {"members": {}}, "Invalid legacy snapshot member inventory"), + ("parserCache", {"version": 2}, "Snapshot parser data changed"), + ], +) +def test_invalid_manifest_contract_preserves_artifacts(cache_artifact, field, value, message): + cache_artifact[1][field] = value + _assert_rejected(cache_artifact, message) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("sliceCount", 2, "Incomplete snapshot slice inventory"), + ("slices", [], "Incomplete snapshot slice inventory"), + ("slices", [None], r"slices\[0\]: expected object"), + ("readySlices", [0, 0], "Snapshot slices are not ready"), + ("eventCount", True, "Invalid nonnegative integer"), + ], +) +def test_invalid_device_inventory_preserves_artifacts(cache_artifact, field, value, message): + cache_artifact[1]["devices"]["0"][field] = value + _assert_rejected(cache_artifact, message) + + +@pytest.mark.parametrize( + ("field", "value"), + [("index", 1), ("startEventId", 1), ("endEventId", 1), ("ready", 1)], +) +def test_invalid_slice_range_preserves_artifacts(cache_artifact, field, value): + cache_artifact[1]["devices"]["0"]["slices"][0][field] = value + _assert_rejected(cache_artifact, "Invalid snapshot slice range") + + +def test_duplicate_member_between_devices_preserves_artifacts(cache_artifact): + devices = cache_artifact[1]["devices"] + devices["1"] = copy.deepcopy(devices["0"]) + _assert_rejected(cache_artifact, "Duplicate snapshot member") + + +@pytest.mark.parametrize( + ("relative", "message"), + [ + (None, "Invalid snapshot member path"), + ("", "Invalid snapshot member path"), + ("slice\\db", "Invalid snapshot member path"), + ("./slice.db", "Snapshot member escapes artifact directory"), + ("missing.db", "Missing or aliased snapshot member"), + ], +) +def test_invalid_member_paths_preserve_artifacts(cache_artifact, relative, message): + cache_artifact[1]["devices"]["0"]["slices"][0]["file"] = relative + _assert_rejected(cache_artifact, message) + + +def test_absolute_member_path_is_rejected(cache_artifact): + # A drive-qualified POSIX spelling is also absolute under Windows pathlib. + absolute = (cache_artifact[0] / "outside.db").as_posix() + cache_artifact[1]["devices"]["0"]["slices"][0]["file"] = absolute + _assert_rejected(cache_artifact, "Snapshot member escapes artifact directory") + + +def test_symlink_member_is_rejected_without_touching_target(cache_artifact): + directory, manifest, _ = cache_artifact + (directory / "alias.db").symlink_to(directory / "slice.db") + manifest["devices"]["0"]["slices"][0]["file"] = "alias.db" + _assert_rejected(cache_artifact, "Missing or aliased snapshot member") + + +@pytest.mark.parametrize("suffix", ["-wal", "-shm", "-journal"]) +def test_symlink_sidecar_is_rejected_without_touching_target(cache_artifact, suffix): + directory = cache_artifact[0] + target = directory / "consumer-data" + target.write_bytes(b"must not open or recover this sidecar target") + (directory / f"slice.db{suffix}").symlink_to(target) + _assert_rejected(cache_artifact, "Snapshot has an aliased sidecar") + + +@pytest.mark.parametrize( + ("sql", "message"), + [ + ("DROP TABLE block_0", "Missing parser table: block_0"), + ( + "ALTER TABLE block_0 RENAME TO stored_blocks;" + "CREATE VIEW block_0 AS SELECT * FROM stored_blocks", + "Missing parser table: block_0", + ), + ( + "ALTER TABLE trace_entry_0 RENAME COLUMN callstack TO consumer_stack", + r"Invalid parser column: trace_entry_0\.callstack", + ), + ( + "DROP TABLE block_0;" + "CREATE TABLE block_0(id INTEGER PRIMARY KEY, address INTEGER, size TEXT, " + "requestedSize INTEGER, state INTEGER, allocEventId INTEGER, freeEventId INTEGER)", + r"Invalid parser column: block_0\.size", + ), + ( + "DROP TABLE block_0;" + "CREATE TABLE block_0(id INTEGER, address INTEGER, size INTEGER, " + "requestedSize INTEGER, state INTEGER, allocEventId INTEGER, freeEventId INTEGER)", + r"Missing parser primary key: block_0\.id", + ), + ( + "DROP TABLE block_0;" + "CREATE TABLE block_0(id INTEGER, address INTEGER, size INTEGER, " + "requestedSize INTEGER, state INTEGER, allocEventId INTEGER, freeEventId INTEGER, " + "PRIMARY KEY(id, address))", + r"Missing parser primary key: block_0\.id", + ), + ], +) +def test_invalid_parser_schema_is_not_repaired(cache_artifact, sql, message): + with closing(sqlite3.connect(cache_artifact[0] / "slice.db")) as conn, conn: + conn.executescript(sql) + _assert_rejected(cache_artifact, message) + + +def test_matching_producer_and_parser_identities_validate_read_only(cache_artifact): + directory, manifest, cache_hash = cache_artifact + manifest["parserCache"]["requestedDevice"] = 0 + manifest["ptSnap"] = { + "identity": {"requestedDevice": 0}, + "memberHashKind": "parser-v1", + "members": dict(manifest["parserCache"]["members"]), + } + (directory / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + before = _tree_state(directory) + assert validate_cache(directory, cache_hash, device=0) == manifest + assert _tree_state(directory) == before + + +@pytest.mark.parametrize("device", ["01", "0; DROP TABLE block_0"]) +def test_member_digest_rejects_invalid_device_before_database_access(cache_artifact, device): + directory = cache_artifact[0] + before = _tree_state(directory) + with pytest.raises(ValueError, match="Invalid device ID"): + member_digest(directory / "slice.db", device, 0, 0) + assert _tree_state(directory) == before + + +@pytest.mark.parametrize("replace_path", [False, True]) +def test_source_change_during_hashing_is_rejected_and_descriptor_closed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, replace_path: bool +): + source = tmp_path / "source.pkl" + source.write_bytes(b"original test-owned source") + replacement = b"replacement bytes from the concurrent producer" + original_fstat = os.fstat + original_stat = Path.stat + descriptors: list[int] = [] + + def producer_change_at_stat(descriptor: int) -> os.stat_result: + result = original_fstat(descriptor) + descriptors.append(descriptor) + if not replace_path and len(descriptors) == 1: + # Change the open file between the before/after descriptor signatures. + source.write_bytes(replacement) + return result + + def producer_replace_before_path_stat(path: Path, *args, **kwargs) -> os.stat_result: + if path == source: + # Replace after the hashing handle closes, as required on Windows. + # Both descriptor signatures match; only the final path check detects it. + new_source = tmp_path / "replacement.pkl" + new_source.write_bytes(replacement) + new_source.replace(source) + return original_stat(path, *args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(snapshot_cache.os, "fstat", producer_change_at_stat) + if replace_path: + patch.setattr(Path, "stat", producer_replace_before_path_stat) + with pytest.raises(ValueError, match="Snapshot source changed while hashing"): + source_hashes(source) + + assert len(descriptors) == 2 + assert descriptors[0] == descriptors[1] + with pytest.raises(OSError) as closed: + original_fstat(descriptors[0]) + assert closed.value.errno == errno.EBADF + assert source.read_bytes() == replacement + assert list(tmp_path.iterdir()) == [source] diff --git a/tests/core/test_critical_coverage_focus.py b/tests/core/test_critical_coverage_focus.py new file mode 100644 index 0000000..35a1bc4 --- /dev/null +++ b/tests/core/test_critical_coverage_focus.py @@ -0,0 +1,211 @@ +"""Focus compensation preserves exact bytes, configuration state, and recovery evidence.""" + +import copy +import os +import sqlite3 +from contextlib import closing +from pathlib import Path + +import pytest + +import pt_snap_cli.core.focus_service as focus_module +from pt_snap_cli.config import Config +from pt_snap_cli.core.errors import ImportExecutionError +from pt_snap_cli.core.focus_service import FocusService + + +@pytest.fixture +def focus(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("PT_SNAP_DB_PATH", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + service = FocusService(Config()) + path = service._config.project_focus_path() + path.parent.mkdir() + original = b'{ "db_path" : "previous.db", "extra" : [1,2] }\n' + path.write_bytes(original) + service._config._config = {"nested": {"value": "original"}} + return service, path, original + + +@pytest.fixture +def checkpoint_descriptors(monkeypatch): + descriptors = [] + mkstemp = focus_module.tempfile.mkstemp + + def track(*args, **kwargs): + descriptor, name = mkstemp(*args, **kwargs) + descriptors.append(descriptor) + return descriptor, name + + monkeypatch.setattr(focus_module.tempfile, "mkstemp", track) + yield descriptors + leaked = [] + for descriptor in descriptors: + try: + os.fstat(descriptor) + except OSError: + continue + os.close(descriptor) + leaked.append(descriptor) + assert not leaked, "focus checkpoint descriptors must close on every outcome" + + +def test_short_checkpoint_writes_restore_exact_focus_and_memory( + focus, checkpoint_descriptors, monkeypatch +): + service, path, original = focus + old_state = copy.deepcopy(service._config._config) + write = os.write + writes = [] + + def short_write(descriptor, content): + writes.append(len(content)) + return write(descriptor, content[:3]) + + monkeypatch.setattr(focus_module.os, "write", short_write) + with pytest.raises(RuntimeError, match="focus mutation failed"): + with service.project_focus_transaction(): + path.write_bytes(b"partial focus") + service._config._config["nested"]["value"] = "changed" + raise RuntimeError("focus mutation failed") + assert path.read_bytes() == original + assert service._config._config == old_state + assert len(writes) > 1 + assert len(checkpoint_descriptors) == 1 + assert list(path.parent.iterdir()) == [path] + + +def test_zero_checkpoint_write_fails_before_any_focus_mutation( + focus, checkpoint_descriptors, monkeypatch +): + service, path, original = focus + monkeypatch.setattr(focus_module.os, "write", lambda *args: 0) + with pytest.raises(OSError, match="recovery checkpoint"): + with service.project_focus_transaction(): + pytest.fail("a failed checkpoint must prevent publication") + assert path.read_bytes() == original + assert len(checkpoint_descriptors) == 1 + assert list(path.parent.iterdir()) == [path] + + +def test_checkpoint_permission_fallback_is_verified_and_cleaned( + focus, checkpoint_descriptors, monkeypatch +): + service, path, _ = focus + with monkeypatch.context() as patch: + patch.delattr(focus_module.os, "fchmod", raising=False) + with service.project_focus_transaction(): + path.write_bytes(b"committed focus") + assert path.read_bytes() == b"committed focus" + assert len(checkpoint_descriptors) == 1 + assert list(path.parent.iterdir()) == [path] + + +def test_late_restore_error_does_not_hide_successful_rollback(focus, monkeypatch): + service, path, original = focus + replace = os.replace + + def restore_then_raise(source, destination): + replace(source, destination) + raise OSError("late rename error") + + monkeypatch.setattr(focus_module.os, "replace", restore_then_raise) + with pytest.raises(RuntimeError, match="original focus failure"): + with service.project_focus_transaction(): + path.write_bytes(b"new focus") + raise RuntimeError("original focus failure") + assert path.read_bytes() == original + assert list(path.parent.iterdir()) == [path] + + +def test_unrestored_focus_reports_failure_and_keeps_exact_recovery(focus, monkeypatch): + service, path, original = focus + monkeypatch.setattr(service, "_restore_project_focus", lambda *args: None) + with pytest.raises(ImportExecutionError, match="rollback also failed") as caught: + with service.project_focus_transaction(): + path.write_bytes(b"new focus") + raise RuntimeError("focus failed") + backups = list(path.parent.glob("*.recovery")) + assert len(backups) == 1 + assert backups[0].read_bytes() == original + assert str(backups[0]) in str(caught.value) + assert "prior focus existed: True" in str(caught.value) + assert path.read_bytes() == b"new focus" + assert isinstance(caught.value.__cause__, RuntimeError) + + +def test_cleanup_failure_preserves_checkpoint_without_masking_commit(focus, monkeypatch): + service, path, original = focus + verify = service._verify_checkpoint + committed = False + + def fail_cleanup(checkpoint, identity): + if committed: + raise OSError("cleanup denied") + verify(checkpoint, identity) + + monkeypatch.setattr(FocusService, "_verify_checkpoint", staticmethod(fail_cleanup)) + with service.project_focus_transaction(): + path.write_bytes(b"committed focus") + committed = True + backups = list(path.parent.glob("*.recovery")) + assert len(backups) == 1 + assert backups[0].read_bytes() == original + assert path.read_bytes() == b"committed focus" + + +def test_failure_before_first_focus_write_leaves_no_focus_or_checkpoint(focus): + service, path, _ = focus + path.unlink() + with pytest.raises(RuntimeError, match="before write"): + with service.project_focus_transaction(): + raise RuntimeError("before write") + assert not path.exists() + assert list(path.parent.iterdir()) == [] + + +def test_unsafe_focus_and_substituted_checkpoint_are_not_deleted(focus): + service, path, original = focus + unsafe = path.parent / "directory" + unsafe.mkdir() + with pytest.raises(OSError, match="Unsafe project focus"): + service._safe_focus_path(unsafe) + with pytest.raises(OSError, match="checkpoint substituted"): + service._verify_checkpoint(path, (-1, -1)) + assert unsafe.is_dir() + assert path.read_bytes() == original + + +def test_invalid_focused_schema_keeps_focus_but_has_no_available_devices(focus): + service, path, _ = focus + db = path.parent.parent / "invalid.db" + with closing(sqlite3.connect(db)) as conn, conn: + conn.execute("CREATE TABLE unrelated (value INTEGER)") + service._config.write_project_focus(db) + before = path.read_bytes() + state = service.get_focus() + assert state.db_path == db + assert state.available_devices == [] + assert path.read_bytes() == before + + +def test_global_device_update_keeps_project_focus_unchanged(focus, monkeypatch): + service, path, original = focus + db = path.parent.parent / "snapshot.db" + with closing(sqlite3.connect(db)) as conn, conn: + conn.execute( + 'CREATE TABLE dictionary ("table" TEXT, "column" TEXT, "key" TEXT, "value" TEXT)' + ) + conn.execute("CREATE TABLE trace_entry_0 (id INTEGER PRIMARY KEY)") + conn.execute("CREATE TABLE trace_entry_1 (id INTEGER PRIMARY KEY)") + service.set_global_focus(db, device_id=0) + # An explicit environment focus selects the real DB while retaining the + # unrelated project file; --global still updates only the global selection. + monkeypatch.setenv("PT_SNAP_DB_PATH", str(db)) + state = service.set_device(1, scope="global") + assert state.source == "global" + assert state.device_id == 1 + assert service._config.current_device_id == 1 + assert path.read_bytes() == original diff --git a/tests/core/test_critical_coverage_metadata.py b/tests/core/test_critical_coverage_metadata.py new file mode 100644 index 0000000..39894bb --- /dev/null +++ b/tests/core/test_critical_coverage_metadata.py @@ -0,0 +1,208 @@ +"""Malformed metadata must not become a cache hit or replace valid metadata.""" + +import sqlite3 +from contextlib import closing, contextmanager +from dataclasses import asdict, replace +from types import SimpleNamespace + +import pytest + +import pt_snap_cli.core.import_metadata as metadata_module +from pt_snap_cli.core.errors import ( + DatabaseMissingError, + DatabaseSchemaError, + ImportMetadataError, +) +from pt_snap_cli.core.import_metadata import ImportMetadataService +from pt_snap_cli.core.import_metadata_contract import ( + metadata_from_mapping, + required_int, + required_str, + validate_import_metadata, +) +from pt_snap_cli.core.models import ImportMetadata + + +def metadata(**changes): + original = ImportMetadata( + metadata_module.METADATA_SCHEMA_VERSION, + metadata_module.IMPORT_FORMAT_VERSION, + "a" * 64, + 12, + "snapshot.pkl", + None, + "pt-snap-cli", + "1", + "2026-11-01T00:00:00+00:00", + ) + return replace(original, **changes) + + +def create_database(path): + with closing(sqlite3.connect(path)) as conn, conn: + conn.execute( + 'CREATE TABLE dictionary ("table" TEXT, "column" TEXT, "key" TEXT, "value" TEXT)' + ) + conn.execute("CREATE TABLE trace_entry_0 (id INTEGER PRIMARY KEY)") + return path + + +@pytest.mark.parametrize("value", [True, False, 1.0, "1", None]) +def test_metadata_integer_contract_rejects_coercion(value): + with pytest.raises(TypeError, match="integer"): + required_int(value) + assert required_int(0) == 0 + + +@pytest.mark.parametrize("value", ["", 0, None, b"name"]) +def test_metadata_string_contract_requires_nonempty_text(value): + with pytest.raises(TypeError, match="non-empty"): + required_str(value) + assert required_str("snapshot.pkl") == "snapshot.pkl" + + +@pytest.mark.parametrize( + "changes", + [ + {"metadata_schema_version": 0}, + {"import_format_version": 0}, + {"source_sha256": "A" * 64}, + {"source_sha256": "a" * 63}, + {"source_size": -1}, + {"source_name": ""}, + {"requested_device": -1}, + {"importer_name": ""}, + {"importer_version": ""}, + {"completed_at": "2026-11-01T00:00:00"}, + {"completed_at": "2026-11-01T01:00:00+01:00"}, + {"completed_at": "invalid"}, + ], +) +def test_invalid_values_are_rejected_before_metadata_publication(tmp_path, changes): + path = create_database(tmp_path / "snapshot.db") + service = ImportMetadataService() + original = metadata() + service.write(path, original) + before = path.read_bytes() + invalid = metadata(**changes) + + with pytest.raises(ValueError): + validate_import_metadata(invalid) + with pytest.raises(ImportMetadataError, match="invalid import metadata"): + service.write(path, invalid) + + assert path.read_bytes() == before + assert service.inspect(path).metadata == original + + +@pytest.mark.parametrize("device", [None, 0, 7]) +def test_mapping_preserves_exact_device_and_utc_metadata(device): + original = metadata(requested_device=device) + assert metadata_from_mapping(asdict(original)) == original + + +@pytest.mark.parametrize("field", ["metadata_schema_version", "source_size", "requested_device"]) +def test_mapping_rejects_boolean_integer_fields(field): + row = asdict(metadata()) + row[field] = True + with pytest.raises(TypeError, match="integer"): + metadata_from_mapping(row) + + +@pytest.mark.parametrize("operation", ["hash", "stat"]) +def test_source_io_errors_are_metadata_errors_without_creating_source(tmp_path, operation): + missing = tmp_path / "missing.pkl" + service = ImportMetadataService() + with pytest.raises(ImportMetadataError) as caught: + if operation == "hash": + service.calculate_sha256(missing) + else: + service.build_metadata(missing, "a" * 64, None) + assert isinstance(caught.value.__cause__, FileNotFoundError) + assert not missing.exists() + + +@pytest.mark.parametrize("kind", ["missing", "wrong-schema", "not-sqlite"]) +def test_invalid_databases_never_become_cache_hits(tmp_path, kind): + path = tmp_path / "snapshot.db" + if kind == "wrong-schema": + with closing(sqlite3.connect(path)) as conn, conn: + conn.execute("CREATE TABLE unrelated (value INTEGER)") + elif kind == "not-sqlite": + path.write_bytes(b"not sqlite") + original = path.read_bytes() if path.exists() else None + service = ImportMetadataService() + error = DatabaseMissingError if kind == "missing" else DatabaseSchemaError + with pytest.raises(error): + service.inspect(path) + decision = service.evaluate_cache(path, "a" * 64, None) + assert decision.reused is False + assert decision.reason == ("database_missing" if kind == "missing" else "database_invalid") + assert (path.read_bytes() if path.exists() else None) == original + + +def test_metadata_query_failure_is_invalid_and_closes_reader(tmp_path, monkeypatch): + path = create_database(tmp_path / "snapshot.db") + service = ImportMetadataService() + service.write(path, metadata()) + before = path.read_bytes() + closed = [] + + class BrokenContext: + @contextmanager + def connect(self): + try: + yield self + finally: + closed.append(True) + + def execute(self, *args): + raise sqlite3.DatabaseError("reader failure") + + monkeypatch.setattr(metadata_module, "Context", lambda path: BrokenContext()) + inspection = service.inspect(path) + assert (inspection.status, inspection.reason, inspection.metadata) == ( + "invalid", + "metadata_invalid", + None, + ) + assert closed == [True] + assert path.read_bytes() == before + + +def test_metadata_write_open_failure_is_normalized_without_creating_output(tmp_path): + path = tmp_path / "missing-parent" / "snapshot.db" + with pytest.raises(ImportMetadataError, match="Cannot write import metadata") as caught: + ImportMetadataService().write(path, metadata()) + assert isinstance(caught.value.__cause__, sqlite3.DatabaseError) + assert not path.exists() + assert not path.parent.exists() + + +@pytest.mark.parametrize("raw", [b"not-json", b'{"ptSnap":{}}', None]) +def test_dataset_metadata_corruption_is_invalid_not_reusable(tmp_path, monkeypatch, raw): + manifest = tmp_path / "manifest.json" + if raw is not None: + manifest.write_bytes(raw) + dataset = SimpleNamespace(root=tmp_path, validation=SimpleNamespace(manifest=object())) + monkeypatch.setattr(metadata_module.DatasetResolver, "inspect", lambda self, path: dataset) + + def invalid_owned_cache(path): + raise ValueError("producer metadata corrupt") + + monkeypatch.setattr(metadata_module, "inspect_owned_cache", invalid_owned_cache) + inspection = ImportMetadataService().inspect(tmp_path) + assert (inspection.status, inspection.reason, inspection.metadata) == ( + "invalid", + "metadata_invalid", + None, + ) + assert (manifest.read_bytes() if manifest.exists() else None) == raw + + +def test_unattested_dataset_metadata_is_explicitly_unavailable(tmp_path, monkeypatch): + (tmp_path / "manifest.json").write_bytes(b"{}") + dataset = SimpleNamespace(root=tmp_path, validation=SimpleNamespace(manifest=object())) + monkeypatch.setattr(metadata_module.DatasetResolver, "inspect", lambda self, path: dataset) + inspection = ImportMetadataService().inspect(tmp_path) + assert (inspection.status, inspection.reason) == ("unavailable", "metadata_missing") diff --git a/tests/core/test_critical_coverage_skills.py b/tests/core/test_critical_coverage_skills.py new file mode 100644 index 0000000..1d3d4a4 --- /dev/null +++ b/tests/core/test_critical_coverage_skills.py @@ -0,0 +1,157 @@ +"""Skill publication faults preserve the previous copy or explicit recovery bytes.""" + +from pathlib import Path + +import pytest + +import pt_snap_cli.core.skill_service as skill_module +from pt_snap_cli.core.errors import SkillCatalogError, SkillInstallError +from pt_snap_cli.core.skill_service import SkillService + + +@pytest.fixture +def skills(tmp_path): + catalog = tmp_path / "catalog" + source = catalog / "pt-snap-test" + source.mkdir(parents=True) + (source / "SKILL.md").write_text( + "---\nname: pt-snap-test\ndescription: Test skill\n---\nNew content\n", encoding="utf-8" + ) + service = SkillService( + catalog_dir=catalog, home=tmp_path / "home", cwd=tmp_path / "project", environ={} + ) + return service, source, tmp_path / "installed" / "pt-snap-test" + + +@pytest.mark.parametrize("preexisting", [False, True]) +def test_failed_staging_copy_does_not_publish_or_remove_previous_skill( + skills, monkeypatch, preexisting +): + service, source, dest = skills + if preexisting: + dest.mkdir(parents=True) + (dest / "SKILL.md").write_bytes(b"previous skill") + before = (source / "SKILL.md").read_bytes() + + def fail_copy(source, staging): + staging.mkdir() + (staging / "SKILL.md").write_bytes(b"partial copy") + raise OSError("copy interrupted") + + monkeypatch.setattr(skill_module.shutil, "copytree", fail_copy) + with pytest.raises(SkillInstallError, match="copy interrupted"): + service.install_skills(dest_dir=dest.parent, force=True) + assert (source / "SKILL.md").read_bytes() == before + if preexisting: + assert (dest / "SKILL.md").read_bytes() == b"previous skill" + assert list(dest.parent.iterdir()) == [dest] + else: + assert not dest.exists() + assert list(dest.parent.iterdir()) == [] + + +def test_failed_publish_and_restore_keep_previous_backup(skills, monkeypatch): + service, source, dest = skills + dest.mkdir(parents=True) + (dest / "SKILL.md").write_bytes(b"previous skill") + replace = Path.replace + + def fail_staging_and_restore(path, target): + if path.name.endswith((".pt-snap-staging", ".pt-snap-backup")): + raise OSError("rename denied") + return replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_staging_and_restore) + with pytest.raises(SkillInstallError, match="rename denied"): + service.install_skills(dest_dir=dest.parent, force=True) + backups = list(dest.parent.glob("*.pt-snap-backup")) + assert len(backups) == 1 + assert (backups[0] / "SKILL.md").read_bytes() == b"previous skill" + assert not dest.exists() + assert not list(dest.parent.glob("*.pt-snap-staging")) + assert (source / "SKILL.md").is_file() + + +def test_backup_cleanup_failure_does_not_mask_successful_publication(skills, monkeypatch): + service, source, dest = skills + dest.mkdir(parents=True) + (dest / "SKILL.md").write_bytes(b"previous skill") + remove = skill_module._remove_path + + def fail_backup_cleanup(path): + if path.name.endswith(".pt-snap-backup"): + raise OSError("cleanup denied") + remove(path) + + monkeypatch.setattr(skill_module, "_remove_path", fail_backup_cleanup) + report = service.install_skills(dest_dir=dest.parent, force=True) + assert report.results[0].action == "updated" + assert (dest / "SKILL.md").read_bytes() == (source / "SKILL.md").read_bytes() + backups = list(dest.parent.glob("*.pt-snap-backup")) + assert len(backups) == 1 + assert (backups[0] / "SKILL.md").read_bytes() == b"previous skill" + assert not list(dest.parent.glob("*.pt-snap-staging")) + + +def test_uninstall_oserror_is_normalized_without_touching_other_paths(skills, monkeypatch): + service, _, dest = skills + service.install_skills(dest_dir=dest.parent) + original = (dest / "SKILL.md").read_bytes() + + def denied(path): + assert path == dest + raise PermissionError("delete denied") + + monkeypatch.setattr(skill_module, "_remove_path", denied) + with pytest.raises(SkillInstallError, match="Failed to uninstall") as caught: + service.uninstall_skills(dest_dir=dest.parent) + assert isinstance(caught.value.__cause__, PermissionError) + assert (dest / "SKILL.md").read_bytes() == original + assert list(dest.parent.iterdir()) == [dest] + + +def test_upgrade_current_skill_is_an_explicit_noop(skills): + service, _, dest = skills + service.install_skills(dest_dir=dest.parent) + before = (dest / "SKILL.md").stat() + report = service.upgrade_skills(dest_dir=dest.parent) + after = (dest / "SKILL.md").stat() + assert report.results[0].action == "already_installed" + assert (after.st_ino, after.st_mtime_ns) == (before.st_ino, before.st_mtime_ns) + + +def test_catalog_falls_back_to_package_and_fails_when_no_source_exists(tmp_path, monkeypatch): + packaged = tmp_path / "bundled" + monkeypatch.setattr(skill_module, "_discover_repo_skills_dir", lambda: None) + monkeypatch.setattr(skill_module, "_packaged_skills_dir", lambda: packaged) + assert skill_module.default_catalog_dir({}) == packaged + monkeypatch.setattr(skill_module, "_packaged_skills_dir", lambda: None) + with pytest.raises(SkillCatalogError, match="not available"): + skill_module.default_catalog_dir({}) + + +@pytest.mark.parametrize( + "content", ["plain skill", "---\nname: incomplete", "---\n- list\n---\nbody"] +) +def test_legacy_frontmatter_falls_back_to_directory_name(tmp_path, content): + catalog = tmp_path / "catalog" + skill = catalog / "pt-snap-fallback" + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text(content, encoding="utf-8") + specs = skill_module.list_catalog_skills(catalog) + assert [(spec.name, spec.description) for spec in specs] == [("pt-snap-fallback", "")] + + +def test_unreadable_frontmatter_is_a_catalog_error_without_mutation(skills, monkeypatch): + service, source, dest = skills + read_text = Path.read_text + + def denied(path, *args, **kwargs): + if path == source / "SKILL.md": + raise PermissionError("read denied") + return read_text(path, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", denied) + with pytest.raises(SkillCatalogError, match="Failed to read skill file"): + service.install_skills(dest_dir=dest.parent) + assert not dest.parent.exists() diff --git a/tests/core/test_critical_coverage_split.py b/tests/core/test_critical_coverage_split.py new file mode 100644 index 0000000..9fc0231 --- /dev/null +++ b/tests/core/test_critical_coverage_split.py @@ -0,0 +1,193 @@ +"""Unsuccessful replay or exclusive publication must leave no partial result.""" + +import errno +import os +import pickle +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import pt_snap_cli.core.split_service as split_module +from pt_snap_cli.core.errors import SplitError +from pt_snap_cli.core.models import SplitOptions +from pt_snap_cli.core.split_service import SplitService + + +@pytest.fixture +def split_input(tmp_path): + source = tmp_path / "snapshot.pkl" + # Test-owned input only. Fault-injected replay below owns the generated + # output; actual allocator validity is covered by the ordinary split suite. + source.write_bytes(pickle.dumps({"segments": [], "device_traces": [[{"id": 0}]]})) + return SplitOptions(snapshot_file=source, output=tmp_path / "split", slices=1) + + +def assert_unpublished(options): + assert not options.output.exists() + assert not list(options.output.parent.glob(".split.pt-snap-*")) + + +def test_missing_device_trace_list_fails_without_staging(split_input): + split_input.snapshot_file.write_bytes(pickle.dumps({"device_traces": {}})) + before = split_input.snapshot_file.read_bytes() + with pytest.raises(SplitError, match="no device trace list") as caught: + SplitService().split(split_input) + assert caught.value.phase == "device" + assert split_input.snapshot_file.read_bytes() == before + assert_unpublished(split_input) + + +def test_stage_creation_failure_is_normalized_without_loading_replay(split_input, monkeypatch): + def denied(*args, **kwargs): + raise PermissionError("staging denied") + + monkeypatch.setattr(split_module.tempfile, "mkdtemp", denied) + monkeypatch.setattr( + split_module, + "replay_snapshot", + lambda *args, **kwargs: pytest.fail("no replay before staging"), + ) + with pytest.raises(SplitError, match="cannot create staging directory") as caught: + SplitService().split(split_input) + assert caught.value.phase == "publication" + assert isinstance(caught.value.__cause__, PermissionError) + assert_unpublished(split_input) + + +@pytest.mark.parametrize( + ("failure", "phase", "message"), + [ + ("source-replay", "load/engine", "replay engine returned false"), + ("empty-output", "load/engine", "engine generated no slices"), + ("bad-filename", "generated-validation", "Unexpected engine output name"), + ("generated-replay", "generated-validation", "replay returned false"), + ], +) +def test_replay_failures_remove_owned_stage_and_preserve_source( + split_input, monkeypatch, failure, phase, message +): + before = split_input.snapshot_file.read_bytes() + calls = [] + + def replay(representation, device, **kwargs): + calls.append("source" if "hooker" in kwargs else "generated") + if "hooker" not in kwargs: + return None, False + if failure in ("bad-filename", "generated-replay"): + stages = list(split_input.output.parent.glob(".split.pt-snap-*")) + assert len(stages) == 1 + raw = stages[0] / ".device-0" + name = "unexpected.pkl" if failure == "bad-filename" else "snapshot_entry_0_0.pkl" + (raw / name).write_bytes(before) + return None, failure != "source-replay" + + monkeypatch.setattr(split_module, "replay_snapshot", replay) + with pytest.raises(SplitError, match=message) as caught: + SplitService().split(split_input) + assert caught.value.phase == phase + assert calls == (["source", "generated"] if failure == "generated-replay" else ["source"]) + assert split_input.snapshot_file.read_bytes() == before + assert_unpublished(split_input) + + +@pytest.mark.parametrize("method", ["_stage_identity", "_verify_stage_identity"]) +def test_non_directory_stage_is_never_accepted_or_deleted(tmp_path, method): + path = tmp_path / "not-a-stage" + path.write_bytes(b"another owner's file") + with pytest.raises(OSError, match="not an owned directory"): + if method == "_stage_identity": + SplitService._stage_identity(path) + else: + SplitService._verify_stage_identity(path, (0, 0, None)) + assert path.read_bytes() == b"another owner's file" + + +def test_descriptor_identity_mismatch_is_rejected_without_deleting_stage(tmp_path): + stage = tmp_path / "stage" + stage.mkdir() + unrelated = tmp_path / "unrelated-file" + unrelated.write_bytes(b"another owner's evidence") + status = stage.lstat() + # An actual mismatched descriptor exercises the held-identity guard on every + # platform, including Windows where ordinary stages do not hold directory fds. + descriptor = os.open(unrelated, os.O_RDONLY) + identity = (status.st_dev, status.st_ino, descriptor) + try: + held = os.fstat(descriptor) + assert (held.st_dev, held.st_ino) != identity[:2] + with pytest.raises(OSError, match="identity changed"): + SplitService._verify_stage_identity(stage, identity) + assert stage.is_dir() + assert unrelated.read_bytes() == b"another owner's evidence" + finally: + SplitService._close_stage_identity(identity) + with pytest.raises(OSError) as closed: + os.fstat(descriptor) + assert closed.value.errno == errno.EBADF + + +def test_closed_stage_descriptor_cleanup_is_idempotent(tmp_path): + identity = SplitService._stage_identity(tmp_path) + SplitService._close_stage_identity(identity) + SplitService._close_stage_identity(identity) + assert tmp_path.is_dir() + + +@pytest.mark.parametrize("platform", ["darwin", "linux"]) +def test_native_publication_errors_preserve_both_paths(tmp_path, monkeypatch, platform): + stage = tmp_path / "stage" + destination = tmp_path / "destination" + stage.mkdir() + destination.mkdir() + (stage / "slice.pkl").write_bytes(b"staged") + (destination / "owner.txt").write_bytes(b"other owner") + + class FailedRename: + def __call__(self, *args): + return -1 + + library = SimpleNamespace(renamex_np=FailedRename(), renameat2=FailedRename()) + monkeypatch.setattr(split_module.sys, "platform", platform) + monkeypatch.setattr(split_module.ctypes, "CDLL", lambda *args, **kwargs: library) + monkeypatch.setattr(split_module.ctypes, "get_errno", lambda: errno.EEXIST) + with pytest.raises(OSError) as caught: + SplitService._publish_directory(stage, destination) + assert caught.value.errno == errno.EEXIST + assert Path(caught.value.filename) == destination + assert (stage / "slice.pkl").read_bytes() == b"staged" + assert (destination / "owner.txt").read_bytes() == b"other owner" + + +@pytest.mark.parametrize("platform", ["darwin", "linux", "unsupported"]) +def test_unavailable_no_replace_primitive_fails_closed(tmp_path, monkeypatch, platform): + stage, destination = tmp_path / "stage", tmp_path / "destination" + stage.mkdir() + monkeypatch.setattr(split_module.sys, "platform", platform) + monkeypatch.setattr(split_module, "os", SimpleNamespace(name="posix")) + monkeypatch.setattr(split_module.ctypes, "CDLL", lambda *args, **kwargs: object()) + with pytest.raises(OSError) as caught: + SplitService._preflight_publication() + assert caught.value.errno == errno.ENOTSUP + if platform != "darwin": + with pytest.raises(OSError) as caught: + SplitService._publish_directory(stage, destination) + assert caught.value.errno == errno.ENOTSUP + assert stage.is_dir() + assert not destination.exists() + + +def test_windows_publication_uses_non_replacing_rename(tmp_path, monkeypatch): + stage, destination = tmp_path / "stage", tmp_path / "destination" + calls = [] + + def rename(source, target): + calls.append((source, target)) + raise FileExistsError(errno.EEXIST, "destination exists", target) + + monkeypatch.setattr(split_module.sys, "platform", "win32") + monkeypatch.setattr(split_module, "os", SimpleNamespace(name="nt", rename=rename)) + SplitService._preflight_publication() + with pytest.raises(FileExistsError): + SplitService._publish_directory(stage, destination) + assert calls == [(stage, destination)] diff --git a/tests/snapshot/test_representation_boundaries.py b/tests/snapshot/test_representation_boundaries.py new file mode 100644 index 0000000..a3d2272 --- /dev/null +++ b/tests/snapshot/test_representation_boundaries.py @@ -0,0 +1,214 @@ +"""Representation dispatch, canonical values and real replay entrypoint contracts.""" + +import copy +import hashlib +import json +from pathlib import Path +from typing import cast + +import pytest + +from pt_snap_cli.snapshot.base import DeviceSnapshot, Segment, TraceEntry +from pt_snap_cli.snapshot.representation import ( + SnapshotFormat, + canonical_snapshot_bytes, + canonical_snapshot_sha256, + canonicalize_snapshot, + load_and_replay_snapshot, + load_snapshot_representation, + replay_snapshot, + save_snapshot_representation, + serialize_snapshot_representation, +) +from pt_snap_cli.snapshot.simulate import AllocatorHooker, SimulateHooker + +from .helpers import assert_valid_snapshot + + +def _representation() -> dict[str, object]: + frames = [{"filename": "test.py", "name": "allocate", "line": 17}] + return { + "segments": [ + { + "address": 4096, + "total_size": 64, + "stream": 0, + "segment_type": "small", + "allocated_size": 0, + "active_size": 0, + "device": 1, + "frames": copy.deepcopy(frames), + "blocks": [], + } + ], + "device_traces": [ + [], + [ + { + "id": 17, + "action": "segment_alloc", + "addr": 4096, + "size": 64, + "stream": 0, + "frames": frames, + } + ], + ], + "metadata": {"label": "内存", "ratio": 1.25, "optional": None}, + } + + +@pytest.mark.parametrize("snapshot_format", ["pickle", "json"]) +@pytest.mark.parametrize("valid_replay", [True, False]) +def test_dispatch_round_trip_and_replay_preserve_source( + tmp_path: Path, snapshot_format: SnapshotFormat, valid_replay: bool +) -> None: + value = _representation() + if not valid_replay: + # The event cannot be undone without its segment: replay must report False. + value["segments"] = [] + original = copy.deepcopy(value) + path = tmp_path / f"snapshot.{snapshot_format}" + encoded = serialize_snapshot_representation(value, snapshot_format) + save_snapshot_representation(value, str(path), snapshot_format) + assert path.read_bytes() == encoded + assert load_snapshot_representation(str(path), snapshot_format) == original + + loaded, runtime, replayed = load_and_replay_snapshot(path, snapshot_format, 1) + + assert loaded == original == value + assert runtime.device == 1 + assert replayed is valid_replay + assert [entry.idx for entry in runtime.device_snapshot.trace_entries] == ( + [] if valid_replay else [17] + ) + assert runtime.device_snapshot.segments == [] + assert runtime.device_snapshot.total_reserved == 0 + assert_valid_snapshot(runtime.device_snapshot) + assert path.read_bytes() == encoded + assert list(tmp_path.iterdir()) == [path] + + +def test_default_dispatch_uses_protocol_four_pickle(tmp_path: Path) -> None: + value = _representation() + path = tmp_path / "snapshot.pkl" + encoded = serialize_snapshot_representation(value) + assert encoded.startswith(b"\x80\x04") + save_snapshot_representation(value, path) + assert path.read_bytes() == encoded + assert load_snapshot_representation(path) == value + + +@pytest.mark.parametrize("exists", [True, False]) +def test_unsupported_formats_do_not_read_or_overwrite_outputs(tmp_path: Path, exists: bool): + path = tmp_path / "snapshot.yaml" + sentinel = b"existing output must be preserved" + if exists: + path.write_bytes(sentinel) + value = _representation() + original = copy.deepcopy(value) + unsupported = cast(SnapshotFormat, "yaml") + + with pytest.raises(ValueError, match="Unsupported snapshot format: yaml"): + load_snapshot_representation(path, unsupported) + with pytest.raises(ValueError, match="Unsupported snapshot format: yaml"): + serialize_snapshot_representation(value, unsupported) + with pytest.raises(ValueError, match="Unsupported snapshot format: yaml"): + save_snapshot_representation(value, path, unsupported) + + assert value == original + assert path.exists() is exists + if exists: + assert path.read_bytes() == sentinel + assert list(tmp_path.iterdir()) == ([path] if exists else []) + + +@pytest.mark.parametrize("value", [None, [], "snapshot", 7, 1.25, True]) +def test_json_requires_object_without_rewriting_source(tmp_path: Path, value: object): + path = tmp_path / "snapshot.json" + original = json.dumps(value).encode() + path.write_bytes(original) + with pytest.raises(ValueError, match=f"actual type: {type(value).__name__}"): + load_snapshot_representation(path, "json") + assert path.read_bytes() == original + assert list(tmp_path.iterdir()) == [path] + + +def test_canonical_finite_values_preserve_sequence_order_and_utf8(): + value = {"z": (None, True, -7, 1.25, -0.0), "a": {"标签": "内存", "order": [3, 1, 2]}} + original = copy.deepcopy(value) + expected = '{"a":{"order":[3,1,2],"标签":"内存"},"z":[null,true,-7,1.25,-0.0]}'.encode() + assert canonical_snapshot_bytes(value) == expected + assert canonical_snapshot_sha256(value) == hashlib.sha256(expected).hexdigest() + assert canonicalize_snapshot(value) == json.loads(expected) + assert value == original + + +@pytest.mark.parametrize("value", [float("inf"), float("-inf")]) +def test_json_nonfinite_values_do_not_replace_previous_output(tmp_path: Path, value: float): + path = tmp_path / "snapshot.json" + path.write_bytes(b"previous output") + with pytest.raises(ValueError, match="non-finite floats"): + save_snapshot_representation({"nested": [value]}, path, "json") + assert path.read_bytes() == b"previous output" + assert list(tmp_path.iterdir()) == [path] + + +class _ReplayObserver(SimulateHooker): + def __init__(self) -> None: + self.calls: list[tuple[str, int, int]] = [] + self.raw_frames: list[bool] = [] + + def pre_undo_event(self, event: TraceEntry, snapshot: DeviceSnapshot) -> bool: + self.calls.append(("pre", event.idx, snapshot.total_reserved)) + self.raw_frames.append(event._raw_frames is not None) + return True + + def post_undo_event(self, event: TraceEntry, snapshot: DeviceSnapshot) -> bool: + self.calls.append(("post", event.idx, snapshot.total_reserved)) + return True + + +class _AllocatorObserver(AllocatorHooker): + def __init__(self) -> None: + self.calls: list[tuple[str, int, int]] = [] + + def pre_replay_unmap_or_free_segment(self, segment: Segment, snapshot: DeviceSnapshot): + self.calls.append(("pre", segment.address, snapshot.total_reserved)) + + def post_replay_unmap_or_free_segment(self, segment: Segment, snapshot: DeviceSnapshot): + self.calls.append(("post", segment.address, snapshot.total_reserved)) + + +@pytest.mark.parametrize( + ("with_event_hook", "with_allocator_hook"), + [(False, False), (True, False), (False, True), (True, True)], +) +@pytest.mark.parametrize("raw_frames", [False, True]) +def test_replay_entrypoint_registers_optional_hooks_before_real_replay( + with_event_hook: bool, with_allocator_hook: bool, raw_frames: bool +) -> None: + value = _representation() + original = copy.deepcopy(value) + event_hook = _ReplayObserver() + allocator_hook = _AllocatorObserver() + + runtime, replayed = replay_snapshot( + value, + 1, + hooker=event_hook if with_event_hook else None, + allocator_hooker=allocator_hook if with_allocator_hook else None, + _raw_frames=raw_frames, + ) + + assert replayed is True + assert runtime.device == 1 + assert event_hook.calls == ([("pre", 17, 64), ("post", 17, 0)] if with_event_hook else []) + assert event_hook.raw_frames == ([raw_frames] if with_event_hook else []) + assert allocator_hook.calls == ( + [("pre", 4096, 64), ("post", 4096, 0)] if with_allocator_hook else [] + ) + assert runtime.device_snapshot.trace_entries == [] + assert runtime.device_snapshot.segments == [] + assert_valid_snapshot(runtime.device_snapshot) + assert value == original diff --git a/tests/test_ci_contract_mutations.py b/tests/test_ci_contract_mutations.py index a579e2a..c1a021f 100644 --- a/tests/test_ci_contract_mutations.py +++ b/tests/test_ci_contract_mutations.py @@ -7,6 +7,8 @@ from tests._ci_contract import command_tokens, commands, expression, matrix_rows from tests.test_ci_tiers import project, validate_workflow, workflow +CRITICAL_CHECKER = ".github/scripts/check_critical_coverage.py" + def _step(config, job, program, contains=None): matching = [ @@ -57,13 +59,8 @@ def _quotes_and_commands(config): "python3 -m pytest --cov-report term-missing --cov -m 'not nightly' --cov-report xml", "--cov", ) - guard = _step(config, "test", "coverage") - guard["run"] = ( - guard["run"] - .replace("--include='", '--include "') - .replace(".py'", '.py"') - .replace("--fail-under=", "--fail-under ") - ) + guard = _step(config, "test", CRITICAL_CHECKER) + guard["run"] = 'python3 ".github/scripts/check_critical_coverage.py"' def _conditions(config): @@ -90,19 +87,23 @@ def _excluded_then_restored_matrix(config): def _split_and_merge(config): job = config["jobs"]["test"] - guard = _step(config, "test", "coverage") - index = job["steps"].index(guard) - parts = [dict(guard, run=line) for line in guard["run"].splitlines() if line.strip()] + build = _step(config, "test", "build") + index = job["steps"].index(build) + parts = [dict(build, run=line) for line in build["run"].splitlines() if line.strip()] + assert len(parts) > 1, "Positive split mutation must split real sequential commands" job["steps"][index : index + 1] = parts - # Merge full pytest with the first independent floor; labels are irrelevant. + # Independently merge full pytest with its checker; labels are irrelevant. + guard = _step(config, "test", CRITICAL_CHECKER) full = _step(config, "test", "pytest", "--cov") - full["run"] += "\n" + parts[0]["run"] - job["steps"].remove(parts[0]) + full["run"] += "\n" + guard["run"] + job["steps"].remove(guard) def _and_chain(config): - guard = _step(config, "test", "coverage") - guard["run"] = " && ".join(guard["run"].splitlines()) + guard = _step(config, "test", CRITICAL_CHECKER) + full = _step(config, "test", "pytest", "--cov") + full["run"] += " && " + guard["run"] + config["jobs"]["test"]["steps"].remove(guard) def _include_matrix(config): @@ -136,7 +137,7 @@ def _weaken(config, case): job = config["jobs"]["test"] full = _step(config, "test", "pytest", "--cov") fast = _step(config, "test", "pytest", "not slow and not nightly") - guard = _step(config, "test", "coverage") + guard = _step(config, "test", CRITICAL_CHECKER) browser = _step(config, "browser", "pytest") if case == "missing-python": job["strategy"]["matrix"]["python-version"].remove("3.12") @@ -211,14 +212,30 @@ def _weaken(config, case): full["run"] += " -k passing_only" elif case == "pytest-path-filter": full["run"] += " tests/core" - elif case == "coverage-threshold": - guard["run"] = guard["run"].replace("--fail-under=90", "--fail-under=89") - elif case == "coverage-combined-selector": - guard["run"] = guard["run"].replace("*/core/snapshot_import_backend.py", "*/core/*.py") - elif case == "coverage-wrong-file": - guard["run"] = guard["run"].replace( - "*/core/snapshot_import_backend.py", "*/core/import_metadata.py" - ) + elif case == "coverage-checker-omitted": + job["steps"].remove(guard) + elif case == "coverage-checker-duplicate": + job["steps"].append(deepcopy(guard)) + elif case == "coverage-checker-conditional": + guard["if"] += " && github.event_name == 'push'" + elif case == "coverage-checker-wrong-python": + guard["if"] = "matrix.python-version == '3.12'" + elif case == "coverage-checker-overrides": + guard["run"] += " --fail-under=0" + elif case == "coverage-checker-positional": + guard["run"] += " other.coverage" + elif case == "coverage-checker-ignored-failure": + guard["continue-on-error"] = "true" + elif case == "coverage-checker-or-true": + guard["run"] += " || true" + elif case == "coverage-checker-and-list-swallow": + guard["run"] += " && pip --version\npip --version" + elif case == "coverage-checker-conditional-shell": + guard["run"] = "if false; then " + guard["run"] + "; fi" + elif case == "coverage-alternate-report": + guard["run"] = "python -m coverage report --fail-under=0" + elif case == "coverage-data-erased": + guard["run"] = "python -m coverage erase\n" + guard["run"] elif case == "coverage-before-measurement": job["steps"].remove(guard) job["steps"].insert(job["steps"].index(full), guard) @@ -313,9 +330,18 @@ def _weaken(config, case): "full-includes-nightly", "pytest-k-filter", "pytest-path-filter", - "coverage-threshold", - "coverage-combined-selector", - "coverage-wrong-file", + "coverage-checker-omitted", + "coverage-checker-duplicate", + "coverage-checker-conditional", + "coverage-checker-wrong-python", + "coverage-checker-overrides", + "coverage-checker-positional", + "coverage-checker-ignored-failure", + "coverage-checker-or-true", + "coverage-checker-and-list-swallow", + "coverage-checker-conditional-shell", + "coverage-alternate-report", + "coverage-data-erased", "coverage-before-measurement", "fixture-after-tests", "optional-browser", @@ -349,6 +375,35 @@ def test_coverage_configuration_cannot_weaken(field, value): validate_workflow(workflow(), settings) +def test_coverage_configuration_order_and_display_precision_do_not_change_gate(): + settings = deepcopy(project()) + coverage = settings["tool"]["coverage"] + coverage["run"]["omit"].reverse() + coverage["report"]["exclude_lines"].reverse() + coverage["report"]["precision"] = 4 + validate_workflow(workflow(), settings) + + +@pytest.mark.parametrize( + ("section", "field", "value"), + [ + ("run", "source", ["pt_snap_cli.snapshot"]), + ("run", "omit", ["*/tests/*", "*/__pycache__/*", "*/core/*"]), + ("run", "include", ["*/snapshot/*"]), + ("report", "include", ["*/snapshot/*"]), + ("report", "omit", ["*/core/*"]), + ("report", "exclude_lines", [".*"]), + ("report", "exclude_also", [".*"]), + ("report", "partial_branches", [".*"]), + ], +) +def test_coverage_selectors_and_exclusions_cannot_hide_runtime(section, field, value): + settings = deepcopy(project()) + settings["tool"]["coverage"][section][field] = value + with pytest.raises(AssertionError): + validate_workflow(workflow(), settings) + + def test_include_matrix_rows_never_become_new_cartesian_originals(): assert matrix_rows( { diff --git a/tests/test_ci_tiers.py b/tests/test_ci_tiers.py index 0be3274..496e94b 100644 --- a/tests/test_ci_tiers.py +++ b/tests/test_ci_tiers.py @@ -112,37 +112,42 @@ def validate_tiers(config, settings): coverage = settings["tool"]["coverage"] require(coverage["run"]["branch"] is True, "Branch coverage is mandatory") - require(coverage["report"]["fail_under"] >= 90, "Overall coverage floor must not decrease") - expected = { - "*/core/snapshot_import_backend.py": 90, - "*/simulate/simulated_caching_allocator.py": 88, - "*/simulate/snapshot_mutator.py": 84, - } - reports = select(all_commands, "coverage") - require(len(reports) == len(expected), "Each critical file needs its own coverage floor") - seen = set() - for report in reports: - require(report.words[1] == "report", "Critical coverage must report measured results") - opts, positional = options(report.words[2:], valued=("--include", "--fail-under")) - require( - not positional and set(opts) == {"--include", "--fail-under"}, - "Coverage report must have an independent selector and floor", - ) - require( - all(len(values) == 1 for values in opts.values()), - "Duplicate coverage options are ambiguous", - ) - path = opts["--include"][0] - require( - path in expected and path not in seen, "Wrong/combined critical-file coverage selector" - ) - require( - float(opts["--fail-under"][0]) >= expected[path], - "Critical coverage floor must not decrease", - ) - seen.add(path) - require(full.index < report.index, "Coverage must be measured before its floors") - _full(job, report.step) + require( + coverage["report"]["fail_under"] >= 90, + "Overall coverage floor must not decrease", + ) + require( + coverage["run"].get("source") == ["pt_snap_cli"], + "Coverage must measure the entire runtime package", + ) + require( + set(coverage["run"].get("omit", [])) == {"*/tests/*", "*/__pycache__/*"}, + "Runtime modules cannot be omitted from coverage", + ) + require( + not coverage["run"].get("include") + and not coverage["report"].get("include") + and not coverage["report"].get("omit"), + "Coverage measurement/report selectors must not narrow the package", + ) + require( + set(coverage["report"].get("exclude_lines", [])) + == {"pragma: no cover", "if __name__ == .__main__.", "if TYPE_CHECKING:"} + and not coverage["report"].get("exclude_also") + and not any(key.startswith("partial_") for key in coverage["report"]), + "Coverage exclusions must not hide runtime statements or branches", + ) + require( + not select(all_commands, "coverage"), + "Alternate coverage commands cannot replace or bypass the critical checker", + ) + guard = one(all_commands, ".github/scripts/check_critical_coverage.py") + require( + len(guard.words) == 1, + "Critical coverage checker must use its complete fixed inventory without overrides", + ) + require(full.index < guard.index, "Coverage must be measured before its floors") + _full(job, guard.step) build = one(all_commands, "build") require(build.words[1:] == ("--wheel",), "Build must produce the wheel") diff --git a/tests/test_critical_coverage_gate.py b/tests/test_critical_coverage_gate.py new file mode 100644 index 0000000..76ccc7c --- /dev/null +++ b/tests/test_critical_coverage_gate.py @@ -0,0 +1,170 @@ +"""Exercise the coverage guard with real coverage databases, without running code.""" + +import importlib.util +import sqlite3 +from contextlib import closing +from pathlib import Path + +import pytest +from coverage import CoverageData + + +@pytest.fixture +def guard(): + path = Path(".github/scripts/check_critical_coverage.py") + spec = importlib.util.spec_from_file_location("critical_coverage_guard", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def coverage_connections(tmp_path, monkeypatch): + connect = sqlite3.connect + connections = [] + + def tracked_connect(database, *args, **kwargs): + connection = connect(database, *args, **kwargs) + if Path(database) == tmp_path / ".coverage": + connections.append(connection) + return connection + + monkeypatch.setattr(sqlite3, "connect", tracked_connect) + try: + yield connections + finally: + # The tests assert closure before teardown; clean up even after a failed + # assertion so a primary lifetime regression does not cause a second leak. + for connection in connections: + connection.close() + + +def assert_connections_closed(connections): + assert connections, "The checker must have inspected real SQLite evidence" + for connection in connections: + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + + +def write_evidence(guard, root, *, missing=None, uncovered=None, branches=True): + (root / "pyproject.toml").write_text( + "[tool.coverage.run]\nbranch = true\n[tool.coverage.report]\nfail_under = 90\n", + encoding="utf-8", + ) + with closing(CoverageData(basename=str(root / ".coverage"))) as data: + for relative in guard.CRITICAL_FLOORS: + source = root / relative + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text("if flag:\n result = 1\nelse:\n result = 2\n", encoding="utf-8") + if relative == missing: + continue + if branches: + arcs = [(-1, 1), (1, 2), (2, -1)] + if relative != uncovered: + arcs.extend([(1, 4), (4, -1)]) + data.add_arcs({str(source): arcs}) + else: + data.add_lines({str(source): [1, 2, 4]}) + data.write() + + +def test_critical_floor_inventory_preserves_existing_and_adds_risky_modules(guard): + assert guard.CRITICAL_FLOORS == { + "src/pt_snap_cli/core/snapshot_import_backend.py": 90, + "src/pt_snap_cli/snapshot/simulate/simulated_caching_allocator.py": 88, + "src/pt_snap_cli/snapshot/simulate/snapshot_mutator.py": 84, + "src/pt_snap_cli/core/import_metadata_contract.py": 100, + "src/pt_snap_cli/core/import_metadata.py": 90, + "src/pt_snap_cli/core/split_service.py": 90, + "src/pt_snap_cli/core/focus_service.py": 90, + "src/pt_snap_cli/core/skill_service.py": 90, + "src/pt_snap_cli/core/snapshot_cache.py": 90, + "src/pt_snap_cli/snapshot/representation.py": 95, + } + + +def test_guard_accepts_complete_branch_evidence(guard, tmp_path, capsys, coverage_connections): + write_evidence(guard, tmp_path) + read_start = len(coverage_connections) + guard.check_coverage(tmp_path) + output = capsys.readouterr().out + assert output.count("100.0000%") == len(guard.CRITICAL_FLOORS) + assert_connections_closed(coverage_connections[read_start:]) + + +def test_guard_rejects_missing_data_without_creating_it(guard, tmp_path): + with pytest.raises(guard.CoverageGuardError, match="Missing coverage data"): + guard.check_coverage(tmp_path) + assert not (tmp_path / ".coverage").exists() + + +def test_guard_rejects_unsupported_coverage_before_opening_evidence(guard, tmp_path, monkeypatch): + write_evidence(guard, tmp_path) + before = (tmp_path / ".coverage").read_bytes() + monkeypatch.setattr(guard, "CoverageData", object) + monkeypatch.setattr( + guard, "Coverage", lambda **kwargs: pytest.fail("Unsupported coverage must not open data") + ) + with pytest.raises(guard.CoverageGuardError, match=r"requires coverage>=7\.10\.3"): + guard.check_coverage(tmp_path) + assert (tmp_path / ".coverage").read_bytes() == before + + +@pytest.mark.parametrize("failure", ["corrupt", "empty", "line-only"]) +def test_guard_rejects_unusable_evidence(guard, tmp_path, failure, coverage_connections): + write_evidence(guard, tmp_path, branches=failure != "line-only") + if failure == "corrupt": + (tmp_path / ".coverage").write_bytes(b"not a coverage database") + elif failure == "empty": + (tmp_path / ".coverage").write_bytes(b"") + read_start = len(coverage_connections) + with pytest.raises(guard.CoverageGuardError): + guard.check_coverage(tmp_path) + assert_connections_closed(coverage_connections[read_start:]) + if failure == "corrupt": + assert (tmp_path / ".coverage").read_bytes() == b"not a coverage database" + + +@pytest.mark.parametrize("failure", ["unmeasured", "deleted", "no-statements", "below-floor"]) +def test_one_critical_file_cannot_be_hidden_by_other_full_coverage(guard, tmp_path, failure): + relative = "src/pt_snap_cli/core/focus_service.py" + write_evidence( + guard, + tmp_path, + missing=relative if failure == "unmeasured" else None, + uncovered=relative if failure == "below-floor" else None, + ) + if failure == "deleted": + (tmp_path / relative).unlink() + elif failure == "no-statements": + (tmp_path / relative).write_text("# no executable source\n", encoding="utf-8") + with pytest.raises(guard.CoverageGuardError, match="core/focus_service.py"): + guard.check_coverage(tmp_path) + + +def test_guard_compares_unrounded_percentage(guard, tmp_path, monkeypatch): + write_evidence(guard, tmp_path) + real_report = guard.Coverage.report + + def almost_passes(coverage, **kwargs): + if Path(kwargs["morfs"][0]).as_posix().endswith("core/focus_service.py"): + return 89.9999 + return real_report(coverage, **kwargs) + + monkeypatch.setattr(guard.Coverage, "report", almost_passes) + with pytest.raises(guard.CoverageGuardError, match="89.9999% is below 90%"): + guard.check_coverage(tmp_path) + + +@pytest.mark.parametrize("failure", [False, True]) +def test_command_exit_status_reflects_gate_result(guard, monkeypatch, capsys, failure): + def check(root): + assert root == Path(guard.__file__).resolve().parents[2] + if failure: + raise guard.CoverageGuardError("independent floor failed") + + monkeypatch.setattr(guard, "check_coverage", check) + assert guard.main() == (1 if failure else 0) + error = capsys.readouterr().err + assert ("independent floor failed" in error) is failure