sand-boxer/tests/test_vm_packer.py

"""VMPackerExtension unit tests."""

from __future__ import annotations

from pathlib import Path
from subprocess import CompletedProcess
from unittest.mock import patch

import pytest

from sandboxer.extensions.vm_packer import VMPackerExtension
from sandboxer.models import Profile


def _profile() -> Profile:
    return Profile.model_validate(
        {
            "id": "profile.vm-haskell-build",
            "version": "1.0.0",
            "extension": "ext.vm-packer",
        }
    )


def _build_profile() -> Profile:
    return Profile.model_validate(
        {
            "id": "profile.vm-packer-build",
            "version": "1.0.0",
            "extension": "ext.vm-packer",
        }
    )


def test_provision_attach_via_alias(tmp_path: Path) -> None:
    repo = tmp_path / "proj"
    repo.mkdir()
    ext = VMPackerExtension()

    with (
        patch("sandboxer.extensions.vm_packer.SSHConfig.run", return_value=(0, "")),
        patch("sandboxer.extensions.vm_packer.SSHConfig.rsync") as rsync,
    ):
        handle = ext.provision(
            _profile(),
            {"vm": "haskell-build", "repo": str(repo)},
            "localhost",
        )

    assert handle["vm_target"] == "haskell-build"
    assert handle["remote_dir"].startswith("/build/sbx-")
    rsync.assert_called_once()


def test_provision_requires_vm_input() -> None:
    ext = VMPackerExtension()
    with pytest.raises(ValueError, match="inputs.vm"):
        ext.provision(_profile(), {}, "localhost")


def test_wait_ready_success() -> None:
    ext = VMPackerExtension()
    handle = {
        "vm_host": "haskell-build",
        "remote_dir": "/build/sbx-abc12345",
        "host": "localhost",
        "ssh_user": "build",
        "ssh_port": "",
    }
    with patch("sandboxer.extensions.vm_packer.SSHConfig.run", return_value=(0, "ready\n")):
        reach = ext.wait_ready(handle)
    assert reach["remote_dir"] == "/build/sbx-abc12345"
    assert "haskell-build" in (reach["ssh"] or "")


def test_teardown_preserves_vm() -> None:
    ext = VMPackerExtension()
    handle = {
        "vm_host": "localhost",
        "remote_dir": "/build/sbx-deadbeef",
        "ssh_user": "build",
        "ssh_port": "12222",
    }
    with patch("sandboxer.extensions.vm_packer.SSHConfig.run", return_value=(0, "")):
        report = ext.teardown(handle)
    assert report["vm_preserved"] == "true"
    assert report["workspace_removed"] == "True"


def test_provision_build_runs_packer(tmp_path: Path) -> None:
    template = tmp_path / "haskell"
    template.mkdir()
    (template / "haskell-build.pkr.hcl").write_text('packer {}')
    ova = template / "haskell-build-20260624.ova"
    ova.write_bytes(b"ova")

    ext = VMPackerExtension()
    with (
        patch(
            "sandboxer.extensions.vm_packer.subprocess.run",
            return_value=CompletedProcess(args=[], returncode=0, stdout="", stderr=""),
        ) as run,
        patch("sandboxer.extensions.vm_packer.emit_progress_note"),
    ):
        handle = ext.provision(
            _build_profile(),
            {"vm_name": "haskell-build", "packer_template": str(template)},
            "localhost",
        )

    assert handle["mode"] == "build"
    assert handle["artifact_path"] == str(ova)
    assert run.call_count == 2
    init_cmd, build_cmd = [c.args[0] for c in run.call_args_list]
    assert init_cmd[:2] == ["packer", "init"]
    assert build_cmd[:2] == ["packer", "build"]
    assert "-var" in build_cmd and "vm_name=haskell-build" in build_cmd


def test_provision_build_requires_template() -> None:
    ext = VMPackerExtension()
    with pytest.raises(ValueError, match="packer_template"):
        ext.provision(_build_profile(), {"vm_name": "haskell-build"}, "localhost")


def test_wait_ready_build_checks_artifact(tmp_path: Path) -> None:
    ova = tmp_path / "haskell-build.ova"
    ova.write_bytes(b"ova")
    ext = VMPackerExtension()
    reach = ext.wait_ready(
        {
            "mode": "build",
            "artifact_path": str(ova),
            "host": "localhost",
        }
    )
    assert reach["endpoint"] == str(ova)


def test_teardown_build_preserves_artifact() -> None:
    ext = VMPackerExtension()
    report = ext.teardown(
        {
            "mode": "build",
            "artifact_path": "/tmp/haskell-build.ova",
        }
    )
    assert report["artifact_preserved"] == "true"
    assert report["workspace_removed"] == "false"