"""Path and live-authority regressions for the inversion evidence verifier. API assumptions (intentional v1.3 contract): * ``verify_inversion_session`` accepts all five live authority paths and validates them before delegating to their loaders. * Each authority is opened descriptor-first: no symlink component, hard-linked authority file, unsafe writable mode, missing path, or lexical traversal is accepted. * ``_verify_inversion_session_with_authorities`` pins the session tree for its full lifetime, rejects grammar changes, and detects replacement/mutation. * ``publish_inversion_session`` writes leaf manifests after leaf payloads, writes the session manifest last, and rolls back an interrupted promotion. The bundle is synthetic and CPU-only. No canonical evidence, model snapshot, or GPU is used by these tests. """ from __future__ import annotations import os from pathlib import Path import shutil import pytest import music3lab.fd_io as fd_io import music3lab.inversion as inversion from music3lab.inversion import ( _verify_inversion_session_with_authorities, build_experiment_artifacts, load_inversion_config, publish_inversion_session, verify_inversion_session, ) from test_inversion_phase2 import CONFIG, _evidence_adapter, _fake_result, _oracle AUTHORITY_NAMES = ("config", "snapshot", "base_manifest", "diffusers", "phase0") @pytest.fixture(scope="module") def canonical_bundle(tmp_path_factory: pytest.TempPathFactory): loaded, oracle = load_inversion_config(CONFIG), _oracle() adapter = _evidence_adapter() built = tuple( build_experiment_artifacts( _fake_result(experiment_id), loaded_config=loaded, adapter=adapter, oracle=oracle, device_name="NVIDIA H100 80GB HBM3", device_capability=(9, 0), cuda_runtime="13.0", ) for experiment_id in ("P2-E1", "P2-E2") ) root = tmp_path_factory.mktemp("path-authority") / "session" publish_inversion_session( output_root=root, loaded_config=loaded, adapter=adapter, oracle=oracle, experiments=built, ) return root, loaded, adapter, oracle, built @pytest.fixture def bundle(tmp_path: Path, canonical_bundle): source, loaded, adapter, oracle, _built = canonical_bundle root = tmp_path / "session" shutil.copytree(source, root) root.parent.chmod(0o755) return root, loaded, adapter, oracle def _reject(root: Path, loaded, adapter, oracle) -> None: with pytest.raises((RuntimeError, ValueError, OSError)): _verify_inversion_session_with_authorities( root, loaded_config=loaded, adapter=adapter, oracle=oracle, ) @pytest.fixture def live_authorities(tmp_path: Path): root = tmp_path / "authorities" root.mkdir(mode=0o755) config = root / "inversion.yaml" config.write_bytes(CONFIG.read_bytes()) base_manifest = root / "base.json" base_manifest.write_text("{}\n", encoding="utf-8") snapshot, diffusers, phase0 = (root / name for name in ("snapshot", "diffusers", "phase0")) for directory in (snapshot, diffusers, phase0): directory.mkdir(mode=0o755) directory.chmod(0o755) for path in (config, base_manifest): path.chmod(0o644) return { "config": config, "snapshot": snapshot, "base_manifest": base_manifest, "diffusers": diffusers, "phase0": phase0, } def _public_verifier_without_model_load(monkeypatch: pytest.MonkeyPatch, loaded, adapter, oracle): """Make the public path gate observable without a GPU/model dependency.""" sentinel = object() monkeypatch.setattr(inversion, "load_inversion_config", lambda path: loaded) monkeypatch.setattr(inversion, "load_phase0_vocoder_oracle", lambda root, kind: oracle) monkeypatch.setattr(inversion, "load_frozen_vocoder", lambda **kwargs: adapter) monkeypatch.setattr(inversion.torch.cuda, "is_available", lambda: True) monkeypatch.setattr(inversion.torch.cuda, "get_device_name", lambda device: "NVIDIA H100") monkeypatch.setattr(adapter, "to", lambda **kwargs: adapter) monkeypatch.setattr(inversion, "_verify_inversion_session_with_authorities", lambda *args, **kwargs: sentinel) return sentinel def _public_call(authorities: dict[str, Path], *, output: Path) -> object: return verify_inversion_session( output, config_path=authorities["config"], snapshot=authorities["snapshot"], base_manifest=authorities["base_manifest"], diffusers_root=authorities["diffusers"], phase0_artifacts=authorities["phase0"], ) def test_public_verifier_requires_all_five_live_authorities(monkeypatch, live_authorities, canonical_bundle, tmp_path): _root, loaded, adapter, oracle, _built = canonical_bundle sentinel = _public_verifier_without_model_load(monkeypatch, loaded, adapter, oracle) assert _public_call(live_authorities, output=tmp_path / "output") is sentinel for name in AUTHORITY_NAMES: missing = dict(live_authorities) missing[name] = tmp_path / f"missing-{name}" with pytest.raises((RuntimeError, ValueError, FileNotFoundError, OSError)): _public_call(missing, output=tmp_path / "output") @pytest.mark.parametrize("fail", (False, True)) def test_public_verifier_restores_caller_determinism_state( monkeypatch, live_authorities, canonical_bundle, tmp_path, fail: bool ) -> None: _root, loaded, adapter, oracle, _built = canonical_bundle sentinel = _public_verifier_without_model_load( monkeypatch, loaded, adapter, oracle ) def observed_verify(*args, **kwargs): assert inversion.torch.are_deterministic_algorithms_enabled() assert not inversion.torch.backends.cudnn.benchmark if fail: raise RuntimeError("injected verifier failure") return sentinel monkeypatch.setattr( inversion, "_verify_inversion_session_with_authorities", observed_verify, ) original_enabled = ( inversion.torch.are_deterministic_algorithms_enabled() ) original_warn_only = ( inversion.torch.is_deterministic_algorithms_warn_only_enabled() ) original_benchmark = inversion.torch.backends.cudnn.benchmark try: inversion.torch.use_deterministic_algorithms(False) inversion.torch.backends.cudnn.benchmark = True if fail: with pytest.raises(RuntimeError, match="injected verifier failure"): _public_call(live_authorities, output=tmp_path / "output") else: assert ( _public_call(live_authorities, output=tmp_path / "output") is sentinel ) assert not inversion.torch.are_deterministic_algorithms_enabled() assert inversion.torch.backends.cudnn.benchmark finally: inversion.torch.backends.cudnn.benchmark = original_benchmark inversion.torch.use_deterministic_algorithms( original_enabled, warn_only=original_warn_only ) @pytest.mark.parametrize("name", AUTHORITY_NAMES) def test_public_verifier_rejects_symlinked_authority_path_components(monkeypatch, live_authorities, canonical_bundle, tmp_path, name): _root, loaded, adapter, oracle, _built = canonical_bundle _public_verifier_without_model_load(monkeypatch, loaded, adapter, oracle) bridge = tmp_path / f"bridge-{name}" bridge.symlink_to(live_authorities[name].parent, target_is_directory=True) hostile = dict(live_authorities) hostile[name] = bridge / live_authorities[name].name with pytest.raises((RuntimeError, ValueError, OSError)): _public_call(hostile, output=tmp_path / "output") @pytest.mark.parametrize("name", ("config", "base_manifest")) def test_public_verifier_rejects_hardlinked_authoritative_files(monkeypatch, live_authorities, canonical_bundle, tmp_path, name): _root, loaded, adapter, oracle, _built = canonical_bundle _public_verifier_without_model_load(monkeypatch, loaded, adapter, oracle) hardlink = tmp_path / f"hardlinked-{name}" os.link(live_authorities[name], hardlink) hostile = dict(live_authorities) hostile[name] = hardlink with pytest.raises((RuntimeError, ValueError, OSError)): _public_call(hostile, output=tmp_path / "output") @pytest.mark.parametrize("name", AUTHORITY_NAMES) def test_public_verifier_rejects_unsafe_writable_authorities(monkeypatch, live_authorities, canonical_bundle, tmp_path, name): _root, loaded, adapter, oracle, _built = canonical_bundle _public_verifier_without_model_load(monkeypatch, loaded, adapter, oracle) live_authorities[name].chmod(0o775 if live_authorities[name].is_dir() else 0o664) with pytest.raises((RuntimeError, ValueError, OSError)): _public_call(live_authorities, output=tmp_path / "output") def test_public_verifier_rejects_lexical_traversal_authority_path(monkeypatch, live_authorities, canonical_bundle, tmp_path): _root, loaded, adapter, oracle, _built = canonical_bundle _public_verifier_without_model_load(monkeypatch, loaded, adapter, oracle) hostile = dict(live_authorities) hostile["phase0"] = live_authorities["phase0"].parent / "phase0" / ".." / "phase0" with pytest.raises((RuntimeError, ValueError, OSError)): _public_call(hostile, output=tmp_path / "output") def test_session_rejects_symlinked_experiment_component(bundle, tmp_path): root, loaded, adapter, oracle = bundle displaced = tmp_path / "real-e1" (root / "P2-E1").replace(displaced) (root / "P2-E1").symlink_to(displaced, target_is_directory=True) _reject(root, loaded, adapter, oracle) def test_session_rejects_hardlinked_authoritative_manifest(bundle): root, loaded, adapter, oracle = bundle source = root / "P2-E1" / "manifest.json" linked = root / "P2-E1" / "manifest-linked.json" os.link(source, linked) _reject(root, loaded, adapter, oracle) @pytest.mark.parametrize("path,mode", [("session.json", 0o664), ("P2-E1", 0o775)]) def test_session_rejects_unsafe_writable_artifact_modes(bundle, path, mode): root, loaded, adapter, oracle = bundle (root / path).chmod(mode) _reject(root, loaded, adapter, oracle) @pytest.mark.parametrize("mutation", ("replace", "in_place")) def test_session_rejects_atomic_replacement_or_in_place_mutation_during_verification(bundle, monkeypatch, mutation): root, loaded, adapter, oracle = bundle target = root / "session.json" original = target.read_bytes() fired = False def race(event: str, label: str, relative: str) -> None: nonlocal fired if fired or event != "after_file_read" or label != "inversion session manifest": return fired = True if mutation == "replace": replacement = target.with_name("replacement.json") replacement.write_bytes(original) replacement.chmod(0o644) os.replace(replacement, target) else: target.write_bytes(original + b" ") target.chmod(0o644) monkeypatch.setattr(fd_io, "_TEST_RACE_HOOK", race) _reject(root, loaded, adapter, oracle) assert fired @pytest.mark.parametrize("member", ("run_id", "case_name", "oracle_descriptor", "config_file_sha256")) def test_session_rejects_swapped_run_case_oracle_or_config_authority(bundle, member): root, loaded, adapter, oracle = bundle manifest = root / "P2-E1" / "manifest.json" content = manifest.read_text(encoding="utf-8") replacements = { "run_id": (oracle.descriptor.run_id, "0" * 64), "case_name": (oracle.descriptor.case_name, "foreign_case"), "oracle_descriptor": ("expected_audio_content_sha256", "foreign_audio_content_sha256"), "config_file_sha256": (loaded.file_sha256, "f" * 64), } old, new = replacements[member] assert old in content manifest.write_text(content.replace(old, new, 1), encoding="utf-8") manifest.chmod(0o644) _reject(root, loaded, adapter, oracle) @pytest.mark.parametrize("kind", ("missing", "extra", "traversal", "foreign_key")) def test_session_rejects_missing_extra_traversal_or_foreign_artifact_grammar(bundle, kind): root, loaded, adapter, oracle = bundle leaf = root / "P2-E2" if kind == "missing": (leaf / "target.wav").rename(leaf / "target.wav.removed") elif kind == "extra": (leaf / "unexpected.bin").write_bytes(b"not governed") (leaf / "unexpected.bin").chmod(0o644) else: manifest = leaf / "manifest.json" content = manifest.read_text(encoding="utf-8") if kind == "traversal": content = content.replace('"audio.safetensors"', '"../audio.safetensors"', 1) else: content = content.replace('"artifacts":{', '"artifacts":{"foreign":{"path":"foreign.bin","sha256":"' + "0" * 64 + '","size":0},', 1) manifest.write_text(content, encoding="utf-8") manifest.chmod(0o644) _reject(root, loaded, adapter, oracle) def test_publication_writes_manifests_last_and_leaves_no_partial_output_on_rename_failure(monkeypatch, canonical_bundle, tmp_path): _source, loaded, adapter, oracle, built = canonical_bundle writes = [] real_write = inversion.atomic_write_bytes def observed_write(path, data, *, mode): writes.append((path.parent.name, path.name)) return real_write(path, data, mode=mode) monkeypatch.setattr(inversion, "atomic_write_bytes", observed_write) successful = tmp_path / "manifest-order" publish_inversion_session( output_root=successful, loaded_config=loaded, adapter=adapter, oracle=oracle, experiments=built, ) assert writes[-1][1] == "session.json" for experiment_id in ("P2-E1", "P2-E2"): positions = [ index for index, (parent, _name) in enumerate(writes) if parent == experiment_id ] manifest_position = next( index for index, (parent, name) in enumerate(writes) if parent == experiment_id and name == "manifest.json" ) assert manifest_position == max(positions) real_replace = inversion.os.replace rename_failure = tmp_path / "rename-failure" def fail_root_rename(source, target): if Path(target) == rename_failure: raise OSError("injected root rename failure") return real_replace(source, target) monkeypatch.setattr(inversion.os, "replace", fail_root_rename) with pytest.raises(OSError, match="injected root rename"): publish_inversion_session( output_root=rename_failure, loaded_config=loaded, adapter=adapter, oracle=oracle, experiments=built, ) assert not os.path.lexists(rename_failure) assert not tuple(tmp_path.glob(f".{rename_failure.name}.*.tmp")) monkeypatch.setattr(inversion.os, "replace", real_replace) fsync_failure = tmp_path / "fsync-failure" real_fsync_directory = inversion._fsync_directory failed = False def fail_first_post_rename_fsync(path): nonlocal failed if Path(path) == tmp_path and fsync_failure.exists() and not failed: failed = True raise OSError("injected parent fsync failure") return real_fsync_directory(path) monkeypatch.setattr( inversion, "_fsync_directory", fail_first_post_rename_fsync ) with pytest.raises(OSError, match="injected parent fsync"): publish_inversion_session( output_root=fsync_failure, loaded_config=loaded, adapter=adapter, oracle=oracle, experiments=built, ) assert failed assert not os.path.lexists(fsync_failure) assert not tuple(tmp_path.glob(f".{fsync_failure.name}.*.tmp"))