cronekorkn/left4me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(l4d2-web): workshop overlay UI (routes + templates)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
mwiegand 2026-05-07 16:50:54 +02:00
parent 38a6fbbe1e
commit df1ccb4cca
No known key found for this signature in database
10 changed files with 824 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
l4d2web/app.py
View file

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

120
l4d2web/routes/overlay_routes.py
View file

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

44
l4d2web/routes/page_routes.py
View file

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

173
l4d2web/routes/workshop_routes.py Normal file
View file

@ -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/<int:overlay_id>/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/<int:overlay_id>/items/<int:item_id>/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/<int:overlay_id>/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")

50
l4d2web/templates/_overlay_item_table.html Normal file
View file

@ -0,0 +1,50 @@
{% set can_edit = g.user.admin or (overlay and overlay.type == 'workshop' and overlay.user_id == g.user.id) %}
<table class="table">
<thead>
<tr>
<th></th>
<th>Steam ID</th>
<th>Title</th>
<th>Filename</th>
<th>Size</th>
<th>Updated</th>
<th>Status</th>
{% if can_edit %}<th></th>{% endif %}
</tr>
</thead>
<tbody>
{% for item in workshop_items %}
<tr>
<td>
{% if item.preview_url %}
<img src="{{ item.preview_url }}" alt="" width="48" height="48" loading="lazy">
{% endif %}
</td>
<td><a href="https://steamcommunity.com/sharedfiles/filedetails/?id={{ item.steam_id }}" target="_blank" rel="noopener">{{ item.steam_id }}</a></td>
<td>{{ item.title }}</td>
<td class="muted">{{ item.filename }}</td>
<td class="muted">{{ item.file_size }}</td>
<td class="muted">{{ item.time_updated }}</td>
<td>
{% if item.last_error %}
<span class="warning">{{ item.last_error }}</span>
{% elif item.last_downloaded_at %}
cached
{% else %}
<span class="muted">pending</span>
{% endif %}
</td>
{% if can_edit %}
<td>
<form method="post" action="/overlays/{{ overlay.id }}/items/{{ item.id }}/delete" class="inline-form">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<button type="submit" class="button-secondary">Remove</button>
</form>
</td>
{% endif %}
</tr>
{% else %}
<tr><td colspan="{% if can_edit %}8{% else %}7{% endif %}" class="muted">No workshop items yet.</td></tr>
{% endfor %}
</tbody>
</table>

9
l4d2web/templates/admin.html
View file

@ -19,4 +19,13 @@
<button type="submit">Install or update runtime</button>
</form>
</section>
<section class="panel">
<h2>Workshop</h2>
<p class="muted">Re-fetch metadata for every workshop item; re-download those with newer versions on Steam, then enqueue rebuilds for the affected overlays.</p>
<form method="post" action="/admin/workshop/refresh">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<button type="submit">Refresh all workshop items</button>
</form>
</section>
{% endblock %}

60
l4d2web/templates/overlay_detail.html
View file

