left4me/l4d2web/tests/test_l4d2_facade.py

from datetime import UTC, datetime
from pathlib import Path

import pytest

from l4d2web.app import create_app
from l4d2web.auth import hash_password
from l4d2web.db import init_db, session_scope
from l4d2web.models import (
    Blueprint,
    BlueprintOverlay,
    Overlay,
    OverlayWorkshopItem,
    Server,
    User,
    WorkshopItem,
)
from l4d2web.services.host_commands import CommandResult


@pytest.fixture
def server_with_blueprint(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'facade.db'}"
    monkeypatch.setenv("DATABASE_URL", db_url)
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
    init_db()

    with app.app_context():
        with session_scope() as session:
            user = User(username="alice", password_digest=hash_password("secret"), admin=False)
            session.add(user)
            session.flush()

            overlay = Overlay(name="Standard Overlay", path="standard", type="workshop", user_id=user.id)
            session.add(overlay)
            session.flush()

            blueprint = Blueprint(
                user_id=user.id,
                name="default",
                arguments='["-tickrate 100"]',
                config='["sv_consistency 1"]',
            )
            session.add(blueprint)
            session.flush()

            session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay.id, position=0))

            server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
            session.add(server)
            session.flush()
            server_id = server.id
            user_id = user.id

    return server_id, user_id


def test_initialize_uses_l4d2ctl_with_latest_blueprint_data(
    monkeypatch: pytest.MonkeyPatch,
    server_with_blueprint,
) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))
        spec_path = Path(cmd[cmd.index("-f") + 1])
        spec = spec_path.read_text()
        assert "sv_consistency 1" in spec
        assert "standard" in spec
        assert "Standard Overlay" not in spec
        return CommandResult(returncode=0, stdout="", stderr="")

    monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
    monkeypatch.setattr(
        "l4d2web.services.l4d2_facade.initialize_instance",
        lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.initialize_instance directly"),
        raising=False,
    )

    from l4d2web.services.l4d2_facade import initialize_server

    server_id, _ = server_with_blueprint
    initialize_server(server_id)

    assert calls[0][:3] == ["l4d2ctl", "initialize", "alpha"]
    assert calls[0][3] == "-f"


def test_install_and_lifecycle_commands_use_l4d2ctl(
    monkeypatch: pytest.MonkeyPatch,
    server_with_blueprint,
) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))
        return CommandResult(returncode=0, stdout="", stderr="")

    monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
    for name in ["SteamInstaller", "start_instance", "stop_instance", "delete_instance"]:
        monkeypatch.setattr(
            f"l4d2web.services.l4d2_facade.{name}",
            lambda *args, **kwargs: pytest.fail(f"facade must not call l4d2host {name} directly"),
            raising=False,
        )

    from l4d2web.services.l4d2_facade import delete_server, install_runtime, start_server, stop_server

    server_id, _ = server_with_blueprint
    install_runtime()
    start_server(server_id)
    stop_server(server_id)
    delete_server(server_id)

    assert calls == [
        ["l4d2ctl", "install"],
        ["l4d2ctl", "start", "alpha"],
        ["l4d2ctl", "stop", "alpha"],
        ["l4d2ctl", "delete", "alpha"],
    ]


