Spaces:

asnannp
/

DocDoeAI

Sleeping

DocDoeAI / tests /test_hf_binary_transport.py

deploy: sync backend to Space root (learn-lesson HF cache fix)

65d68f5 25 days ago

9.93 kB

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

	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