From df1ccb4cca502f809fe6b411bdf5bfb6f7376a71 Mon Sep 17 00:00:00 2001 From: mwiegand Date: Thu, 7 May 2026 16:50:54 +0200 Subject: [PATCH] feat(l4d2-web): workshop overlay UI (routes + templates) Adds workshop_routes blueprint with add-items / remove-item / manual- build endpoints plus admin /admin/workshop/refresh. Add-items handles single ID, single URL, multi-line batch, or a collection ID; auto- enqueues a coalesced build_overlay job per call. Reject non-L4D2 items with 400, duplicate associations with friendly toast, intruders with 403. Generalizes overlay_routes: type+name only on create (no path field); external is admin-only and system-wide, workshop is per-user and auto-pathed. Update is name-only. Delete recursively removes the on-disk dir only for managed paths (path == str(id)); legacy externals are left in place. The pre-existing in-use guard is preserved. Page routes filter the overlay listing by user permissions and load workshop items + the latest related job for the detail view. Templates: unified Create modal with type radio (no path field). Type-aware overlay detail: workshop overlays show a multi-line input + items/collection radio + item table partial with thumbnails, manual Rebuild button, and a small status indicator pulled from the latest related job. Admin page gets a "Refresh all workshop items" button. Co-Authored-By: Claude Opus 4.7 (1M context) --- l4d2web/app.py | 2 + l4d2web/routes/overlay_routes.py | 120 ++++++--- l4d2web/routes/page_routes.py | 44 +++- l4d2web/routes/workshop_routes.py | 173 +++++++++++++ l4d2web/templates/_overlay_item_table.html | 50 ++++ l4d2web/templates/admin.html | 9 + l4d2web/templates/overlay_detail.html | 60 ++++- l4d2web/templates/overlays.html | 23 +- l4d2web/tests/test_overlays.py | 160 ++++++++---- l4d2web/tests/test_workshop_routes.py | 279 +++++++++++++++++++++ 10 files changed, 824 insertions(+), 96 deletions(-) create mode 100644 l4d2web/routes/workshop_routes.py create mode 100644 l4d2web/templates/_overlay_item_table.html create mode 100644 l4d2web/tests/test_workshop_routes.py diff --git a/l4d2web/app.py b/l4d2web/app.py index 34375ec..8ffd67a 100644 --- a/l4d2web/app.py +++ b/l4d2web/app.py @@ -16,6 +16,7 @@ from l4d2web.routes.log_routes import bp as log_bp from l4d2web.routes.overlay_routes import bp as overlay_bp from l4d2web.routes.page_routes import bp as page_bp from l4d2web.routes.server_routes import bp as server_bp +from l4d2web.routes.workshop_routes import bp as workshop_bp from l4d2web.services.job_worker import recover_stale_jobs, start_job_workers @@ -69,6 +70,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(workshop_bp) app.register_blueprint(blueprint_bp) app.register_blueprint(server_bp) app.register_blueprint(job_bp) diff --git a/l4d2web/routes/overlay_routes.py b/l4d2web/routes/overlay_routes.py index 3e0d145..7ebedaf 100644 --- a/l4d2web/routes/overlay_routes.py +++ b/l4d2web/routes/overlay_routes.py @@ -1,72 +1,132 @@ +import shutil + from flask import Blueprint, Response, redirect, request from sqlalchemy import select -from l4d2web.auth import require_admin +from l4d2host.paths import get_left4me_root + +from l4d2web.auth import current_user, require_login from l4d2web.db import session_scope from l4d2web.models import BlueprintOverlay, Overlay -from l4d2web.services.security import validate_overlay_ref +from l4d2web.services.overlay_creation import ( + create_overlay_directory, + generate_overlay_path, +) bp = Blueprint("overlay", __name__) -@bp.post("/overlays") -@require_admin -def create_overlay() -> Response: - name = request.form.get("name", "").strip() - raw_path = request.form.get("path", "") - if not name or not raw_path: - return Response("missing fields", status=400) +VALID_TYPES = {"external", "workshop"} - try: - overlay_ref = validate_overlay_ref(raw_path) - except ValueError as exc: - return Response(str(exc), status=400) + +def _is_managed_path(overlay: Overlay) -> bool: + return overlay.path == str(overlay.id) + + +def _can_edit_overlay(overlay: Overlay, user) -> bool: + if user is None: + return False + if user.admin: + return True + if overlay.type == "external": + return False + if overlay.type == "workshop": + return overlay.user_id == user.id + return False + + +def _name_already_taken(db, name: str, scope_user_id: int | None, *, except_id: int | None = None) -> bool: + query = select(Overlay).where(Overlay.name == name) + if scope_user_id is None: + query = query.where(Overlay.user_id.is_(None)) + else: + query = query.where(Overlay.user_id == scope_user_id) + if except_id is not None: + query = query.where(Overlay.id != except_id) + return db.scalar(query) is not None + + +@bp.post("/overlays") +@require_login +def create_overlay() -> Response: + user = current_user() + assert user is not None + + name = request.form.get("name", "").strip() + overlay_type = request.form.get("type", "external").strip().lower() + if not name: + return Response("missing fields", status=400) + if overlay_type not in VALID_TYPES: + return Response(f"unknown overlay type: {overlay_type}", status=400) + + if overlay_type == "external": + if not user.admin: + return Response("admin only", status=403) + scope_user_id: int | None = None + else: # workshop + scope_user_id = user.id with session_scope() as db: - existing = db.scalar(select(Overlay).where(Overlay.name == name)) - if existing is not None: + if _name_already_taken(db, name, scope_user_id): return Response("overlay already exists", status=409) - db.add(Overlay(name=name, path=overlay_ref)) - return redirect("/overlays") + overlay = Overlay(name=name, path="", type=overlay_type, user_id=scope_user_id) + db.add(overlay) + db.flush() + overlay.path = generate_overlay_path(overlay.id) + db.flush() + create_overlay_directory(overlay) + new_id = overlay.id + + return redirect(f"/overlays/{new_id}") @bp.post("/overlays/") -@require_admin +@require_login def update_overlay(overlay_id: int) -> Response: - name = request.form.get("name", "").strip() - raw_path = request.form.get("path", "") - if not name or not raw_path: - return Response("missing fields", status=400) + user = current_user() + assert user is not None - try: - overlay_ref = validate_overlay_ref(raw_path) - except ValueError as exc: - return Response(str(exc), status=400) + name = request.form.get("name", "").strip() + if not name: + return Response("missing fields", status=400) with session_scope() as db: overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id)) if overlay is None: return Response(status=404) - duplicate = db.scalar(select(Overlay).where(Overlay.name == name, Overlay.id != overlay_id)) - if duplicate is not None: + if not _can_edit_overlay(overlay, user): + return Response(status=403) + if _name_already_taken(db, name, overlay.user_id, except_id=overlay_id): return Response("overlay already exists", status=409) overlay.name = name - overlay.path = overlay_ref return redirect(f"/overlays/{overlay_id}") @bp.post("/overlays//delete") -@require_admin +@require_login def delete_overlay(overlay_id: int) -> Response: + user = current_user() + assert user is not None + with session_scope() as db: overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id)) if overlay is None: return Response(status=404) + if not _can_edit_overlay(overlay, user): + return Response(status=403) in_use = db.scalar(select(BlueprintOverlay).where(BlueprintOverlay.overlay_id == overlay_id)) if in_use is not None: return Response("overlay is in use", status=409) + path_value = overlay.path + path_is_managed = _is_managed_path(overlay) db.delete(overlay) + + if path_is_managed and path_value: + target = get_left4me_root() / "overlays" / path_value + if target.exists(): + shutil.rmtree(target) + return redirect("/overlays") diff --git a/l4d2web/routes/page_routes.py b/l4d2web/routes/page_routes.py index bbd3423..16f0f2f 100644 --- a/l4d2web/routes/page_routes.py +++ b/l4d2web/routes/page_routes.py @@ -6,7 +6,15 @@ from sqlalchemy import select from l4d2web.auth import current_user, require_admin, require_login from l4d2web.db import session_scope from l4d2web.models import Blueprint as BlueprintModel -from l4d2web.models import BlueprintOverlay, Job, Overlay, Server, User +from l4d2web.models import ( + BlueprintOverlay, + Job, + Overlay, + OverlayWorkshopItem, + Server, + User, + WorkshopItem, +) bp = Blueprint("pages", __name__) @@ -141,28 +149,60 @@ def server_jobs_page(server_id: int): @bp.get("/overlays") @require_login def overlays() -> str: + user = current_user() + assert user is not None with session_scope() as db: - overlays = db.scalars(select(Overlay).order_by(Overlay.name)).all() + query = select(Overlay).order_by(Overlay.name) + if not user.admin: + query = query.where( + (Overlay.type == "external") | (Overlay.user_id == user.id) + ) + overlays = db.scalars(query).all() return render_template("overlays.html", overlays=overlays) @bp.get("/overlays/") @require_login def overlay_detail(overlay_id: int): + user = current_user() + assert user is not None with session_scope() as db: overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id)) if overlay is None: return Response(status=404) + # Visibility: externals are visible to all; workshop overlays are + # visible to the owner and admins. + if overlay.type == "workshop" and not user.admin and overlay.user_id != user.id: + return Response(status=403) using_blueprints = db.scalars( select(BlueprintModel) .join(BlueprintOverlay, BlueprintOverlay.blueprint_id == BlueprintModel.id) .where(BlueprintOverlay.overlay_id == overlay.id) .order_by(BlueprintModel.name) ).all() + workshop_items = [] + if overlay.type == "workshop": + workshop_items = db.scalars( + select(WorkshopItem) + .join( + OverlayWorkshopItem, + OverlayWorkshopItem.workshop_item_id == WorkshopItem.id, + ) + .where(OverlayWorkshopItem.overlay_id == overlay.id) + .order_by(WorkshopItem.created_at) + ).all() + latest_build_job = db.scalar( + select(Job) + .where(Job.operation == "build_overlay", Job.overlay_id == overlay.id) + .order_by(Job.created_at.desc()) + .limit(1) + ) return render_template( "overlay_detail.html", overlay=overlay, using_blueprints=using_blueprints, + workshop_items=workshop_items, + latest_build_job=latest_build_job, ) diff --git a/l4d2web/routes/workshop_routes.py b/l4d2web/routes/workshop_routes.py new file mode 100644 index 0000000..7685a22 --- /dev/null +++ b/l4d2web/routes/workshop_routes.py @@ -0,0 +1,173 @@ +"""Routes for the workshop overlay type (add/remove items, manual rebuild, +admin global refresh).""" +from __future__ import annotations + +from flask import Blueprint, Response, redirect, render_template, request +from sqlalchemy import delete as sa_delete +from sqlalchemy import select + +from l4d2web.auth import current_user, require_admin, require_login +from l4d2web.db import session_scope +from l4d2web.models import ( + Job, + Overlay, + OverlayWorkshopItem, + WorkshopItem, +) +from l4d2web.services import steam_workshop +from l4d2web.services.job_worker import enqueue_build_overlay + + +bp = Blueprint("workshop", __name__) + + +def _check_workshop_overlay_access(overlay_id: int, user, db): + overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id)) + if overlay is None: + return None, Response(status=404) + if overlay.type != "workshop": + return None, Response("not a workshop overlay", status=400) + if overlay.user_id != user.id and not user.admin: + return None, Response(status=403) + return overlay, None + + +def _render_item_table(overlay_id: int): + with session_scope() as db: + overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id)) + items = db.scalars( + select(WorkshopItem) + .join( + OverlayWorkshopItem, + OverlayWorkshopItem.workshop_item_id == WorkshopItem.id, + ) + .where(OverlayWorkshopItem.overlay_id == overlay_id) + .order_by(WorkshopItem.created_at) + ).all() + # Detach so attributes survive after the session closes. + for item in items: + db.expunge(item) + if overlay is not None: + db.expunge(overlay) + return render_template( + "_overlay_item_table.html", + overlay=overlay, + workshop_items=items, + ) + + +@bp.post("/overlays//items") +@require_login +def add_items(overlay_id: int) -> Response: + user = current_user() + assert user is not None + + raw_input = request.form.get("input", "").strip() + mode = request.form.get("input_mode", "items") + if not raw_input: + return Response("missing input", status=400) + + try: + ids = steam_workshop.parse_workshop_input(raw_input) + except ValueError as exc: + return Response(str(exc), status=400) + + if mode == "collection": + if len(ids) != 1: + return Response("collection mode expects exactly one id or url", status=400) + try: + ids = steam_workshop.resolve_collection(ids[0]) + except Exception as exc: + return Response(f"failed to resolve collection: {exc}", status=502) + if not ids: + return Response("collection has no items", status=400) + + try: + metas = steam_workshop.fetch_metadata_batch(ids, mode="add") + except steam_workshop.WorkshopValidationError as exc: + return Response(str(exc), status=400) + except Exception as exc: + return Response(f"steam api error: {exc}", status=502) + + with session_scope() as db: + overlay, err = _check_workshop_overlay_access(overlay_id, user, db) + if err is not None: + return err + for meta in metas: + wi = db.scalar(select(WorkshopItem).where(WorkshopItem.steam_id == meta.steam_id)) + if wi is None: + wi = WorkshopItem(steam_id=meta.steam_id) + db.add(wi) + wi.title = meta.title + wi.filename = meta.filename + wi.file_url = meta.file_url + wi.file_size = meta.file_size + wi.time_updated = meta.time_updated + wi.preview_url = meta.preview_url + wi.last_error = "" if meta.result == 1 else f"steam result {meta.result}" + db.flush() + + existing = db.scalar( + select(OverlayWorkshopItem).where( + OverlayWorkshopItem.overlay_id == overlay_id, + OverlayWorkshopItem.workshop_item_id == wi.id, + ) + ) + if existing is None: + db.add(OverlayWorkshopItem(overlay_id=overlay_id, workshop_item_id=wi.id)) + + enqueue_build_overlay(db, overlay_id=overlay_id, user_id=user.id) + + return _render_item_table(overlay_id) + + +@bp.post("/overlays//items//delete") +@require_login +def remove_item(overlay_id: int, item_id: int) -> Response: + user = current_user() + assert user is not None + with session_scope() as db: + overlay, err = _check_workshop_overlay_access(overlay_id, user, db) + if err is not None: + return err + result = db.execute( + sa_delete(OverlayWorkshopItem).where( + OverlayWorkshopItem.overlay_id == overlay_id, + OverlayWorkshopItem.workshop_item_id == item_id, + ) + ) + if result.rowcount == 0: + return Response(status=404) + enqueue_build_overlay(db, overlay_id=overlay_id, user_id=user.id) + return _render_item_table(overlay_id) + + +@bp.post("/overlays//build") +@require_login +def manual_build(overlay_id: int) -> Response: + user = current_user() + assert user is not None + with session_scope() as db: + overlay, err = _check_workshop_overlay_access(overlay_id, user, db) + if err is not None: + return err + job = enqueue_build_overlay(db, overlay_id=overlay_id, user_id=user.id) + job_id = job.id + return redirect(f"/jobs/{job_id}") + + +@bp.post("/admin/workshop/refresh") +@require_admin +def admin_refresh() -> Response: + user = current_user() + assert user is not None + with session_scope() as db: + db.add( + Job( + user_id=user.id, + server_id=None, + operation="refresh_workshop_items", + state="queued", + ) + ) + return redirect("/admin/jobs") diff --git a/l4d2web/templates/_overlay_item_table.html b/l4d2web/templates/_overlay_item_table.html new file mode 100644 index 0000000..20d1130 --- /dev/null +++ b/l4d2web/templates/_overlay_item_table.html @@ -0,0 +1,50 @@ +{% set can_edit = g.user.admin or (overlay and overlay.type == 'workshop' and overlay.user_id == g.user.id) %} + + + + + + + + + + + {% if can_edit %}{% endif %} + + + + {% for item in workshop_items %} + + + + + + + + + {% if can_edit %} + + {% endif %} + + {% else %} + + {% endfor %} + +
Steam IDTitleFilenameSizeUpdatedStatus
+ {% if item.preview_url %} + + {% endif %} + {{ item.steam_id }}{{ item.title }}{{ item.filename }}{{ item.file_size }}{{ item.time_updated }} + {% if item.last_error %} + {{ item.last_error }} + {% elif item.last_downloaded_at %} + cached + {% else %} + pending + {% endif %} + +
+ + +
+
No workshop items yet.
diff --git a/l4d2web/templates/admin.html b/l4d2web/templates/admin.html index c7541aa..f0eea62 100644 --- a/l4d2web/templates/admin.html +++ b/l4d2web/templates/admin.html @@ -19,4 +19,13 @@ + +
+

