feat(l4d2-web): add server creation and blueprint reassignment routes

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

@@ -9,6 +9,7 @@ from l4d2web.db import init_db
 from l4d2web.routes.blueprint_routes import bp as blueprint_bp
 from l4d2web.routes.auth_routes import bp as auth_bp
 from l4d2web.routes.overlay_routes import bp as overlay_bp
+from l4d2web.routes.server_routes import bp as server_bp
 
 
 def create_app(test_config: dict[str, object] | None = None) -> Flask:
@@ -24,6 +25,7 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
     app.register_blueprint(auth_bp)
     app.register_blueprint(overlay_bp)
     app.register_blueprint(blueprint_bp)
+    app.register_blueprint(server_bp)
     register_cli(app)
 
     @app.get("/health")

components/l4d2-web-app/src/l4d2web/routes/server_routes.py

@@ -0,0 +1,69 @@
+from flask import Blueprint, Response, jsonify, request
+from sqlalchemy import select
+
+from l4d2web.auth import current_user, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint as BlueprintModel
+from l4d2web.models import Server
+
+
+bp = Blueprint("server", __name__)
+
+
+@bp.post("/servers")
+@require_login
+def create_server() -> Response:
+    user = current_user()
+    assert user is not None
+    payload = request.get_json(silent=True) or {}
+
+    with session_scope() as db:
+        blueprint = db.scalar(
+            select(BlueprintModel).where(
+                BlueprintModel.id == int(payload["blueprint_id"]),
+                BlueprintModel.user_id == user.id,
+            )
+        )
+        if blueprint is None:
+            return Response("blueprint not found", status=404)
+
+        server = Server(
+            user_id=user.id,
+            blueprint_id=blueprint.id,
+            name=str(payload["name"]),
+            port=int(payload["port"]),
+            desired_state="stopped",
+            actual_state="unknown",
+            last_error="",
+        )
+        db.add(server)
+        db.flush()
+        server_id = server.id
+
+    return jsonify({"id": server_id}), 201
+
+
+@bp.patch("/servers/<int:server_id>")
+@require_login
+def update_server(server_id: int) -> Response:
+    user = current_user()
+    assert user is not None
+    payload = request.get_json(silent=True) or {}
+
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
+        if server is None:
+            return Response(status=404)
+
+        blueprint = db.scalar(
+            select(BlueprintModel).where(
+                BlueprintModel.id == int(payload["blueprint_id"]),
+                BlueprintModel.user_id == user.id,
+            )
+        )
+        if blueprint is None:
+            return Response("blueprint not found", status=404)
+
+        server.blueprint_id = blueprint.id
+
+    return jsonify({"id": server_id}), 200

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

@@ -0,0 +1,59 @@
+import json
+
+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, User
+
+
+@pytest.fixture
+def user_client_with_blueprints(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'servers.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+        blueprint_one = Blueprint(user_id=user.id, name="bp1", arguments="[]", config="[]")
+        blueprint_two = Blueprint(user_id=user.id, name="bp2", arguments="[]", config="[]")
+        session.add_all([blueprint_one, blueprint_two])
+        session.flush()
+        payload = {
+            "user_id": user.id,
+            "blueprint_id": blueprint_one.id,
+            "other_blueprint_id": blueprint_two.id,
+        }
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = payload["user_id"]
+
+    return client, payload
+
+
+def test_create_server_from_blueprint(user_client_with_blueprints) -> None:
+    client, data = user_client_with_blueprints
+    payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
+    response = client.post("/servers", data=json.dumps(payload), content_type="application/json")
+    assert response.status_code == 201
+
+
+def test_reassign_blueprint_anytime(user_client_with_blueprints) -> None:
+    client, data = user_client_with_blueprints
+
+    create_payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
+    create_response = client.post("/servers", data=json.dumps(create_payload), content_type="application/json")
+    server_id = create_response.get_json()["id"]
+
+    patch_payload = {"blueprint_id": data["other_blueprint_id"]}
+    response = client.patch(
+        f"/servers/{server_id}",
+        data=json.dumps(patch_payload),
+        content_type="application/json",
+    )
+    assert response.status_code == 200