cronekorkn/left4me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
left4me/l4d2web/tests/test_workshop_routes.py
mwiegand 5c56f18d0c
refactor(workshop): autodetect collections; drop input_mode form field 
add_items now always calls expand_collections after parsing input, so a
single textarea accepts any mix of item IDs/URLs and collection IDs/URLs
without a mode toggle. The legacy "items vs collection" branching in the
handler is gone. Existing tests strip the now-ignored input_mode field;
two new tests cover the autodetect (collection-only) and mixed-paste
paths.

Plan deviation: rather than baking the expand_collections passthrough
into the _patch_steam helper (the plan's suggestion), uses a module-
autouse fixture _stub_expand_collections to stub it to identity by
default. The autouse approach handles the patch-shadowing problem the
helper version would have introduced for the new autodetect tests (which
need to assert specific args on a per-test expand_collections patch).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-19 00:27:50 +02:00

495 lines
18 KiB
Python
Raw Blame History

"""Tests for the workshop overlay routes (add items, remove items, build,
admin refresh)."""
from __future__ import annotations
from datetime import UTC, datetime
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["pw_changed_at"] = datetime.now(UTC).isoformat()
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))
@pytest.fixture(autouse=True)
def _stub_expand_collections():
"""By default, expand_collections is a passthrough so existing item-only
tests don't make real Steam API calls. Tests that exercise autodetect
override this with their own patch.object(..., return_value=[...]) — the
explicit per-test patch wins inside the `with` block."""
with patch.object(steam_workshop, "expand_collections", side_effect=lambda x: list(x)):
yield
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"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
assert response.headers["Location"].startswith("/jobs/")
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"
assert response.headers["Location"] == f"/jobs/{jobs[0].id}"
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"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
assert response.headers["Location"].startswith("/jobs/")
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_autodetects_and_expands_children(overlay_for):
"""Pasting a collection ID expands to its children via autodetect — no
input_mode field is needed in the request."""
app, login, user_id, _admin_id, overlay_id = overlay_for
user_client = login(user_id)
with patch.object(
steam_workshop, "expand_collections", return_value=["1001", "1002", "1003"]
) as expand, _patch_steam([_meta("1001"), _meta("1002"), _meta("1003")]):
response = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "555"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
expand.assert_called_once_with(["555"])
with session_scope() as session:
steam_ids = {wi.steam_id for wi in session.query(WorkshopItem).all()}
assert steam_ids == {"1001", "1002", "1003"}
def test_add_mixed_items_and_collection_in_one_paste(overlay_for):
"""A single submission can mix item IDs, item URLs, and a collection URL;
expand_collections flattens collections in place, and metadata fetch covers
the resulting flat ID list."""
app, login, user_id, _admin_id, overlay_id = overlay_for
user_client = login(user_id)
raw = (
"1001\n"
"https://steamcommunity.com/sharedfiles/filedetails/?id=555\n"
"https://steamcommunity.com/sharedfiles/filedetails/?id=2001\n"
)
with patch.object(
steam_workshop, "expand_collections", return_value=["1001", "3001", "3002", "2001"]
) as expand, _patch_steam([_meta(s) for s in ("1001", "3001", "3002", "2001")]):
response = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": raw},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
# Parse yields [1001, 555, 2001] from the mixed input; expand_collections
# receives that flat list, flattens the collection in place.
expand.assert_called_once_with(["1001", "555", "2001"])
with session_scope() as session:
steam_ids = {wi.steam_id for wi in session.query(WorkshopItem).all()}
assert steam_ids == {"1001", "2001", "3001", "3002"}
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"},
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"},
headers={"X-CSRF-Token": "test-token"},
)
assert first.status_code == 302
with _patch_steam([_meta("1001")]):
second = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
assert second.status_code == 302
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"},
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 == 302
assert response.headers["Location"].startswith("/jobs/")
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"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 403
def test_overlay_refresh_owner_can_refresh_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")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
job.state = "succeeded"
fresh_meta = steam_workshop.WorkshopMetadata(
steam_id="1001", title="Item 1001 (updated)", filename="1001.vpk",
file_url="https://example.com/1001.vpk", file_size=99, time_updated=1800000000,
preview_url="https://example.com/preview-1001.jpg", consumer_app_id=550, result=1,
)
with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[fresh_meta]) as fetch:
response = user_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
assert response.headers["Location"].startswith("/jobs/")
fetch.assert_called_once_with(["1001"], mode="refresh")
with session_scope() as session:
wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
assert wi.title == "Item 1001 (updated)"
assert wi.time_updated == 1800000000
new_jobs = session.query(Job).filter_by(
operation="build_overlay", overlay_id=overlay_id, state="queued"
).all()
assert len(new_jobs) == 1
def test_overlay_refresh_returns_400_when_overlay_empty(overlay_for):
app, login, user_id, _admin_id, overlay_id = overlay_for
user_client = login(user_id)
with patch.object(steam_workshop, "fetch_metadata_batch") as fetch:
response = user_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 400
fetch.assert_not_called()
def test_overlay_refresh_forbidden_for_non_owner(overlay_for, env_user):
app, login, _user_id, _admin_id, overlay_id = overlay_for
with session_scope() as session:
bob = User(username="bob", password_digest=hash_password("x"), admin=False)
session.add(bob)
session.flush()
bob_id = bob.id
bob_client = login(bob_id)
response = bob_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 403
def test_overlay_refresh_admin_can_refresh_anyone(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"},
headers={"X-CSRF-Token": "test-token"},
)
admin_client = login(admin_id)
with _patch_steam([_meta("1001")]):
response = admin_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
def test_overlay_refresh_502_on_steam_error(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"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
job.state = "succeeded"
baseline_job_count = session.query(Job).filter_by(
operation="build_overlay", overlay_id=overlay_id, state="queued"
).count()
import requests as _requests
with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=_requests.ConnectionError("boom")):
response = user_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 502
assert b"steam api error" in response.data
with session_scope() as session:
n = session.query(Job).filter_by(
operation="build_overlay", overlay_id=overlay_id, state="queued"
).count()
assert n == baseline_job_count
def test_overlay_refresh_non_requests_exception_propagates(overlay_for):
"""A non-requests exception (e.g., ValueError from a server-side data
issue) must not be disguised as a 502 — let it bubble up as 500."""
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"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
job.state = "succeeded"
# Flask in TESTING mode re-raises uncaught exceptions instead of returning
# 500, so we assert the ValueError propagates rather than checking a status.
with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=ValueError("bad steam_id in db")):
with pytest.raises(ValueError, match="bad steam_id in db"):
user_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
def test_overlay_refresh_missing_item_records_last_error(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"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
job.state = "succeeded"
with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[]):
response = user_client.post(
f"/overlays/{overlay_id}/refresh",
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
with session_scope() as session:
wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
assert "no entry" in wi.last_error
new_jobs = session.query(Job).filter_by(
operation="build_overlay", overlay_id=overlay_id, state="queued"
).all()
assert len(new_jobs) == 1, "refresh should enqueue build even when Steam returns no entries"
Reference in a new issue View git blame Copy permalink