evidence/source/e3_rollout_v3/scripts/test_publish_hf.py

"""
CPU mutation checks for the Hugging Face publication gate, no network calls.

Matches test_release_audit.py's own strategy: hand-build minimal, schema-correct
fixtures and test the real validate_* / ensure_* / verify_* functions directly,
rather than reconstructing a full audit through release_audit.py's CLI. The
network-touching functions (`ensure_empty_remote`, `verify_remote_release`) are
tested against a small fake standing in for `HfApi`, never a real Hub call.

This closes a gap named explicitly in HANDOFF.md: `publish_hf.py` was read in
full and judged sound on inspection, but a read-through is not a test, which is
this project's own stated standard. This is the test.
"""

import hashlib
import json
import os
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from hf_metadata import RELEASE_FILES, file_sha256  # noqa: E402
import publish_hf as publish_hf_module  # noqa: E402
from publish_hf import (  # noqa: E402
    AUDIT_SCHEMA_VERSION,
    REQUIRED_AUDIT_ARTIFACTS,
    _require_sha256,
    ensure_empty_remote,
    load_json_snapshot,
    validate_publish_inputs,
    verify_remote_release,
)


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def write(path, content):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    mode = "wb" if isinstance(content, bytes) else "w"
    with open(path, mode) as f:
        f.write(content)
    return path


def write_json(path, value):
    return write(path, json.dumps(value))


def build_export(root, repo_id, source_checkpoint):
    """A minimal, schema-correct export directory: the ten release files plus
    a matching export_manifest.json (schema_version 3, release_complete)."""
    for name in RELEASE_FILES:
        # verify_export_manifest requires the rendered card to actually name
        # the repo_id and to contain neither template placeholder.
        content = (
            f"dummy content for {name}, repo {repo_id}" if name == "README.md"
            else f"dummy content for {name}"
        )
        write(os.path.join(root, name), content)
    files = {
        name: {
            "bytes": os.path.getsize(os.path.join(root, name)),
            "sha256": file_sha256(os.path.join(root, name)),
        }
        for name in RELEASE_FILES
    }
    manifest = {
        "schema_version": 3,
        "repo_id": repo_id,
        "release_complete": True,
        "evaluation_sources": {
            key: {"sha256": "a" * 64}
            for key in (
                "validation", "acceptance_comparison",
                "format_ablation", "rollout",
            )
        },
        "model_card_template_sha256": "b" * 64,
        "source_checkpoint": source_checkpoint,
        "files": files,
    }
    return write_json(os.path.join(root, "export_manifest.json"), manifest)


def build_checkpoint(root, step):
    """A minimal checkpoint directory with the three canonical files."""
    meta = {
        "step": step,
        "config": {},
        "model_args": {},
        "optimizer_state_included": True,
    }
    meta_path = write_json(os.path.join(root, "meta.json"), meta)
    write(os.path.join(root, "master.safetensors"), b"master-bytes")
    write(os.path.join(root, "optimizer.safetensors"), b"optimizer-bytes")
    return {
        "path": root,
        "step": step,
        "meta_sha256": file_sha256(meta_path),
        "master_sha256": file_sha256(os.path.join(root, "master.safetensors")),
        "optimizer_sha256": file_sha256(os.path.join(root, "optimizer.safetensors")),
    }


