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

refactor(workshop): autodetect collections; drop input_mode form field

Browse source 
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>
This commit is contained in:
mwiegand 2026-05-19 00:27:50 +02:00
parent 6a04594c19
commit 5c56f18d0c
No known key found for this signature in database
2 changed files with 72 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
l4d2web/l4d2web/routes/workshop_routes.py
View file

@ -40,7 +40,6 @@ def add_items(overlay_id: int) -> Response:
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)
@ -49,21 +48,18 @@ def add_items(overlay_id: int) -> Response:
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:
ids = steam_workshop.expand_collections(ids)
except requests.RequestException as exc:
return Response(f"steam api error: {exc}", status=502)
if not ids:
return Response("no items to add (collections may be empty)", 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:
except requests.RequestException as exc:
return Response(f"steam api error: {exc}", status=502)
with session_scope() as db:

89
l4d2web/tests/test_workshop_routes.py
View file

@ -85,6 +85,16 @@ 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)
@ -92,7 +102,7 @@ def test_add_single_item_creates_association_and_enqueues_build(overlay_for):
with _patch_steam([_meta("1001")]):
response = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
@ -118,7 +128,7 @@ def test_add_multiline_batch_coalesces_into_one_build_job(overlay_for):
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"},
data={"input": "1001\n1002\n1003"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 302
@ -130,23 +140,54 @@ def test_add_multiline_batch_coalesces_into_one_build_job(overlay_for):
assert len(jobs) == 1, "multi-item add should coalesce into a single build job"
def test_add_collection_resolves_members(overlay_for):
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, "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"},
)
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
assert response.headers["Location"].startswith("/jobs/")
resolve.assert_called_once_with("555")
expand.assert_called_once_with(["555"])
with session_scope() as session:
assert session.query(OverlayWorkshopItem).count() == 2
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):
@ -159,7 +200,7 @@ def test_add_non_l4d2_item_returns_400(overlay_for):
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"},
data={"input": "9999"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 400
@ -177,7 +218,7 @@ def test_add_duplicate_item_does_not_500(overlay_for):
with _patch_steam([_meta("1001")]):
first = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
assert first.status_code == 302
@ -185,7 +226,7 @@ def test_add_duplicate_item_does_not_500(overlay_for):
with _patch_steam([_meta("1001")]):
second = user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
assert second.status_code == 302
@ -201,7 +242,7 @@ def test_remove_item_drops_association_and_enqueues_rebuild(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
@ -279,7 +320,7 @@ def test_other_user_cannot_modify_workshop_overlay(overlay_for):
intruder_client = login(intruder_id)
response = intruder_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
assert response.status_code == 403
@ -291,7 +332,7 @@ def test_overlay_refresh_owner_can_refresh_and_enqueues_build(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
@ -356,7 +397,7 @@ def test_overlay_refresh_admin_can_refresh_anyone(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
@ -375,7 +416,7 @@ def test_overlay_refresh_502_on_steam_error(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
@ -409,7 +450,7 @@ def test_overlay_refresh_non_requests_exception_propagates(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session:
@ -432,7 +473,7 @@ def test_overlay_refresh_missing_item_records_last_error(overlay_for):
with _patch_steam([_meta("1001")]):
user_client.post(
f"/overlays/{overlay_id}/items",
data={"input": "1001", "input_mode": "items"},
data={"input": "1001"},
headers={"X-CSRF-Token": "test-token"},
)
with session_scope() as session: