sand-boxer/tests/test_bwrap.py

"""ext.bwrap — local namespace isolation extension."""

from unittest.mock import MagicMock, patch

import pytest

from sandboxer.extensions.base import SandboxExtension
from sandboxer.extensions.bwrap import BwrapExtension
from sandboxer.models import IsolationSpec, Profile


def _profile() -> Profile:
    return Profile.model_validate(
        {
            "id": "profile.bwrap-local",
            "version": "1.0.0",
            "extension": "ext.bwrap",
            "isolation": {"level": "process"},
        }
    )


def test_is_sandbox_extension_subclass() -> None:
    assert issubclass(BwrapExtension, SandboxExtension)


def test_process_is_a_valid_isolation_level() -> None:
    assert IsolationSpec(level="process").level == "process"


def test_bwrap_bin_env_override(monkeypatch) -> None:
    monkeypatch.setenv("SANDBOXER_BWRAP_BIN", "/custom/bwrap")
    ext = BwrapExtension({"bwrap_bin": "bwrap"})
    assert ext._bwrap_bin() == "/custom/bwrap"


def test_bwrap_argv_unshares_net_and_binds_workspace(tmp_path) -> None:
    ext = BwrapExtension({"ro_binds": []})
    argv = ext._bwrap_argv(str(tmp_path))
    assert "--unshare-net" in argv
    assert "--unshare-user" in argv
    assert str(tmp_path) in argv
    assert argv[-2:] == ["sleep", "infinity"]


def test_existing_ro_binds_filters_missing_paths(tmp_path) -> None:
    real_dir = tmp_path / "real"
    real_dir.mkdir()
    ext = BwrapExtension({"ro_binds": [str(real_dir), "/definitely/does/not/exist"]})
    assert ext._existing_ro_binds() == [str(real_dir)]


def test_provision_spawns_bwrap_and_returns_handle(tmp_path) -> None:
    ext = BwrapExtension({"base_dir": str(tmp_path)})
    fake_proc = MagicMock()
    fake_proc.pid = 12345

    with patch("sandboxer.extensions.bwrap.subprocess.Popen", return_value=fake_proc) as popen:
        handle = ext.provision(_profile(), {"sandbox_id": "abc12345"}, "localhost")

    popen.assert_called_once()
    assert handle["sandbox_id"] == "abc12345"
    assert handle["host"] == "localhost"
    assert handle["pid"] == "12345"
    assert handle["workspace_dir"].endswith("abc12345")


def test_provision_copies_repo_into_workspace(tmp_path) -> None:
    repo = tmp_path / "repo"
    repo.mkdir()
    (repo / "file.txt").write_text("hello")
    base_dir = tmp_path / "sandboxes"

    ext = BwrapExtension({"base_dir": str(base_dir)})
    fake_proc = MagicMock()
    fake_proc.pid = 1

    with patch("sandboxer.extensions.bwrap.subprocess.Popen", return_value=fake_proc):
        handle = ext.provision(
            _profile(), {"sandbox_id": "cafe1234", "repo": str(repo)}, "localhost"
        )

    assert (base_dir / "cafe1234" / "file.txt").read_text() == "hello"
    assert handle["workspace_dir"] == str(base_dir / "cafe1234")


def test_provision_missing_repo_raises(tmp_path) -> None:
    ext = BwrapExtension({"base_dir": str(tmp_path)})
    with pytest.raises(FileNotFoundError):
        ext.provision(
            _profile(),
            {"sandbox_id": "x", "repo": str(tmp_path / "missing")},
            "localhost",
        )


def test_wait_ready_checks_process_and_workspace(tmp_path) -> None:
    workspace = tmp_path / "ws"
    workspace.mkdir()
    ext = BwrapExtension()
    handle = {"pid": str(1), "workspace_dir": str(workspace), "host": "localhost"}

    with patch.object(BwrapExtension, "_pid_alive", return_value=True):
        result = ext.wait_ready(handle)

    assert result["host"] == "localhost"
    assert result["endpoint"] == "pid:1"


def test_wait_ready_raises_if_process_dead(tmp_path) -> None:
    workspace = tmp_path / "ws"
    workspace.mkdir()
    ext = BwrapExtension()
    handle = {"pid": "999999", "workspace_dir": str(workspace), "host": "localhost"}

    with patch.object(BwrapExtension, "_pid_alive", return_value=False):
        with pytest.raises(RuntimeError, match="not running"):
            ext.wait_ready(handle)


def test_wait_ready_raises_if_workspace_missing(tmp_path) -> None:
    ext = BwrapExtension()
    handle = {"pid": "1", "workspace_dir": str(tmp_path / "gone"), "host": "localhost"}

    with patch.object(BwrapExtension, "_pid_alive", return_value=True):
        with pytest.raises(RuntimeError, match="workspace missing"):
            ext.wait_ready(handle)


def test_teardown_kills_process_group_and_removes_workspace(tmp_path) -> None:
    workspace = tmp_path / "ws"
    workspace.mkdir()
    ext = BwrapExtension()
    handle = {"pid": "42", "workspace_dir": str(workspace), "host": "localhost"}

    with (
        patch.object(BwrapExtension, "_pid_alive", return_value=True),
        patch("sandboxer.extensions.bwrap.os.getpgid", return_value=42),
        patch("sandboxer.extensions.bwrap.os.killpg") as killpg,
    ):
        result = ext.teardown(handle)

    killpg.assert_called_once_with(42, __import__("signal").SIGKILL)
    assert result["process_killed"] == "True"
    assert result["workspace_removed"] == "True"
    assert not workspace.exists()


def test_teardown_handles_already_dead_process(tmp_path) -> None:
    workspace = tmp_path / "ws"
    workspace.mkdir()
    ext = BwrapExtension()
    handle = {"pid": "42", "workspace_dir": str(workspace), "host": "localhost"}

    with patch.object(BwrapExtension, "_pid_alive", return_value=False):
        result = ext.teardown(handle)

    assert result["process_killed"] == "False"
    assert result["workspace_removed"] == "True"


def test_supports_snapshots_is_false() -> None:
    ext = BwrapExtension()
    assert ext.supports_snapshots() is False
    with pytest.raises(NotImplementedError):
        ext.snapshot({})