def build_fixture(tmp):
    """A complete, internally consistent export + audit pair."""
    repo_id = "test-namespace/wisp-quant-fixture"
    export_dir = os.path.join(tmp, "export")
    run1_dir = os.path.join(tmp, "run1_ckpt")
    run2_dir = os.path.join(tmp, "run2_ckpt")

    run1 = build_checkpoint(run1_dir, step=19073)
    run2 = build_checkpoint(run2_dir, step=19073)
    source_checkpoint = {
        "step": run1["step"],
        "meta_sha256": run1["meta_sha256"],
        "master_sha256": run1["master_sha256"],
        "optimizer_sha256": run1["optimizer_sha256"],
    }
    manifest_path = build_export(export_dir, repo_id, source_checkpoint)
    manifest_sha256 = file_sha256(manifest_path)

    audit_source = os.path.join(
        os.path.dirname(os.path.abspath(__file__)), "release_audit.py"
    )
    artifact_dir = os.path.join(tmp, "artifacts")
    artifacts = {}
    for name in sorted(REQUIRED_AUDIT_ARTIFACTS):
        path = write_json(
            os.path.join(artifact_dir, f"{name}.json"), {"artifact": name}
        )
        artifacts[name] = {"path": path, "sha256": file_sha256(path)}
    # export_manifest must point at the real manifest already on disk.
    artifacts["export_manifest"] = {
        "path": manifest_path,
        "sha256": manifest_sha256,
    }

    audit = {
        "schema_version": AUDIT_SCHEMA_VERSION,
        "publication_ready": True,
        "research_outcomes_are_not_release_gates": True,
        "audit_source_sha256": file_sha256(audit_source),
        "export": {
            "path": export_dir,
            "repo_id": repo_id,
            "manifest_sha256": manifest_sha256,
        },
        "checkpoint": run1,
        "ablation_checkpoint": run2,
        "artifacts": artifacts,
        "rollout_verification": {
            "path": artifacts["rollout_verification"]["path"],
            "sha256": artifacts["rollout_verification"]["sha256"],
            "branch_quality_independently_verified": True,
            "rollout_execution_reproduced": True,
            "policy_documents": 600,
            "output_tokens": 38400,
            "exact_argmax_tokens": 38399,
            "certified_near_tie_tokens": 1,
            "provenance_scope": (
                publish_hf_module.V3_ATTESTATION_PROVENANCE_SCOPE
            ),
        },
    }
    audit_path = write_json(os.path.join(tmp, "audit.json"), audit)
    return export_dir, audit_path, repo_id, audit


def test_happy_path_succeeds():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, _ = build_fixture(tmp)
        evidence = validate_publish_inputs(export_dir, audit_path, repo_id)
        assert set(evidence["files"]) == set(RELEASE_FILES) | {
            "export_manifest.json"
        }
        assert evidence["export_dir"] == os.path.abspath(export_dir)


def expect_raises(fn, label):
    """The functions under test refuse with ValueError for a bad audit/export
    fact and FileExistsError specifically for a non-empty remote repository
    (matching publish_hf.py's own choice, not a test artifact); both count as
    a refusal here."""
    try:
        fn()
    except (ValueError, FileExistsError):
        return
    raise AssertionError(f"expected a refusal for: {label}")


def test_mutations_are_all_rejected():
    """Each mutation matches test_release_audit.py's style: change exactly one
    fact, confirm validate_publish_inputs refuses it, restore, move on."""
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)

        def with_audit(mutator, label):
            mutated = json.loads(json.dumps(audit))
            mutator(mutated)
            path = os.path.join(tmp, "mutated_audit.json")
            write_json(path, mutated)
            expect_raises(
                lambda: validate_publish_inputs(export_dir, path, repo_id),
                label,
            )

        with_audit(
            lambda a: a.__setitem__("schema_version", 1),
            "wrong schema_version",
        )
        with_audit(
            lambda a: a.__setitem__("publication_ready", False),
            "publication_ready false",
        )
        with_audit(
            lambda a: a.pop("research_outcomes_are_not_release_gates"),
            "missing outcome policy",
        )
        with_audit(
            lambda a: a.__setitem__("audit_source_sha256", "0" * 64),
            "audit code changed since the decision",
        )
        with_audit(
            lambda a: a["export"].__setitem__("repo_id", "someone/else"),
            "audit points at a different repo_id",
        )
        with_audit(
            lambda a: a["export"].__setitem__("manifest_sha256", "1" * 64),
            "audit manifest hash differs",
        )
        with_audit(
            lambda a: a["checkpoint"].__setitem__("master_sha256", "2" * 64),
            "run 1 checkpoint changed",
        )
        with_audit(
            lambda a: a["ablation_checkpoint"].__setitem__(
                "master_sha256", "3" * 64
            ),
            "run 2 checkpoint changed",
        )
        with_audit(
            lambda a: a["artifacts"].pop(sorted(REQUIRED_AUDIT_ARTIFACTS)[0]),
            "missing a required artifact",
        )
        with_audit(
            lambda a: a["artifacts"][sorted(REQUIRED_AUDIT_ARTIFACTS)[0]]
            .__setitem__("sha256", "4" * 64),
            "an audited artifact file changed",
        )
        with_audit(
            lambda a: a.pop("rollout_verification"),
            "missing independent rollout verification result",
        )
        with_audit(
            lambda a: a["rollout_verification"].__setitem__(
                "rollout_execution_reproduced", False
            ),
            "false independent rollout reproduction verdict",
        )
        with_audit(
            lambda a: a["rollout_verification"].__setitem__(
                "sha256", "5" * 64
            ),
            "independent rollout result differs from its artifact",
        )

        # A requested repo_id that does not match the audited one.
        expect_raises(
            lambda: validate_publish_inputs(
                export_dir, audit_path, "wrong/repo-id"
            ),
            "requested repo_id does not match the audit",
        )

        # The export manifest itself claims a different repo_id than requested.
        manifest_path = os.path.join(export_dir, "export_manifest.json")
        manifest = json.loads(open(manifest_path).read())
        original_manifest_repo = manifest["repo_id"]
        manifest["repo_id"] = "someone/else"
        write_json(manifest_path + ".tmp", manifest)
        os.replace(manifest_path + ".tmp", manifest_path)
        expect_raises(
            lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
            "export manifest repo_id changed on disk",
        )
        manifest["repo_id"] = original_manifest_repo
        write_json(manifest_path + ".tmp", manifest)
        os.replace(manifest_path + ".tmp", manifest_path)

        # A release file replaced with a symlink must be refused outright.
        target = os.path.join(export_dir, RELEASE_FILES[0])
        real_target = os.path.join(tmp, "elsewhere.txt")
        write(real_target, "elsewhere")
        os.unlink(target)
        os.symlink(real_target, target)
        expect_raises(
            lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
            "release file is a symlink",
        )


