df1ccb4cca
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>
279 lines
9.6 KiB
Python
279 lines
9.6 KiB
Python
"""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
|