def test_server_status_parses_l4d2ctl_json(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        del kwargs
        calls.append(list(cmd))
        return CommandResult(
            returncode=0,
            stdout='{"state":"running","raw_active_state":"active","raw_sub_state":"running"}',
            stderr="",
        )

    monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
    monkeypatch.setattr(
        "l4d2web.services.l4d2_facade.get_instance_status",
        lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.get_instance_status directly"),
        raising=False,
    )

    from l4d2web.services.l4d2_facade import server_status

    status = server_status("alpha")

    assert calls == [["l4d2ctl", "status", "alpha", "--json"]]
    assert status.state == "running"
    assert status.raw_active_state == "active"
    assert status.raw_sub_state == "running"


def test_server_logs_stream_l4d2ctl_logs(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[list[str]] = []

    def fake_stream_command(cmd):
        calls.append(list(cmd))
        return iter(["one", "two"])

    monkeypatch.setattr("l4d2web.services.host_commands.stream_command", fake_stream_command)
    monkeypatch.setattr(
        "l4d2web.services.l4d2_facade.stream_instance_logs",
        lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.stream_instance_logs directly"),
        raising=False,
    )

    from l4d2web.services.l4d2_facade import stream_server_logs

    lines = list(stream_server_logs("alpha", lines=10, follow=False))

    assert calls == [["l4d2ctl", "logs", "alpha", "--lines", "10", "--no-follow"]]
    assert lines == ["one", "two"]


def _attach_workshop_overlay_to_blueprint(
    server_id: int, user_id: int, *, item_cached: bool, tmp_path: Path
) -> tuple[int, str]:
    """Add a workshop overlay with a single workshop item to the server's
    blueprint. Returns (overlay_id, steam_id)."""
    with session_scope() as session:
        server = session.query(Server).filter_by(id=server_id).one()
        overlay = Overlay(name="ws", path="placeholder", type="workshop", user_id=user_id)
        session.add(overlay)
        session.flush()
        # Path matches id, like the production create_overlay flow does.
        overlay.path = str(overlay.id)
        wi = WorkshopItem(
            steam_id="1001",
            title="A",
            filename="a.vpk",
            file_url="https://example.com/a.vpk",
            file_size=3,
            time_updated=1700000000,
            last_downloaded_at=datetime.now(UTC) if item_cached else None,
        )
        session.add(wi)
        session.flush()
        session.add(
            BlueprintOverlay(blueprint_id=server.blueprint_id, overlay_id=overlay.id, position=1)
        )
        session.add(OverlayWorkshopItem(overlay_id=overlay.id, workshop_item_id=wi.id))
        overlay_id = overlay.id

    if item_cached:
        cache_root = tmp_path / "workshop_cache"
        cache_root.mkdir(exist_ok=True)
        (cache_root / "1001.vpk").write_bytes(b"abc")

    return overlay_id, "1001"


def test_initialize_runs_overlay_builders_synchronously(
    monkeypatch: pytest.MonkeyPatch, server_with_blueprint, tmp_path
) -> None:
    server_id, user_id = server_with_blueprint
    overlay_id, _steam_id = _attach_workshop_overlay_to_blueprint(
        server_id, user_id, item_cached=True, tmp_path=tmp_path
    )

    monkeypatch.setattr(
        "l4d2web.services.host_commands.run_command",
        lambda *args, **kwargs: CommandResult(returncode=0, stdout="", stderr=""),
    )

    from l4d2web.services.l4d2_facade import initialize_server

    initialize_server(server_id)

    addons = tmp_path / "overlays" / str(overlay_id) / "left4dead2" / "addons"
    assert (addons / "1001.vpk").is_symlink(), "workshop builder must run before l4d2ctl initialize"


def test_initialize_fails_fast_on_uncached_workshop_items(
    monkeypatch: pytest.MonkeyPatch, server_with_blueprint, tmp_path
) -> None:
    server_id, user_id = server_with_blueprint
    overlay_id, steam_id = _attach_workshop_overlay_to_blueprint(
        server_id, user_id, item_cached=False, tmp_path=tmp_path
    )

    invocations: list[list[str]] = []

    def fake_run_command(cmd, **kwargs):
        invocations.append(list(cmd))
        return CommandResult(returncode=0, stdout="", stderr="")

    monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)

    from l4d2web.services.l4d2_facade import initialize_server

    with pytest.raises(Exception) as excinfo:
        initialize_server(server_id)

    msg = str(excinfo.value)
    assert steam_id in msg
    assert str(overlay_id) in msg or "ws" in msg
    # l4d2ctl initialize MUST NOT run when uncached items are present.
    assert all("initialize" not in cmd for cmd in invocations), invocations