class FakeInfo:
    def __init__(self, sha, private):
        self.sha = sha
        self.private = private


class FakeEntry:
    def __init__(self, path, size, sha256=None, lfs_sha256=None):
        self.path = path
        self.size = size
        self.lfs = None
        if lfs_sha256 is not None:
            self.lfs = type("Lfs", (), {"sha256": lfs_sha256})()
        self._sha256 = sha256


# publish_hf.py checks isinstance(entry, RepoFile) to separate files from
# directories in a repo tree listing. Subclassing the real RepoFile, rather
# than a lookalike class, keeps that isinstance check honest under mutation:
# a fake that merely duck-typed the same attributes would still pass a
# structural check but silently stop testing the real code's isinstance
# branch the day RepoFile's shape changes.
from huggingface_hub.hf_api import RepoFile  # noqa: E402


class FakeRepoFile(RepoFile, FakeEntry):
    def __init__(self, path, size, sha256=None, lfs_sha256=None):
        FakeEntry.__init__(self, path, size, sha256, lfs_sha256)


class FakeApi:
    def __init__(self, existing=None, tree=None):
        self.existing = existing
        self.tree = tree or []
        self.created = []

    def repo_info(self, repo_id, repo_type):
        if self.existing is None:
            import httpx
            from huggingface_hub.errors import RepositoryNotFoundError
            response = httpx.Response(
                404, request=httpx.Request("GET", "https://example.invalid")
            )
            raise RepositoryNotFoundError(repo_id, response=response)
        return self.existing

    def create_repo(self, repo_id, repo_type, private, exist_ok):
        self.created.append((repo_id, private))
        self.existing = FakeInfo(sha="deadbeef", private=private)

    def list_repo_tree(self, repo_id, recursive, expand, repo_type, revision=None):
        return self.tree


def test_ensure_empty_remote_creates_when_missing():
    api = FakeApi(existing=None, tree=[])
    info = ensure_empty_remote(api, "ns/repo", private=True)
    assert api.created == [("ns/repo", True)]
    assert info.private is True


def test_ensure_empty_remote_accepts_existing_empty_matching_visibility():
    api = FakeApi(existing=FakeInfo(sha="x", private=True), tree=[])
    info = ensure_empty_remote(api, "ns/repo", private=True)
    assert info.sha == "x"
    assert api.created == []


def test_ensure_empty_remote_rejects_visibility_mismatch():
    api = FakeApi(existing=FakeInfo(sha="x", private=False), tree=[])
    expect_raises(
        lambda: ensure_empty_remote(api, "ns/repo", private=True),
        "existing repo has the wrong visibility",
    )


