feat(l4d2-web): resolve live-linked blueprints to runtime specs via l4d2host

components/l4d2-web-app/src/l4d2web/services/l4d2_facade.py

@@ -0,0 +1,77 @@
+import json
+from pathlib import Path
+
+from sqlalchemy import select
+
+from l4d2host.instances import delete_instance, initialize_instance, start_instance, stop_instance
+from l4d2host.logs import stream_instance_logs
+from l4d2host.status import get_instance_status
+from l4d2host.steam_install import SteamInstaller
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint, BlueprintOverlay, Overlay, Server
+from l4d2web.services.spec_yaml import write_temp_spec
+
+
+def build_server_spec_payload(server: Server, blueprint: Blueprint, overlay_names: list[str]) -> dict:
+    return {
+        "port": server.port,
+        "overlays": overlay_names,
+        "arguments": json.loads(blueprint.arguments),
+        "config": json.loads(blueprint.config),
+    }
+
+
+def load_server_blueprint_bundle(server_id: int) -> tuple[Server, Blueprint, list[str]]:
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id))
+        if server is None:
+            raise ValueError("server not found")
+
+        blueprint = db.scalar(select(Blueprint).where(Blueprint.id == server.blueprint_id))
+        if blueprint is None:
+            raise ValueError("blueprint not found")
+
+        rows = db.execute(
+            select(Overlay.name)
+            .join(BlueprintOverlay, BlueprintOverlay.overlay_id == Overlay.id)
+            .where(BlueprintOverlay.blueprint_id == blueprint.id)
+            .order_by(BlueprintOverlay.position)
+        ).all()
+        overlay_names = [row[0] for row in rows]
+        return server, blueprint, overlay_names
+
+
+def install_runtime(on_stdout=None, on_stderr=None) -> None:
+    SteamInstaller().install_or_update(on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def initialize_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, blueprint, overlay_names = load_server_blueprint_bundle(server_id)
+    spec_path = write_temp_spec(build_server_spec_payload(server, blueprint, overlay_names))
+    try:
+        initialize_instance(server.name, spec_path, on_stdout=on_stdout, on_stderr=on_stderr)
+    finally:
+        spec_path.unlink(missing_ok=True)
+
+
+def start_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    start_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def stop_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    stop_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def delete_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    delete_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def server_status(server_name: str):
+    return get_instance_status(server_name)
+
+
+def stream_server_logs(server_name: str, *, lines: int = 200, follow: bool = True):
+    return stream_instance_logs(server_name, lines=lines, follow=follow)

components/l4d2-web-app/src/l4d2web/services/spec_yaml.py

@@ -0,0 +1,13 @@
+import os
+import tempfile
+from pathlib import Path
+
+import yaml
+
+
+def write_temp_spec(payload: dict) -> Path:
+    handle, filename = tempfile.mkstemp(prefix="l4d2-spec-", suffix=".yaml")
+    path = Path(filename)
+    with os.fdopen(handle, "w", encoding="utf-8") as file_obj:
+        file_obj.write(yaml.safe_dump(payload, sort_keys=False))
+    return path

components/l4d2-web-app/tests/test_l4d2_facade.py

@@ -0,0 +1,62 @@
+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, Server, User
+
+
+@pytest.fixture
+def server_with_blueprint(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'facade.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    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", path="/opt/l4d2/overlays/standard")
+            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
+
+    return server_id
+
+
+def test_initialize_uses_latest_blueprint_data(monkeypatch: pytest.MonkeyPatch, server_with_blueprint) -> None:
+    called: dict[str, str] = {}
+
+    def fake_initialize(name, spec_path, **kwargs):
+        del kwargs
+        called["name"] = name
+        called["spec"] = Path(spec_path).read_text()
+
+    monkeypatch.setattr("l4d2web.services.l4d2_facade.initialize_instance", fake_initialize)
+
+    from l4d2web.services.l4d2_facade import initialize_server
+
+    initialize_server(server_with_blueprint)
+
+    assert called["name"] == "alpha"
+    assert "sv_consistency 1" in called["spec"]