diff --git a/components/l4d2-web-app/src/l4d2web/app.py b/components/l4d2-web-app/src/l4d2web/app.py index 96e2456..45a30ff 100644 --- a/components/l4d2-web-app/src/l4d2web/app.py +++ b/components/l4d2-web-app/src/l4d2web/app.py @@ -6,6 +6,7 @@ from l4d2web.auth import load_current_user from l4d2web.cli import register_cli from l4d2web.config import DEFAULT_CONFIG 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 @@ -22,6 +23,7 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask: app.before_request(load_current_user) app.register_blueprint(auth_bp) app.register_blueprint(overlay_bp) + app.register_blueprint(blueprint_bp) register_cli(app) @app.get("/health") diff --git a/components/l4d2-web-app/src/l4d2web/routes/blueprint_routes.py b/components/l4d2-web-app/src/l4d2web/routes/blueprint_routes.py new file mode 100644 index 0000000..0af6df0 --- /dev/null +++ b/components/l4d2-web-app/src/l4d2web/routes/blueprint_routes.py @@ -0,0 +1,75 @@ +import json + +from flask import Blueprint, Response, jsonify, request +from sqlalchemy import delete, func, 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 BlueprintOverlay, Server + + +bp = Blueprint("blueprint", __name__) + + +@bp.post("/blueprints") +@require_login +def create_blueprint() -> Response: + payload = request.get_json(silent=True) or {} + user = current_user() + assert user is not None + + name = str(payload.get("name", "")).strip() + if not name: + return Response("name is required", status=400) + + with session_scope() as db: + blueprint = BlueprintModel( + user_id=user.id, + name=name, + arguments=json.dumps(payload.get("arguments", [])), + config=json.dumps(payload.get("config", [])), + ) + db.add(blueprint) + db.flush() + + for position, overlay_id in enumerate(payload.get("overlay_ids", [])): + db.add( + BlueprintOverlay( + blueprint_id=blueprint.id, + overlay_id=int(overlay_id), + position=position, + ) + ) + + blueprint_id = blueprint.id + + return jsonify({"id": blueprint_id}), 201 + + +@bp.delete("/blueprints/") +@require_login +def delete_blueprint(blueprint_id: int) -> Response: + user = current_user() + assert user is not None + + with session_scope() as db: + blueprint = db.scalar( + select(BlueprintModel).where( + BlueprintModel.id == blueprint_id, + BlueprintModel.user_id == user.id, + ) + ) + if blueprint is None: + return Response(status=404) + + linked_count = db.scalar( + select(func.count(Server.id)).where(Server.blueprint_id == blueprint.id) + ) or 0 + if linked_count > 0: + return Response("blueprint is in use", status=409) + + db.execute(delete(BlueprintOverlay).where(BlueprintOverlay.blueprint_id == blueprint.id)) + db.delete(blueprint) + + return Response(status=204) diff --git a/components/l4d2-web-app/tests/test_blueprints.py b/components/l4d2-web-app/tests/test_blueprints.py new file mode 100644 index 0000000..6eb8aaf --- /dev/null +++ b/components/l4d2-web-app/tests/test_blueprints.py @@ -0,0 +1,78 @@ +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, BlueprintOverlay, Overlay, Server, User + + +@pytest.fixture +def user_client(tmp_path, monkeypatch): + db_url = f"sqlite:///{tmp_path/'blueprint.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() + user_id = user.id + session.add_all( + [ + Overlay(name="o1", path="/opt/l4d2/overlays/o1"), + Overlay(name="o2", path="/opt/l4d2/overlays/o2"), + ] + ) + + client = app.test_client() + with client.session_transaction() as sess: + sess["user_id"] = user_id + return client + + +@pytest.fixture +def linked_blueprint(tmp_path, monkeypatch): + db_url = f"sqlite:///{tmp_path/'linked.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="bob", password_digest=hash_password("secret"), admin=False) + session.add(user) + session.flush() + + blueprint = Blueprint(user_id=user.id, name="linked", arguments="[]", config="[]") + session.add(blueprint) + session.flush() + + session.add(Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)) + blueprint_id = blueprint.id + user_id = user.id + + client = app.test_client() + with client.session_transaction() as sess: + sess["user_id"] = user_id + + return client, blueprint_id + + +def test_user_can_create_private_blueprint(user_client) -> None: + payload = { + "name": "comp", + "arguments": ["-tickrate 100"], + "config": ["sv_consistency 1"], + "overlay_ids": [1, 2], + } + + response = user_client.post("/blueprints", data=json.dumps(payload), content_type="application/json") + assert response.status_code == 201 + + +def test_delete_blueprint_blocked_when_in_use(linked_blueprint) -> None: + client, blueprint_id = linked_blueprint + response = client.delete(f"/blueprints/{blueprint_id}") + assert response.status_code == 409