src/gitea_proxy.py

@@ -40,6 +40,10 @@ async def pull_requests() -> list[dict]:
 async def activity_events() -> list[dict]:
     user = await current_user()
     events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
+    if events is None:
+        events = []
+    elif not isinstance(events, list):
+        raise ValueError("Gitea activity feed response was not a list")
     return [
         {
             "type": (

tests/test_event_stream.py

@@ -164,6 +164,25 @@ async def test_activity_events_normalizes_null_feed_payload(monkeypatch):
     assert await gitea_proxy.activity_events() == []
 
 
+@pytest.mark.anyio
+async def test_event_stream_rejects_malformed_activity_feed_payload(monkeypatch):
+    async def fake_current_user():
+        return {"login": "timmy"}
+
+    async def fake_fetch(path):
+        return {"message": "unexpected upstream shape"}
+
+    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
+    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
+
+    response = await main.event_stream()
+
+    assert getattr(response, "status_code", None) == 503
+    assert json.loads(response.body) == {
+        "error": "Gitea event stream is temporarily unavailable"
+    }
+
+
 @pytest.mark.anyio
 async def test_activity_events_normalizes_malformed_nested_metadata(monkeypatch):
     async def fake_current_user():