def test_ensure_empty_remote_rejects_nonempty_repo():
    api = FakeApi(
        existing=FakeInfo(sha="x", private=True),
        tree=[FakeRepoFile("model.safetensors", 10)],
    )
    expect_raises(
        lambda: ensure_empty_remote(api, "ns/repo", private=True),
        "existing repo already has payload files",
    )


def test_ensure_empty_remote_ignores_gitattributes():
    api = FakeApi(
        existing=FakeInfo(sha="x", private=True),
        tree=[FakeRepoFile(".gitattributes", 10)],
    )
    info = ensure_empty_remote(api, "ns/repo", private=True)
    assert info.sha == "x"


def test_verify_remote_release_accepts_matching_lfs_hash():
    expected = {"model.safetensors": {"bytes": 10, "sha256": "a" * 64}}
    api = FakeApi(
        tree=[FakeRepoFile("model.safetensors", 10, lfs_sha256="a" * 64)]
    )
    verified = verify_remote_release(api, "ns/repo", "rev", expected)
    assert verified["model.safetensors"]["verification"] == "remote_lfs_sha256"


def test_verify_remote_release_downloads_when_no_lfs_hash():
    with tempfile.TemporaryDirectory() as tmp:
        local = write(os.path.join(tmp, "downloaded"), "hello")
        expected = {
            "small.json": {"bytes": os.path.getsize(local), "sha256": file_sha256(local)}
        }
        api = FakeApi(tree=[FakeRepoFile("small.json", os.path.getsize(local))])
        verified = verify_remote_release(
            api, "ns/repo", "rev", expected,
            download_file=lambda *a, **k: local,
        )
        assert verified["small.json"]["verification"] == "downloaded_sha256"


def test_verify_remote_release_rejects_size_mismatch():
    expected = {"model.safetensors": {"bytes": 10, "sha256": "a" * 64}}
    api = FakeApi(
        tree=[FakeRepoFile("model.safetensors", 999, lfs_sha256="a" * 64)]
    )
    expect_raises(
        lambda: verify_remote_release(api, "ns/repo", "rev", expected),
        "remote size does not match the expected size",
    )


def test_verify_remote_release_rejects_hash_mismatch():
    expected = {"model.safetensors": {"bytes": 10, "sha256": "a" * 64}}
    api = FakeApi(
        tree=[FakeRepoFile("model.safetensors", 10, lfs_sha256="b" * 64)]
    )
    expect_raises(
        lambda: verify_remote_release(api, "ns/repo", "rev", expected),
        "remote hash does not match the expected hash",
    )


def test_verify_remote_release_rejects_missing_file():
    expected = {
        "model.safetensors": {"bytes": 10, "sha256": "a" * 64},
        "config.json": {"bytes": 5, "sha256": "b" * 64},
    }
    api = FakeApi(
        tree=[FakeRepoFile("model.safetensors", 10, lfs_sha256="a" * 64)]
    )
    expect_raises(
        lambda: verify_remote_release(api, "ns/repo", "rev", expected),
        "remote is missing an expected file",
    )


def test_verify_remote_release_rejects_unexpected_file():
    expected = {"model.safetensors": {"bytes": 10, "sha256": "a" * 64}}
    api = FakeApi(
        tree=[
            FakeRepoFile("model.safetensors", 10, lfs_sha256="a" * 64),
            FakeRepoFile("extra.bin", 3, lfs_sha256="c" * 64),
        ]
    )
    expect_raises(
        lambda: verify_remote_release(api, "ns/repo", "rev", expected),
        "remote has an unexpected extra file",
    )


def expect_message(fn, expected_substring):
    """Unlike expect_raises, checks the exact guard fired -- several checks
    in this file were found shadowed by an earlier, unrelated ValueError
    from hf_metadata.verify_export_manifest firing first on the same
    mutated fixture, undetectable by a type-only check."""
    try:
        fn()
    except (ValueError, FileExistsError) as exc:
        assert expected_substring in str(exc), (
            f"expected {expected_substring!r} in {exc}"
        )
        return
    raise AssertionError(f"expected a refusal containing {expected_substring!r}")


