left4me/l4d2host/tests/test_lifecycle.py

import subprocess
from pathlib import Path

import pytest

from l4d2host.instances import (
    delete_instance,
    initialize_instance,
    start_instance,
    stop_instance,
)


def test_start_order(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))

    instance_dir = tmp_path / "instances" / "alpha"
    runtime_dir = tmp_path / "runtime" / "alpha"
    (runtime_dir / "merged" / "left4dead2" / "cfg").mkdir(parents=True, exist_ok=True)
    instance_dir.mkdir(parents=True, exist_ok=True)
    (instance_dir / "instance.env").write_text(
        "L4D2_PORT=27015\nL4D2_ARGS=-tickrate 100\nL4D2_LOWERDIRS=/x:/y\n"
    )
    (instance_dir / "server.cfg").write_text("sv_consistency 1")

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)

    start_instance("alpha", root=tmp_path)

    assert calls[0] == [
        "sudo",
        "-n",
        "/usr/local/libexec/left4me/left4me-overlay",
        "mount",
        "alpha",
    ]
    assert calls[1] == ["sudo", "-n", "/usr/local/libexec/left4me/left4me-systemctl", "start", "alpha"]


def test_start_refuses_to_double_mount(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))

    instance_dir = tmp_path / "instances" / "alpha"
    runtime_dir = tmp_path / "runtime" / "alpha"
    (runtime_dir / "merged").mkdir(parents=True)
    instance_dir.mkdir(parents=True)
    (instance_dir / "instance.env").write_text("L4D2_PORT=27015\nL4D2_ARGS=\nL4D2_LOWERDIRS=/x\n")
    (instance_dir / "server.cfg").write_text("")

    merged = runtime_dir / "merged"

    def fake_ismount(path):
        return Path(path) == merged

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.instances.os.path.ismount", fake_ismount)

    with pytest.raises(subprocess.CalledProcessError) as exc_info:
        start_instance("alpha", root=tmp_path)

    assert "already mounted" in (exc_info.value.stderr or "")
    assert calls == [], "no mount/start commands must be issued when refusing"


def test_delete_missing_is_noop(tmp_path: Path) -> None:
    delete_instance("missing", root=tmp_path)


def test_delete_succeeds_when_stop_service_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))
        if cmd[:2] == ["sudo", "-n"] and "left4me-systemctl" in cmd[2] and "stop" in cmd:
            raise subprocess.CalledProcessError(
                returncode=5,
                cmd=list(cmd),
                stderr="Unit left4me-server@alpha.service not loaded.",
            )

    (tmp_path / "instances" / "alpha").mkdir(parents=True)
    (tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)

    delete_instance("alpha", root=tmp_path)

    assert not (tmp_path / "instances" / "alpha").exists()
    assert not (tmp_path / "runtime" / "alpha").exists()


@pytest.mark.parametrize("bad_name", ["..", "../escape", "foo/bar", " foo", "Foo"])
def test_lifecycle_rejects_unsafe_instance_names(tmp_path: Path, bad_name: str) -> None:
    for func in (start_instance, stop_instance, delete_instance):
        with pytest.raises(ValueError):
            func(bad_name, root=tmp_path)
    with pytest.raises(ValueError):
        initialize_instance(bad_name, tmp_path / "spec.yaml", root=tmp_path)
    assert not (tmp_path / "instances").exists()
    assert not (tmp_path / "runtime").exists()


def test_delete_stopped_instance_removes_dirs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))

    (tmp_path / "instances" / "alpha").mkdir(parents=True)
    (tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)

    delete_instance("alpha", root=tmp_path)

    assert not (tmp_path / "instances" / "alpha").exists()
    assert not (tmp_path / "runtime" / "alpha").exists()
    assert ["sudo", "-n", "/usr/local/libexec/left4me/left4me-systemctl", "stop", "alpha"] in calls


def test_stop_succeeds_when_unmount_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    umount_calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        if cmd[:4] == [
            "sudo",
            "-n",
            "/usr/local/libexec/left4me/left4me-overlay",
            "umount",
        ]:
            umount_calls.append(list(cmd))
            raise subprocess.CalledProcessError(
                returncode=1,
                cmd=list(cmd),
                stderr="umount: /var/lib/left4me/runtime/alpha/merged: not mounted",
            )

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)

    stop_instance("alpha", root=tmp_path)

    assert umount_calls, "stop must always attempt the overlay helper (no preflight)"


def test_delete_succeeds_when_unmount_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    umount_calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        if cmd[:4] == [
            "sudo",
            "-n",
            "/usr/local/libexec/left4me/left4me-overlay",
            "umount",
        ]:
            umount_calls.append(list(cmd))
            raise subprocess.CalledProcessError(
                returncode=1,
                cmd=list(cmd),
                stderr="umount: /var/lib/left4me/runtime/alpha/merged: not mounted",
            )

    (tmp_path / "instances" / "alpha").mkdir(parents=True)
    (tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)

    monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
    monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)

    delete_instance("alpha", root=tmp_path)

    assert umount_calls, "delete must always attempt the overlay helper (no preflight)"
    assert not (tmp_path / "instances" / "alpha").exists()
    assert not (tmp_path / "runtime" / "alpha").exists()