aios/tests/baseline_v1_1/test_package_tools.py

"""Exercise package installation against real temporary Git repositories."""
from pathlib import Path
import importlib.util
import json
import subprocess
import zipfile

import pytest


ROOT = Path(__file__).resolve().parents[2]
SPEC = importlib.util.spec_from_file_location("baseline_bundle", ROOT / "tools/baseline_v1_1/bundle.py")
bundle = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(bundle)


def git(root, *args):
    return subprocess.check_output(["git", "-C", str(root), *args], text=True).strip()


@pytest.fixture
def repos(tmp_path):
    source = tmp_path / "source"
    source.mkdir()
    git(source, "init", "--quiet")
    (source / "app.py").write_text("print('before')\n")
    (source / "retired.txt").write_text("historical\n")
    git(source, "add", ".")
    git(source, "-c", "user.name=Test", "-c", "user.email=test@localhost", "commit", "--quiet", "-m", "base")
    revision = git(source, "rev-parse", "HEAD")
    work = tmp_path / "work"
    subprocess.run(["git", "clone", "--quiet", str(source), str(work)], check=True)
    (work / "app.py").write_text("print('after')\n")
    (work / "retired.txt").unlink()
    (work / "new").mkdir()
    (work / "new/tool.py").write_text("print('new')\n")
    artifact = tmp_path / "change.zip"
    bundle.create_bundle(work, revision, artifact)
    return source, work, revision, artifact


def test_stage_dry_run_install_and_conflict_safe_rollback(repos, tmp_path):
    source, work, revision, artifact = repos
    target = tmp_path / "staged"
    result = bundle.stage_bundle(artifact, source, target, dry_run=True)
    assert result["status"] == "DRY_RUN"
    assert not target.exists()
    assert (source / "app.py").read_text() == "print('before')\n"
    result = bundle.stage_bundle(artifact, source, target)
    assert result["status"] == "STAGED"
    assert (target / "app.py").read_text() == "print('after')\n"
    assert not (target / "retired.txt").exists()
    assert (target / "new/tool.py").is_file()
    assert git(target, "rev-parse", "HEAD") == revision
    with pytest.raises(bundle.PackageError, match="exists"):
        bundle.stage_bundle(artifact, source, target)
    (target / "app.py").write_text("owner edited\n")
    with pytest.raises(bundle.PackageError, match="changed"):
        bundle.rollback(target)
    assert (target / "app.py").read_text() == "owner edited\n"
    assert (target / "new/tool.py").exists()
    (target / "app.py").write_text("print('after')\n")
    assert bundle.rollback(target, dry_run=True)["status"] == "DRY_RUN"
    assert bundle.rollback(target)["status"] == "ROLLED_BACK"
    assert not (target / "new/tool.py").exists()
    assert (target / "retired.txt").read_text() == "historical\n"
    assert not git(target, "status", "--porcelain")


def test_dirty_wrong_base_and_unknown_destination_refused(repos, tmp_path):
    source, work, revision, artifact = repos
    (source / "local.txt").write_text("do not overwrite")
    with pytest.raises(bundle.PackageError, match="clean"):
        bundle.stage_bundle(artifact, source, tmp_path / "one")
    (source / "local.txt").unlink()
    (source / "app.py").write_text("another base")
    git(source, "add", ".")
    git(source, "-c", "user.name=Test", "-c", "user.email=test@localhost", "commit", "--quiet", "-m", "different")
    with pytest.raises(bundle.PackageError, match="baseline"):
        bundle.stage_bundle(artifact, source, tmp_path / "two")
    assert not (tmp_path / "two").exists()


def test_tamper_traversal_extra_entries_and_symlink_refused(repos, tmp_path):
    source, work, revision, artifact = repos
    with zipfile.ZipFile(artifact) as z:
        entries = {name: z.read(name) for name in z.namelist()}
    for index, change in enumerate((
        {"new/app.py": b"tampered"},
        {"../escape": b"bad"},
        {"unlisted.py": b"bad"},
    )):
        bad = tmp_path / f"bad-{index}.zip"
        with zipfile.ZipFile(bad, "w") as z:
            for name, body in (entries | change).items():
                z.writestr(name, body)
        with pytest.raises(bundle.PackageError):
            bundle.verify_bundle(bad)
    (work / "escape.py").symlink_to(source / "app.py")
    with pytest.raises(bundle.PackageError, match="symlink"):
        bundle.create_bundle(work, revision, tmp_path / "links.zip")


def test_rollback_refuses_added_file_symlink_without_touching_referent(repos, tmp_path):
    source, work, revision, artifact = repos
    target = tmp_path / "staged"
    bundle.stage_bundle(artifact, source, target)
    outside = tmp_path / "outside"
    outside.write_text("owner data")
    (target / "new/tool.py").unlink()
    (target / "new/tool.py").symlink_to(outside)
    with pytest.raises(bundle.PackageError, match="symlink"):
        bundle.rollback(target)
    assert outside.read_text() == "owner data"
    assert (target / "app.py").read_text() == "print('after')\n"


def test_manifest_semantic_tampering_is_rejected(repos, tmp_path):
    source, work, revision, artifact = repos
    with zipfile.ZipFile(artifact) as z:
        entries = {name: z.read(name) for name in z.namelist()}
    document = json.loads(entries["manifest.json"])
    document["changes"][0]["before_sha256"] = "0" * 64
    entries["manifest.json"] = json.dumps(document).encode()
    bad = tmp_path / "bad-manifest.zip"
    with zipfile.ZipFile(bad, "w") as z:
        for name, body in entries.items():
            z.writestr(name, body)
    with pytest.raises(bundle.PackageError, match="before|base"):
        bundle.verify_bundle(bad)


def test_rollback_preserves_user_staged_changes(repos, tmp_path):
    source, work, revision, artifact = repos
    target = tmp_path / "staged"
    bundle.stage_bundle(artifact, source, target)
    git(target, "add", "app.py")
    with pytest.raises(bundle.PackageError, match="index changed"):
        bundle.rollback(target)
    assert (target / "app.py").read_text() == "print('after')\n"
    assert git(target, "diff", "--cached", "--name-only") == "app.py"