def test_load_json_snapshot_rejects_malformed_input():
    with tempfile.TemporaryDirectory() as tmp:
        malformed = write(os.path.join(tmp, "malformed.json"), "{not valid json")
        expect_message(lambda: load_json_snapshot(malformed), "invalid JSON")

        non_object = write(os.path.join(tmp, "non_object.json"), "[1, 2, 3]")
        expect_message(
            lambda: load_json_snapshot(non_object),
            "top-level JSON must be an object",
        )


def test_require_sha256_rejects_bad_format():
    expect_message(
        lambda: _require_sha256("NOT-LOWERCASE-HEX" + "0" * 46, "a value"),
        "is not a lowercase SHA-256",
    )
    expect_message(
        lambda: _require_sha256("too-short", "a value"),
        "is not a lowercase SHA-256",
    )


def test_ablation_checkpoint_identity_guards():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)

        def with_audit(mutator, expected_substring):
            mutated = json.loads(json.dumps(audit))
            mutator(mutated)
            path = os.path.join(tmp, "mutated_audit.json")
            write_json(path, mutated)
            expect_message(
                lambda: validate_publish_inputs(export_dir, path, repo_id),
                expected_substring,
            )

        with_audit(
            lambda a: a.__setitem__("ablation_checkpoint", "not-a-dict"),
            "run 2 checkpoint evidence is missing",
        )
        with_audit(
            lambda a: a["ablation_checkpoint"].__setitem__(
                "path", "/nonexistent/checkpoint/dir"
            ),
            "run 2 checkpoint path is missing",
        )


def test_export_release_completeness_and_identity_guards():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)
        manifest_path = os.path.join(export_dir, "export_manifest.json")

        def with_manifest(mutator):
            manifest = json.loads(open(manifest_path).read())
            mutator(manifest)
            write_json(manifest_path + ".tmp", manifest)
            os.replace(manifest_path + ".tmp", manifest_path)

        # A development (incomplete) export must not pass as a release.
        with_manifest(lambda m: (
            m.__setitem__("release_complete", False),
            m.__setitem__("evaluation_sources", None),
        ))
        expect_message(
            lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
            "export is not a final release package",
        )
        with_manifest(lambda m: (
            m.__setitem__("release_complete", True),
            m.__setitem__("evaluation_sources", {
                key: {"sha256": "a" * 64}
                for key in (
                    "validation", "acceptance_comparison",
                    "format_ablation", "rollout",
                )
            }),
        ))

        # The manifest's own repo_id differs from what was requested. The
        # rendered README must still mention the *new* repo_id (or
        # hf_metadata.verify_export_manifest's own card-check fires first),
        # and the manifest's own recorded README hash/size must be updated
        # to match the edited content (or its own artifact-hash check fires
        # first instead) -- both found by trying the naive single-field
        # mutation and watching an earlier, unrelated guard fire.
        other_repo_id = "someone/else"
        readme_path = os.path.join(export_dir, "README.md")
        original_readme = open(readme_path).read()
        edited_readme = original_readme + f"\nalso mentions {other_repo_id}\n"
        write(readme_path, edited_readme)
        with_manifest(lambda m: (
            m.__setitem__("repo_id", other_repo_id),
            m["files"].__setitem__("README.md", {
                "bytes": os.path.getsize(readme_path),
                "sha256": file_sha256(readme_path),
            }),
        ))
        expect_message(
            lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
            f"export repo_id {other_repo_id!r} != requested {repo_id!r}",
        )
        write(readme_path, original_readme)
        with_manifest(lambda m: (
            m.__setitem__("repo_id", repo_id),
            m["files"].__setitem__("README.md", {
                "bytes": os.path.getsize(readme_path),
                "sha256": file_sha256(readme_path),
            }),
        ))

        expect_message(
            lambda: validate_publish_inputs(
                export_dir,
                write_json(
                    os.path.join(tmp, "wrong_export_path.json"),
                    {**audit, "export": {**audit["export"], "path": "/elsewhere"}},
                ),
                repo_id,
            ),
            "release audit points to a different export directory",
        )

        expect_message(
            lambda: validate_publish_inputs(
                export_dir,
                write_json(
                    os.path.join(tmp, "wrong_checkpoint_step.json"),
                    {
                        **audit,
                        "checkpoint": {**audit["checkpoint"], "step": 1},
                    },
                ),
                repo_id,
            ),
            "export source checkpoint differs from release audit",
        )


