tests/test_inversion_path_authority.py

"""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"))