workshop_routes: narrow refresh's steam exception handler

Catch only requests.RequestException in refresh_overlay so that
server-side data errors (e.g., ValueError) bubble up as 500 rather
than being disguised as a 502 "steam api error". Update the 502 test to
use a real requests exception, add a sibling test that verifies
non-requests exceptions propagate, and explicitly assert that refresh
enqueues a build_overlay job even when Steam returns no entries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mwiegand 2026-05-11 23:08:41 +02:00
parent f5094c2d9d
commit e1b189ad3c
No known key found for this signature in database
2 changed files with 33 additions and 2 deletions

View file

@ -2,6 +2,7 @@
admin global refresh)."""
from __future__ import annotations
import requests
from flask import Blueprint, Response, redirect, request
from sqlalchemy import delete as sa_delete
from sqlalchemy import select
@ -145,7 +146,7 @@ def refresh_overlay(overlay_id: int) -> Response:
try:
metas = steam_workshop.fetch_metadata_batch(steam_ids, mode="refresh")
except Exception as exc:
except requests.RequestException as exc:
return Response(f"steam api error: {exc}", status=502)
metas_by_id = {m.steam_id: m for m in metas}

View file

@ -385,7 +385,8 @@ def test_overlay_refresh_502_on_steam_error(overlay_for):
operation="build_overlay", overlay_id=overlay_id, state="queued"
).count()
with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=Exception("boom")):
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"},
@ -400,6 +401,31 @@ def test_overlay_refresh_502_on_steam_error(overlay_for):
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", "input_mode": "items"},
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)
@ -422,3 +448,7 @@ def test_overlay_refresh_missing_item_records_last_error(overlay_for):
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"