@ -6,30 +6,74 @@
<section class="panel">
<div class="page-heading">
<h1>Overlay: {{ overlay.name }}</h1>
{% if g.user.admin %}
{% set can_edit = g.user.admin or (overlay.type == 'workshop' and overlay.user_id == g.user.id) %}
{% if can_edit %}
<button type="button" class="danger" data-modal-open="delete-overlay-modal">Delete</button>
{% endif %}
</div>
{% if g.user.admin %}
{% if can_edit %}
<form method="post" action="/overlays/{{ overlay.id }}" class="stack">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<label>Name <input name="name" value="{{ overlay.name }}" required></label>
<label>Path <input name="path" value="{{ overlay.path }}" required></label>
<div>
<button type="submit">Save</button>
</div>
</form>
{% else %}
{% endif %}
<table class="definition-table">
<tbody>
<tr><th>Name</th><td>{{ overlay.name }}</td></tr>
<tr><th>Path</th><td>{{ overlay.path }}</td></tr>
<tr><th>Type</th><td>{{ overlay.type }}</td></tr>
<tr><th>Scope</th><td>{% if overlay.user_id %}private{% else %}system{% endif %}</td></tr>
<tr><th>Path</th><td class="muted">{{ overlay.path }}</td></tr>
</tbody>
</table>
{% endif %}
</section>
{% if overlay.type == 'workshop' %}
<section class="panel">
<div class="page-heading">
<h2>Workshop items</h2>
{% if can_edit %}
<form method="post" action="/overlays/{{ overlay.id }}/build" class="inline-form">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<button type="submit" class="button-secondary">Rebuild</button>
</form>
{% endif %}
</div>
{% if can_edit %}
<form method="post" action="/overlays/{{ overlay.id }}/items" class="stack">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<fieldset class="workshop-input-mode">
<legend>Input mode</legend>
<label><input type="radio" name="input_mode" value="items" checked> Items (paste IDs or URLs; one or many)</label>
<label><input type="radio" name="input_mode" value="collection"> Collection (one ID or URL)</label>
</fieldset>
<label>Workshop input <textarea name="input" rows="3" placeholder="123456789&#10;https://steamcommunity.com/sharedfiles/filedetails/?id=987654321"></textarea></label>
<div>
<button type="submit">Add</button>
</div>
</form>
{% endif %}
<div id="overlay-item-table">
{% include "_overlay_item_table.html" with context %}
</div>
</section>
{% if latest_build_job %}
<section class="panel">
<h2>Latest build</h2>
<p>
<a href="/jobs/{{ latest_build_job.id }}">job #{{ latest_build_job.id }}</a>
— state: <strong>{{ latest_build_job.state }}</strong>
</p>
</section>
{% endif %}
{% endif %}
<section class="panel">
<h2>Used by</h2>
{% if using_blueprints %}
@ -43,7 +87,7 @@
{% endif %}
</section>
{% if g.user.admin %}
{% if can_edit %}
<dialog id="delete-overlay-modal" class="modal" aria-labelledby="delete-overlay-title">
<div class="modal-header">
<h2 id="delete-overlay-title">Delete overlay "{{ overlay.name }}"?</h2>

23
l4d2web/templates/overlays.html
View file

@ -6,43 +6,48 @@
<section class="panel">
<div class="page-heading">
<h1>Overlays</h1>
{% if g.user.admin %}
<button type="button" data-modal-open="create-overlay-modal">+ Create</button>
{% endif %}
</div>
<table class="table">
<thead><tr><th>Name</th><th>Path</th></tr></thead>
<thead><tr><th>Name</th><th>Type</th><th>Scope</th><th>Path</th></tr></thead>
<tbody>
{% for overlay in overlays %}
<tr>
<td><a href="/overlays/{{ overlay.id }}">{{ overlay.name }}</a></td>
<td>{{ overlay.type }}</td>
<td class="muted">{% if overlay.user_id %}private{% else %}system{% endif %}</td>
<td class="muted">{{ overlay.path }}</td>
</tr>
{% else %}
<tr><td colspan="2" class="muted">No overlays configured.</td></tr>
<tr><td colspan="4" class="muted">No overlays yet.</td></tr>
{% endfor %}
</tbody>
</table>
</section>
{% if g.user.admin %}
<dialog id="create-overlay-modal" class="modal" aria-labelledby="create-overlay-title">
<form method="post" action="/overlays" class="stack">
<div class="modal-header">
<h2 id="create-overlay-title">Add overlay</h2>
<h2 id="create-overlay-title">Create overlay</h2>
<button type="button" class="modal-close" data-modal-close aria-label="Close">&times;</button>
</div>
<div class="modal-body">
<input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
<fieldset class="overlay-type-radio">
<legend>Type</legend>
<label><input type="radio" name="type" value="workshop" checked> Workshop (downloads from Steam)</label>
{% if g.user.admin %}
<label><input type="radio" name="type" value="external"> External (admin-managed; populated via filesystem)</label>
{% endif %}
</fieldset>
<label>Name <input name="name" required></label>
<label>Path <input name="path" required placeholder="standard"></label>
<p class="muted">The path is generated automatically.</p>
</div>
<div class="modal-footer">
<button type="button" class="button-secondary" data-modal-close>Cancel</button>
<button type="submit">Add overlay</button>
<button type="submit">Create</button>
</div>
</form>
</dialog>
{% endif %}
{% endblock %}

160
l4d2web/tests/test_overlays.py
View file

@ -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"},
)

279
l4d2web/tests/test_workshop_routes.py Normal file
View file

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