def test_audit_artifact_guards():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)
        some_label = next(
            label for label in REQUIRED_AUDIT_ARTIFACTS
            if label != "export_manifest"
        )

        malformed = json.loads(json.dumps(audit))
        malformed["artifacts"][some_label] = "not-a-dict"
        path = write_json(os.path.join(tmp, "malformed_artifact.json"), malformed)
        expect_message(
            lambda: validate_publish_inputs(export_dir, path, repo_id),
            f"audit artifact {some_label} is malformed",
        )

        missing = json.loads(json.dumps(audit))
        missing["artifacts"][some_label]["path"] = "/nonexistent/artifact.json"
        path = write_json(os.path.join(tmp, "missing_artifact.json"), missing)
        expect_message(
            lambda: validate_publish_inputs(export_dir, path, repo_id),
            f"audit artifact {some_label} is missing",
        )

        # The export_manifest artifact entry must point at the *real* manifest
        # path, not merely a file with identical content/hash. Mutating just
        # the recorded sha256 would trip the generic per-artifact hash check
        # (a different, earlier guard) instead of this one, so this uses a
        # byte-identical duplicate at a different path to isolate it.
        manifest_path = os.path.join(export_dir, "export_manifest.json")
        duplicate_path = os.path.join(tmp, "export_manifest_duplicate.json")
        with open(manifest_path, "rb") as src, open(duplicate_path, "wb") as dst:
            dst.write(src.read())
        wrong_location = json.loads(json.dumps(audit))
        wrong_location["artifacts"]["export_manifest"] = {
            "path": duplicate_path,
            "sha256": file_sha256(duplicate_path),
        }
        path = write_json(
            os.path.join(tmp, "wrong_manifest_artifact_location.json"),
            wrong_location,
        )
        expect_message(
            lambda: validate_publish_inputs(export_dir, path, repo_id),
            "audited export-manifest artifact differs",
        )


def test_release_package_file_safety_guards():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)

        # A release file replaced with a symlink to a file with IDENTICAL
        # content must still be refused outright, not just when its content
        # also happens to differ (which hf_metadata.verify_export_manifest's
        # own hash check -- an earlier, unrelated guard -- would already
        # catch on its own, found by trying a content-differing symlink
        # first and watching that check fire instead of this one).
        target_name = RELEASE_FILES[0]
        target = os.path.join(export_dir, target_name)
        with open(target, "rb") as f:
            original_content = f.read()
        identical_copy = os.path.join(tmp, "identical_copy")
        write(identical_copy, original_content)
        os.unlink(target)
        os.symlink(identical_copy, target)
        expect_message(
            lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
            f"release package file is a symbolic link: {target_name}",
        )
        os.unlink(target)
        write(target, original_content)

        # The "escapes export root" check (line 204-205) is unreachable
        # through this function under any normal construction: by the time
        # it runs, the preceding islink check has already rejected any
        # symlink, and RELEASE_FILES entries are hardcoded flat filenames
        # with no path separators, so a real (non-symlink) file built via
        # os.path.join(export_dir, name) always has
        # dirname(realpath(path)) == realpath(export_dir) by construction --
        # there is no legitimate input that reaches this branch, only a
        # TOCTOU race (export_dir's own path resolving differently between
        # the two realpath calls), which is not worth simulating for a
        # defensive check with no real trigger. Confirmed via mutation_audit.py:
        # every other guard in this file is caught; this is the sole holdout.

        # hf_metadata.verify_export_manifest never calls os.path.isfile for
        # the release-file loop (only os.path.getsize/file_sha256/os.listdir),
        # so lying to os.path.isfile for exactly one target file reaches this
        # module's own "is missing" check without disturbing the earlier
        # manifest verification pass.
        config_name = "config.json"
        config_path = os.path.abspath(os.path.join(export_dir, config_name))
        real_isfile = os.path.isfile

        def lying_isfile(path):
            if os.path.abspath(path) == config_path:
                return False
            return real_isfile(path)

        os.path.isfile = lying_isfile
        try:
            expect_message(
                lambda: validate_publish_inputs(export_dir, audit_path, repo_id),
                f"release package file is missing: {config_name}",
            )
        finally:
            os.path.isfile = real_isfile


