tests/test_hf_binary_transport.py

from __future__ import annotations

import hashlib
import os
from pathlib import Path
import re
import shutil
import subprocess

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "deploy-backend-to-hf.sh"
ASSET_DIRECTORY = Path("public/docdoe/mascot")
POSES = ("idle", "pointing", "thinking", "wave")


def _transport_function() -> str:
    script = SCRIPT.read_text(encoding="utf-8")
    match = re.search(r"prepare_mascot_lfs\(\) \{.*?\n\}", script, re.DOTALL)
    assert match, "Deploy must prepare mascot binary transport before staging."
    return match.group(0)


def _git(directory: Path, *arguments: str) -> bytes:
    return subprocess.check_output(
        ["git", "-C", str(directory), *arguments], env=_git_environment()
    )


def _git_environment() -> dict[str, str]:
    # Keep fixtures and all offline Git processes free of ambient credentials
    # and machine-wide configuration such as signing or network credentials.
    return {
        **{
            key: value
            for key, value in os.environ.items()
            if key.upper()
            in {"PATH", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP"}
        },
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": os.devnull,
        "GIT_TERMINAL_PROMPT": "0",
        "DEBUG_FIX": "1",
    }


@pytest.fixture
def transport_checkout(tmp_path: Path) -> tuple[Path, str, dict[str, str]]:
    # Use Git Bash on Windows, not the unrelated WSL launcher on PATH.
    git_bash = Path("C:/Program Files/Git/bin/bash.exe")
    bash = (
        str(git_bash) if os.name == "nt" and git_bash.exists() else shutil.which("bash")
    )
    if bash is None or shutil.which("git") is None:
        pytest.fail("Deployment transport tests require Bash and Git.")
    subprocess.run(
        ["git", "lfs", "version"],
        env=_git_environment(),
        check=True,
        capture_output=True,
    )
    directory = tmp_path / "space"
    directory.mkdir()
    _git(directory, "init", "--initial-branch=main")
    assets = directory / ASSET_DIRECTORY
    assets.mkdir(parents=True)
    for pose in POSES:
        shutil.copy2(ROOT / ASSET_DIRECTORY / f"body-{pose}.png", assets)
    (directory / ".gitattributes").write_text(
        "*.safetensors filter=lfs diff=lfs merge=lfs -text\n", encoding="utf-8"
    )
    (directory / "app.py").write_text("print('backend')\n", encoding="utf-8")
    return directory, bash, _git_environment()


def test_mascot_transport_is_prepared_in_space_before_commit() -> None:
    script = SCRIPT.read_text(encoding="utf-8")
    function = _transport_function()
    invocation = script.index("\nprepare_mascot_lfs\n")
    assert (
        script.index('cd "$WORK/space"') < invocation < script.index("\ngit add -A\n")
    )
    assert "git lfs install --local" in function
    assert 'git lfs track "public/docdoe/mascot/*.png"' in function
    assert "git lfs pointer --check --strict --stdin" in function
    # Retain the deliberate tracked-asset exception to the broad media exclusion.
    assert '"public/docdoe/mascot/"' in script
    assert '".png"' in script
    assert 'BACKEND_DIR / "generated" / "learn-anything"' in script


def test_real_staged_mascots_are_pointers_with_original_objects(
    transport_checkout: tuple[Path, str, dict[str, str]],
) -> None:
    directory, bash, environment = transport_checkout
    result = subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n" + _transport_function() + "\nprepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        check=True,
        capture_output=True,
        text=True,
    )
    assert "Verified 4 mascot PNG(s)" in result.stdout
    assert "[FIX:hf-binary-transport]" in result.stdout
    attributes = _git(directory, "show", ":.gitattributes").decode()
    assert "*.safetensors filter=lfs diff=lfs merge=lfs -text" in attributes
    assert (
        "public/docdoe/mascot/*.png filter=lfs diff=lfs merge=lfs -text" in attributes
    )
    for pose in POSES:
        path = ASSET_DIRECTORY / f"body-{pose}.png"
        original = (ROOT / path).read_bytes()
        digest = hashlib.sha256(original).hexdigest()
        pointer = _git(directory, "show", f":{path.as_posix()}").decode()
        assert pointer.startswith("version https://git-lfs.github.com/spec/v1\n")
        assert f"oid sha256:{digest}\n" in pointer
        assert f"size {len(original)}\n" in pointer
        # The clean filter must not replace the working file with a pointer.
        assert (directory / path).read_bytes() == original
        object_path = directory / ".git/lfs/objects" / digest[:2] / digest[2:4] / digest
        assert object_path.read_bytes() == original

    # Only the intended temporary Space assets/attributes are staged by this step.
    assert "app.py" not in _git(directory, "diff", "--cached", "--name-only").decode()
    # Repeated preparations do not accumulate attribute rules or alter objects.
    subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n" + _transport_function() + "\nprepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        check=True,
        capture_output=True,
    )
    assert _git(directory, "show", ":.gitattributes").decode() == attributes

    # Exercise the actual clean/smudge round trip, not just pointer text: a
    # fresh renderer build checkout must receive PNG bytes, never pointer files.
    _git(
        directory,
        "-c",
        "user.name=Local deployment QA",
        "-c",
        "user.email=qa@example.test",
        "commit",
        "-m",
        "Verify local binary transport",
    )
    restored = directory.parent / "restored-space"
    subprocess.run(
        ["git", "clone", "--no-hardlinks", str(directory), str(restored)],
        env=environment,
        check=True,
        capture_output=True,
    )
    # Global filters are deliberately disabled in the isolated test. Install
    # them in the fresh checkout, then exercise the real local LFS transfer.
    for arguments in (("lfs", "install", "--local"), ("lfs", "pull")):
        subprocess.run(
            ["git", "-C", str(restored), *arguments],
            env=environment,
            check=True,
            capture_output=True,
        )
    for pose in POSES:
        path = ASSET_DIRECTORY / f"body-{pose}.png"
        assert (restored / path).read_bytes() == (ROOT / path).read_bytes()