Workshop

+

Re-fetch metadata for every workshop item; re-download those with newer versions on Steam, then enqueue rebuilds for the affected overlays.

+
+ + +
+
{% endblock %} diff --git a/l4d2web/templates/overlay_detail.html b/l4d2web/templates/overlay_detail.html index 439e250..8a5c64a 100644 --- a/l4d2web/templates/overlay_detail.html +++ b/l4d2web/templates/overlay_detail.html @@ -6,30 +6,74 @@

Overlay: {{ overlay.name }}

- {% if g.user.admin %} + {% set can_edit = g.user.admin or (overlay.type == 'workshop' and overlay.user_id == g.user.id) %} + {% if can_edit %} {% endif %}
- {% if g.user.admin %} + {% if can_edit %}
-
- {% else %} + {% endif %} + - - + + +
Name{{ overlay.name }}
Path{{ overlay.path }}
Type{{ overlay.type }}
Scope{% if overlay.user_id %}private{% else %}system{% endif %}
Path{{ overlay.path }}
- {% endif %}
+{% if overlay.type == 'workshop' %} +
+
+

Workshop items

+ {% if can_edit %} +
+ + +
+ {% endif %} +
+ + {% if can_edit %} +
+ +
+ Input mode + + +
+ +
+ +
+
+ {% endif %} + +
+ {% include "_overlay_item_table.html" with context %} +
+
+ +{% if latest_build_job %} +
+

