feat(l4d2-web): add private blueprint CRUD with in-use deletion guard

This commit is contained in:
mwiegand 2026-04-23 01:09:58 +02:00
parent d0614b90fb
commit 896e456513
No known key found for this signature in database
3 changed files with 155 additions and 0 deletions

View file

@ -6,6 +6,7 @@ from l4d2web.auth import load_current_user
from l4d2web.cli import register_cli from l4d2web.cli import register_cli
from l4d2web.config import DEFAULT_CONFIG from l4d2web.config import DEFAULT_CONFIG
from l4d2web.db import init_db 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.auth_routes import bp as auth_bp
from l4d2web.routes.overlay_routes import bp as overlay_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.before_request(load_current_user)
app.register_blueprint(auth_bp) app.register_blueprint(auth_bp)
app.register_blueprint(overlay_bp) app.register_blueprint(overlay_bp)
app.register_blueprint(blueprint_bp)
register_cli(app) register_cli(app)
@app.get("/health") @app.get("/health")

View file

@ -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/<int:blueprint_id>")
@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)

View file

@ -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