def test_empty_mascot_payload_refuses_release(
    transport_checkout: tuple[Path, str, dict[str, str]],
) -> None:
    directory, bash, environment = transport_checkout
    for pose in POSES:
        (directory / ASSET_DIRECTORY / f"body-{pose}.png").unlink()
    result = subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n" + _transport_function() + "\nprepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert (
        "No mascot PNGs were staged; refusing an incomplete renderer release."
        in result.stderr
    )


def test_missing_lfs_fails_before_staging_plain_binary_blobs(
    transport_checkout: tuple[Path, str, dict[str, str]],
) -> None:
    directory, bash, environment = transport_checkout
    unavailable_lfs = (
        '\ngit() { if [[ "$1" == "lfs" ]]; then return 1; fi; command git "$@"; }\n'
    )
    result = subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n"
            + _transport_function()
            + unavailable_lfs
            + "prepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert "Git LFS is required to upload the renderer mascot PNGs." in result.stderr
    assert _git(directory, "diff", "--cached", "--name-only") == b""


@pytest.mark.parametrize("pose", POSES)
@pytest.mark.parametrize("empty", (False, True))
def test_each_required_pose_must_be_present_and_nonempty(
    transport_checkout: tuple[Path, str, dict[str, str]], pose: str, empty: bool
) -> None:
    directory, bash, environment = transport_checkout
    asset = directory / ASSET_DIRECTORY / f"body-{pose}.png"
    if empty:
        asset.write_bytes(b"")
    else:
        asset.unlink()
    result = subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n" + _transport_function() + "\nprepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert (
        f"Required mascot pose '{pose}' is missing or empty; refusing release."
        in result.stderr
    )
    assert _git(directory, "diff", "--cached", "--name-only") == b""


def test_pointer_verification_refuses_an_unfiltered_staged_blob(
    transport_checkout: tuple[Path, str, dict[str, str]],
) -> None:
    directory, bash, environment = transport_checkout
    # Simulate a broken clean filter by exposing the raw PNG at the index-read
    # boundary. Keep all other real Git/LFS commands and pointer validation.
    raw_index = '\ngit() { if [[ "$1" == "show" ]]; then command cat public/docdoe/mascot/body-idle.png; else command git "$@"; fi; }\n'
    result = subprocess.run(
        [
            bash,
            "-c",
            "set -euo pipefail\n"
            + _transport_function()
            + raw_index
            + "prepare_mascot_lfs",
        ],
        cwd=directory,
        env=environment,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert (
        "Mascot PNG is not staged as an LFS pointer; refusing release." in result.stderr
    )
    assert "Verified 4 mascot PNG(s)" not in result.stdout