Latest build

+

+ job #{{ latest_build_job.id }} + — state: {{ latest_build_job.state }} +

+
+{% endif %} +{% endif %} +

Used by

{% if using_blueprints %} @@ -43,7 +87,7 @@ {% endif %}
-{% if g.user.admin %} +{% if can_edit %}

Delete overlay "{{ overlay.name }}"?

diff --git a/l4d2web/templates/overlays.html b/l4d2web/templates/overlays.html index 83ee8ef..6cb449c 100644 --- a/l4d2web/templates/overlays.html +++ b/l4d2web/templates/overlays.html @@ -6,43 +6,48 @@

Overlays

- {% if g.user.admin %} - {% endif %}
- + {% for overlay in overlays %} + + {% else %} - + {% endfor %}
NamePath
NameTypeScopePath
{{ overlay.name }}{{ overlay.type }}{% if overlay.user_id %}private{% else %}system{% endif %} {{ overlay.path }}
No overlays configured.
No overlays yet.
-{% if g.user.admin %}
-

Add overlay

+

Create overlay

+
+ Type + + {% if g.user.admin %} + + {% endif %} +
- +

The path is generated automatically.

- +
-{% endif %} {% endblock %} diff --git a/l4d2web/tests/test_overlays.py b/l4d2web/tests/test_overlays.py index 0a2e6bc..1825ba1 100644 --- a/l4d2web/tests/test_overlays.py +++ b/l4d2web/tests/test_overlays.py @@ -10,6 +10,7 @@ from l4d2web.services.security import validate_overlay_ref def admin_client(tmp_path, monkeypatch): db_url = f"sqlite:///{tmp_path/'admin_overlay.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() @@ -30,13 +31,15 @@ def admin_client(tmp_path, monkeypatch): def user_client_with_overlay(tmp_path, monkeypatch): db_url = f"sqlite:///{tmp_path/'user_overlay.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 session_scope() as session: user = User(username="alice", password_digest=hash_password("secret"), admin=False) session.add(user) - session.add(Overlay(name="standard", path="standard")) + # System external overlay (no user_id), pre-existing. + session.add(Overlay(name="standard", path="standard", type="external", user_id=None)) session.flush() user_id = user.id @@ -53,7 +56,8 @@ def test_user_can_view_overlay_catalog(user_client_with_overlay) -> None: assert response.status_code == 200 assert "standard" in text - assert "Add overlay" not in text + # Non-admin users can create workshop overlays, so the Create button shows. + assert "Create overlay" in text def test_admin_can_view_overlay_edit_controls(admin_client) -> None: @@ -61,27 +65,19 @@ def test_admin_can_view_overlay_edit_controls(admin_client) -> None: text = response.get_data(as_text=True) assert response.status_code == 200 - assert "Add overlay" in text + assert "Create overlay" in text assert 'action="/overlays"' in text -def test_admin_can_create_overlay(admin_client) -> None: +def test_admin_can_create_external_overlay(admin_client) -> None: response = admin_client.post( "/overlays", - data={"name": "standard", "path": "standard"}, + data={"name": "standard", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert response.status_code == 302 - assert response.headers["Location"] == "/overlays" - - -def test_overlay_ref_must_be_relative(admin_client) -> None: - response = admin_client.post( - "/overlays", - data={"name": "bad", "path": "/tmp/bad"}, - headers={"X-CSRF-Token": "test-token"}, - ) - assert response.status_code == 400 + # Redirect to the new detail page now that paths are auto-generated. + assert response.headers["Location"].startswith("/overlays/") @pytest.mark.parametrize("overlay_ref", [" standard", "standard ", "a//b", "a/", "./a", "a/.", "."]) @@ -90,71 +86,136 @@ def test_overlay_ref_rejects_unsafe_components(overlay_ref: str) -> None: validate_overlay_ref(overlay_ref) -def test_overlay_route_rejects_whitespace_padded_ref(admin_client) -> None: - response = admin_client.post( - "/overlays", - data={"name": "bad", "path": " standard"}, - headers={"X-CSRF-Token": "test-token"}, - ) - assert response.status_code == 400 - - -def test_non_admin_cannot_create_overlay(user_client_with_overlay) -> None: +def test_non_admin_cannot_create_external_overlay(user_client_with_overlay) -> None: response = user_client_with_overlay.post( "/overlays", - data={"name": "bad", "path": "bad"}, + data={"name": "bad", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert response.status_code == 403 +def test_user_can_create_workshop_overlay(user_client_with_overlay) -> None: + response = user_client_with_overlay.post( + "/overlays", + data={"name": "my-maps", "type": "workshop"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 302 + with session_scope() as session: + overlay = session.query(Overlay).filter_by(name="my-maps").one() + assert overlay.type == "workshop" + assert overlay.user_id is not None + assert overlay.path == str(overlay.id) + + +def test_workshop_overlay_directory_is_created_on_disk(user_client_with_overlay, tmp_path) -> None: + response = user_client_with_overlay.post( + "/overlays", + data={"name": "my-maps", "type": "workshop"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 302 + with session_scope() as session: + overlay = session.query(Overlay).filter_by(name="my-maps").one() + overlay_id = overlay.id + assert (tmp_path / "overlays" / str(overlay_id)).is_dir() + + +def test_two_users_can_have_workshop_overlay_with_same_name(tmp_path, monkeypatch) -> None: + # Set up a fresh app with two users. + db_url = f"sqlite:///{tmp_path/'shared.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 session_scope() as session: + for username in ("alice", "bob"): + session.add(User(username=username, password_digest=hash_password("x"), admin=False)) + session.flush() + alice_id, bob_id = ( + session.query(User).filter_by(username="alice").one().id, + session.query(User).filter_by(username="bob").one().id, + ) + + def client_for(uid): + c = app.test_client() + with c.session_transaction() as sess: + sess["user_id"] = uid + sess["csrf_token"] = "test-token" + return c + + for uid in (alice_id, bob_id): + r = client_for(uid).post( + "/overlays", + data={"name": "my-maps", "type": "workshop"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert r.status_code == 302 + + with session_scope() as session: + rows = session.query(Overlay).filter_by(name="my-maps").all() + assert {r.user_id for r in rows} == {alice_id, bob_id} + + def test_admin_can_update_and_delete_overlay(admin_client) -> None: create = admin_client.post( "/overlays", - data={"name": "standard", "path": "standard"}, + data={"name": "standard", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert create.status_code == 302 + with session_scope() as session: + overlay_id = session.query(Overlay).filter_by(name="standard").one().id + update = admin_client.post( - "/overlays/1", - data={"name": "edited", "path": "edited"}, + f"/overlays/{overlay_id}", + data={"name": "edited"}, headers={"X-CSRF-Token": "test-token"}, ) assert update.status_code == 302 delete = admin_client.post( - "/overlays/1/delete", + f"/overlays/{overlay_id}/delete", headers={"X-CSRF-Token": "test-token"}, ) assert delete.status_code == 302 def test_update_overlay_rejects_duplicate_name(admin_client) -> None: - for name in ["standard", "competitive"]: + ids: list[int] = [] + for name in ("standard", "competitive"): response = admin_client.post( "/overlays", - data={"name": name, "path": name}, + data={"name": name, "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert response.status_code == 302 + with session_scope() as session: + ids = [ + session.query(Overlay).filter_by(name="standard").one().id, + session.query(Overlay).filter_by(name="competitive").one().id, + ] response = admin_client.post( - "/overlays/2", - data={"name": "standard", "path": "competitive"}, + f"/overlays/{ids[1]}", + data={"name": "standard"}, headers={"X-CSRF-Token": "test-token"}, ) - assert response.status_code == 409 def test_overlay_detail_page_lists_using_blueprints(admin_client) -> None: create = admin_client.post( "/overlays", - data={"name": "shared", "path": "shared"}, + data={"name": "shared", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert create.status_code == 302 + with session_scope() as session: + overlay_id = session.query(Overlay).filter_by(name="shared").one().id with session_scope() as session: admin = session.query(User).filter_by(username="admin").one() @@ -162,10 +223,10 @@ def test_overlay_detail_page_lists_using_blueprints(admin_client) -> None: bp_two = Blueprint(user_id=admin.id, name="beta-bp", arguments="[]", config="[]") session.add_all([bp_one, bp_two]) session.flush() - session.add(BlueprintOverlay(blueprint_id=bp_one.id, overlay_id=1, position=0)) - session.add(BlueprintOverlay(blueprint_id=bp_two.id, overlay_id=1, position=0)) + session.add(BlueprintOverlay(blueprint_id=bp_one.id, overlay_id=overlay_id, position=0)) + session.add(BlueprintOverlay(blueprint_id=bp_two.id, overlay_id=overlay_id, position=0)) - response = admin_client.get("/overlays/1") + response = admin_client.get(f"/overlays/{overlay_id}") text = response.get_data(as_text=True) assert response.status_code == 200 @@ -179,7 +240,8 @@ def test_overlay_detail_page_404_when_missing(admin_client) -> None: assert response.status_code == 404 -def test_overlay_detail_hides_edit_for_non_admin(user_client_with_overlay) -> None: +def test_overlay_detail_hides_edit_for_non_admin_external(user_client_with_overlay) -> None: + # The seeded "standard" external overlay (id=1, user_id=NULL) is admin-only edit. response = user_client_with_overlay.get("/overlays/1") text = response.get_data(as_text=True) @@ -192,37 +254,41 @@ def test_overlay_detail_hides_edit_for_non_admin(user_client_with_overlay) -> No def test_overlay_update_redirects_to_detail(admin_client) -> None: create = admin_client.post( "/overlays", - data={"name": "standard", "path": "standard"}, + data={"name": "standard", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert create.status_code == 302 + with session_scope() as session: + overlay_id = session.query(Overlay).filter_by(name="standard").one().id response = admin_client.post( - "/overlays/1", - data={"name": "renamed", "path": "renamed"}, + f"/overlays/{overlay_id}", + data={"name": "renamed"}, headers={"X-CSRF-Token": "test-token"}, ) assert response.status_code == 302 - assert response.headers["Location"] == "/overlays/1" + assert response.headers["Location"] == f"/overlays/{overlay_id}" def test_delete_overlay_rejects_in_use_overlay(admin_client) -> None: create = admin_client.post( "/overlays", - data={"name": "standard", "path": "standard"}, + data={"name": "standard", "type": "external"}, headers={"X-CSRF-Token": "test-token"}, ) assert create.status_code == 302 + with session_scope() as session: + overlay_id = session.query(Overlay).filter_by(name="standard").one().id with session_scope() as session: user = session.query(User).filter_by(username="admin").one() blueprint = Blueprint(user_id=user.id, name="bp", arguments="[]", config="[]") session.add(blueprint) session.flush() - session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=1, position=0)) + session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay_id, position=0)) response = admin_client.post( - "/overlays/1/delete", + f"/overlays/{overlay_id}/delete", headers={"X-CSRF-Token": "test-token"}, ) diff --git a/l4d2web/tests/test_workshop_routes.py b/l4d2web/tests/test_workshop_routes.py new file mode 100644 index 0000000..5bec803 --- /dev/null +++ b/l4d2web/tests/test_workshop_routes.py @@ -0,0 +1,279 @@ +"""Tests for the workshop overlay routes (add items, remove items, build, +admin refresh).""" +from __future__ import annotations + +from typing import Iterable +from unittest.mock import patch + +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 ( + Job, + Overlay, + OverlayWorkshopItem, + User, + WorkshopItem, +) +from l4d2web.services import steam_workshop + + +def _meta(steam_id: str, *, app_id: int = 550, result: int = 1) -> steam_workshop.WorkshopMetadata: + return steam_workshop.WorkshopMetadata( + steam_id=steam_id, + title=f"Item {steam_id}", + filename=f"{steam_id}.vpk", + file_url=f"https://example.com/{steam_id}.vpk", + file_size=42, + time_updated=1700000000, + preview_url=f"https://example.com/preview-{steam_id}.jpg", + consumer_app_id=app_id, + result=result, + ) + + +@pytest.fixture +def env_user(tmp_path, monkeypatch): + db_url = f"sqlite:///{tmp_path/'wr.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 session_scope() as session: + user = User(username="alice", password_digest=hash_password("x"), admin=False) + admin = User(username="admin", password_digest=hash_password("x"), admin=True) + session.add_all([user, admin]) + session.flush() + user_id = user.id + admin_id = admin.id + + def login(uid): + c = app.test_client() + with c.session_transaction() as sess: + sess["user_id"] = uid + sess["csrf_token"] = "test-token" + return c + + return app, login, user_id, admin_id + + +@pytest.fixture +def overlay_for(env_user): + app, login, user_id, admin_id = env_user + + user_client = login(user_id) + response = user_client.post( + "/overlays", + data={"name": "my-maps", "type": "workshop"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 302, response.get_data(as_text=True) + + with session_scope() as session: + overlay = session.query(Overlay).filter_by(name="my-maps").one() + overlay_id = overlay.id + + return app, login, user_id, admin_id, overlay_id + + +def _patch_steam(metas: Iterable[steam_workshop.WorkshopMetadata]): + return patch.object(steam_workshop, "fetch_metadata_batch", return_value=list(metas)) + + +def test_add_single_item_creates_association_and_enqueues_build(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + with _patch_steam([_meta("1001")]): + response = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 200 + assert b"1001" in response.data + + with session_scope() as session: + n_assoc = session.query(OverlayWorkshopItem).count() + assert n_assoc == 1 + wi = session.query(WorkshopItem).filter_by(steam_id="1001").one() + assert wi.title == "Item 1001" + assert wi.preview_url.endswith("preview-1001.jpg") + # Auto-enqueued build_overlay job. + jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all() + assert len(jobs) == 1 + assert jobs[0].state == "queued" + + +def test_add_multiline_batch_coalesces_into_one_build_job(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + with _patch_steam([_meta(s) for s in ("1001", "1002", "1003")]): + response = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001\n1002\n1003", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 200 + + with session_scope() as session: + assert session.query(OverlayWorkshopItem).count() == 3 + jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all() + assert len(jobs) == 1, "multi-item add should coalesce into a single build job" + + +def test_add_collection_resolves_members(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + with patch.object(steam_workshop, "resolve_collection", return_value=["1001", "1002"]) as resolve: + with _patch_steam([_meta("1001"), _meta("1002")]): + response = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "555", "input_mode": "collection"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 200 + resolve.assert_called_once_with("555") + + with session_scope() as session: + assert session.query(OverlayWorkshopItem).count() == 2 + + +def test_add_non_l4d2_item_returns_400(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + def raise_validation(*args, **kwargs): + raise steam_workshop.WorkshopValidationError("not L4D2") + + with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=raise_validation): + response = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "9999", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 400 + assert b"not L4D2" in response.data + + with session_scope() as session: + assert session.query(WorkshopItem).count() == 0 + assert session.query(OverlayWorkshopItem).count() == 0 + + +def test_add_duplicate_item_does_not_500(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + with _patch_steam([_meta("1001")]): + first = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert first.status_code == 200 + + with _patch_steam([_meta("1001")]): + second = user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert second.status_code == 200 + + with session_scope() as session: + assert session.query(OverlayWorkshopItem).count() == 1 + + +def test_remove_item_drops_association_and_enqueues_rebuild(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + with _patch_steam([_meta("1001")]): + user_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + + with session_scope() as session: + wi = session.query(WorkshopItem).filter_by(steam_id="1001").one() + item_id = wi.id + + response = user_client.post( + f"/overlays/{overlay_id}/items/{item_id}/delete", + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 200 + + with session_scope() as session: + assert session.query(OverlayWorkshopItem).count() == 0 + # WorkshopItem itself remains (cache survives the association removal). + assert session.query(WorkshopItem).filter_by(steam_id="1001").one() is not None + # Coalesced into the same queued build_overlay job. + jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all() + assert len(jobs) == 1 + + +def test_manual_build_button_enqueues_job(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + user_client = login(user_id) + + response = user_client.post( + f"/overlays/{overlay_id}/build", + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 302 + assert response.headers["Location"].startswith("/jobs/") + with session_scope() as session: + jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all() + assert len(jobs) == 1 + + +def test_admin_refresh_enqueues_global_job(env_user): + app, login, user_id, admin_id = env_user + admin_client = login(admin_id) + + response = admin_client.post( + "/admin/workshop/refresh", + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 302 + assert response.headers["Location"] == "/admin/jobs" + with session_scope() as session: + jobs = session.query(Job).filter_by(operation="refresh_workshop_items").all() + assert len(jobs) == 1 + assert jobs[0].state == "queued" + + +def test_non_admin_cannot_refresh(env_user): + app, login, user_id, _admin_id = env_user + user_client = login(user_id) + + response = user_client.post( + "/admin/workshop/refresh", + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 403 + + +def test_other_user_cannot_modify_workshop_overlay(overlay_for): + app, login, user_id, _admin_id, overlay_id = overlay_for + + with session_scope() as session: + intruder = User(username="bob", password_digest=hash_password("x"), admin=False) + session.add(intruder) + session.flush() + intruder_id = intruder.id + + intruder_client = login(intruder_id) + response = intruder_client.post( + f"/overlays/{overlay_id}/items", + data={"input": "1001", "input_mode": "items"}, + headers={"X-CSRF-Token": "test-token"}, + ) + assert response.status_code == 403