def test_publication_receipt_and_pending_journal_guards():
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)
        real_argv = sys.argv

        out_path = os.path.join(tmp, "publication.json")
        write(out_path, "already published")
        try:
            sys.argv = [
                "publish_hf.py",
                "--export", export_dir,
                "--audit", audit_path,
                "--repo-id", repo_id,
                "--public",
                "--out", out_path,
            ]
            expect_message(
                publish_hf_module.main,
                "publication receipt already exists",
            )
        finally:
            sys.argv = real_argv

        fresh_out_path = os.path.join(tmp, "fresh_publication.json")
        pending_path = fresh_out_path + ".pending"
        write(pending_path, "unresolved journal from a prior attempt")
        try:
            sys.argv = [
                "publish_hf.py",
                "--export", export_dir,
                "--audit", audit_path,
                "--repo-id", repo_id,
                "--public",
                "--out", fresh_out_path,
            ]
            expect_message(
                publish_hf_module.main,
                "unresolved publication journal exists",
            )
        finally:
            sys.argv = real_argv
            os.unlink(pending_path)


def test_main_writes_failure_state_and_reraises_on_upload_error():
    """Line 455's bare `raise` in main()'s except block: on any exception
    during upload, write a "failed" pending journal, then re-raise rather
    than swallowing it. Reached by monkeypatching HfApi itself so no real
    Hub call is ever made."""
    with tempfile.TemporaryDirectory() as tmp:
        export_dir, audit_path, repo_id, audit = build_fixture(tmp)
        out_path = os.path.join(tmp, "publication.json")
        pending_path = out_path + ".pending"

        class ExplodingApi:
            def repo_info(self, repo_id, repo_type):
                raise RuntimeError("simulated Hub failure, no network involved")

        real_hf_api = publish_hf_module.HfApi
        real_argv = sys.argv
        publish_hf_module.HfApi = ExplodingApi
        try:
            sys.argv = [
                "publish_hf.py",
                "--export", export_dir,
                "--audit", audit_path,
                "--repo-id", repo_id,
                "--public",
                "--out", out_path,
            ]
            try:
                publish_hf_module.main()
            except RuntimeError as exc:
                assert "simulated Hub failure" in str(exc)
            else:
                raise AssertionError(
                    "an upload-time exception was swallowed instead of "
                    "re-raised"
                )
        finally:
            publish_hf_module.HfApi = real_hf_api
            sys.argv = real_argv

        assert os.path.isfile(pending_path), (
            "the pending journal should still exist after a failed publish"
        )
        with open(pending_path) as f:
            state = json.load(f)
        assert state["status"] == "failed"
        os.unlink(pending_path)


def main():
    tests = [
        test_happy_path_succeeds,
        test_mutations_are_all_rejected,
        test_load_json_snapshot_rejects_malformed_input,
        test_require_sha256_rejects_bad_format,
        test_ablation_checkpoint_identity_guards,
        test_export_release_completeness_and_identity_guards,
        test_audit_artifact_guards,
        test_release_package_file_safety_guards,
        test_publication_receipt_and_pending_journal_guards,
        test_main_writes_failure_state_and_reraises_on_upload_error,
        test_ensure_empty_remote_creates_when_missing,
        test_ensure_empty_remote_accepts_existing_empty_matching_visibility,
        test_ensure_empty_remote_rejects_visibility_mismatch,
        test_ensure_empty_remote_rejects_nonempty_repo,
        test_ensure_empty_remote_ignores_gitattributes,
        test_verify_remote_release_accepts_matching_lfs_hash,
        test_verify_remote_release_downloads_when_no_lfs_hash,
        test_verify_remote_release_rejects_size_mismatch,
        test_verify_remote_release_rejects_hash_mismatch,
        test_verify_remote_release_rejects_missing_file,
        test_verify_remote_release_rejects_unexpected_file,
    ]
    for test in tests:
        test()
        print(f"  {test.__name__}: PASS")
    print("\nRESULT: publish_hf.py's validation and remote-verification logic checks out")


if __name__ == "__main__":
    main()