fix(api): validate traffic response containers

src/traffic_archive/api.py

@@ -65,24 +65,48 @@ def _request(path: str, token: str, retries: int = 3) -> Any:
     raise TrafficError(f"{path} failed after {retries} attempts: {last}")
 
 
+def _traffic_rows(
+    path: str, token: str, metric: str | None = None
+) -> list[dict[str, Any]]:
+    """Decode and validate the response container for a traffic endpoint."""
+    try:
+        response = _request(path, token)
+    except json.JSONDecodeError:
+        raise TrafficError(f"{path} returned invalid JSON.") from None
+
+    if metric is None:
+        if not isinstance(response, list):
+            raise TrafficError(f"{path} returned an invalid response: expected a JSON list.")
+        return response
+
+    if not isinstance(response, dict):
+        raise TrafficError(f"{path} returned an invalid response: expected a JSON object.")
+    rows = response.get(metric)
+    if not isinstance(rows, list):
+        raise TrafficError(
+            f"{path} returned an invalid response: expected an explicit '{metric}' list."
+        )
+    return rows
+
+
 def views(repo: str, token: str, per: str = "day") -> list[dict[str, Any]]:
     if per not in ("day", "week"):
         raise ValueError("per must be 'day' or 'week'")
-    return _request(f"/repos/{repo}/traffic/views?per={per}", token).get("views", [])
+    return _traffic_rows(f"/repos/{repo}/traffic/views?per={per}", token, "views")
 
 
 def clones(repo: str, token: str, per: str = "day") -> list[dict[str, Any]]:
     if per not in ("day", "week"):
         raise ValueError("per must be 'day' or 'week'")
-    return _request(f"/repos/{repo}/traffic/clones?per={per}", token).get("clones", [])
+    return _traffic_rows(f"/repos/{repo}/traffic/clones?per={per}", token, "clones")
 
 
 def referrers(repo: str, token: str) -> list[dict[str, Any]]:
-    return _request(f"/repos/{repo}/traffic/popular/referrers", token)
+    return _traffic_rows(f"/repos/{repo}/traffic/popular/referrers", token)
 
 
 def paths(repo: str, token: str) -> list[dict[str, Any]]:
-    return _request(f"/repos/{repo}/traffic/popular/paths", token)
+    return _traffic_rows(f"/repos/{repo}/traffic/popular/paths", token)
 
 
 def owned_repos(owner: str, token: str, include_forks: bool = False) -> list[str]:

tests/test_api.py

@@ -1,6 +1,8 @@
 """Isolated unit tests for traffic_archive.api."""
 
 import io
+import json
+import traceback
 import urllib.error
 from unittest.mock import Mock
 
@@ -48,11 +50,12 @@ def test_traffic_aggregation_rejects_unsupported_values_before_request(
 
 
 @pytest.mark.parametrize("helper", [api.views, api.clones])
-def test_traffic_aggregation_preserves_missing_data_as_empty_list(monkeypatch, helper):
+def test_traffic_aggregation_rejects_missing_metric(monkeypatch, helper):
     request = Mock(return_value={})
     monkeypatch.setattr(api, "_request", request)
 
-    assert helper("example/project", "dummy-token") == []
+    with pytest.raises(api.TrafficError, match="expected an explicit"):
+        helper("example/project", "dummy-token")
 
 
 @pytest.mark.parametrize("helper", [api.views, api.clones])
@@ -226,3 +229,176 @@ def test_request_closes_response(monkeypatch):
     api._request("/repos/owner/repo/traffic/views", SENTINEL_TOKEN)
 
     assert response.closed
+
+
+TRAFFIC_HELPERS = [
+    pytest.param(
+        api.views,
+        "/repos/owner/repo/traffic/views?per=day",
+        "views",
+        id="views",
+    ),
+    pytest.param(
+        api.clones,
+        "/repos/owner/repo/traffic/clones?per=day",
+        "clones",
+        id="clones",
+    ),
+    pytest.param(
+        api.referrers,
+        "/repos/owner/repo/traffic/popular/referrers",
+        None,
+        id="referrers",
+    ),
+    pytest.param(
+        api.paths,
+        "/repos/owner/repo/traffic/popular/paths",
+        None,
+        id="paths",
+    ),
+]
+
+
+def _serve_success_body(monkeypatch, body: str):
+    response = _FakeResponse(body.encode("utf-8"))
+    requests = []
+
+    def fake_urlopen(req, *args, **kwargs):
+        requests.append(req)
+        return response
+
+    monkeypatch.setattr(api.urllib.request, "urlopen", fake_urlopen)
+    return requests, response
+
+
+def _response_payload(metric, rows):
+    if metric is None:
+        return json.dumps(rows)
+    return json.dumps({"extra": "preserved by caller", metric: rows})
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS)
+def test_traffic_helpers_reject_invalid_json_without_disclosure(
+    monkeypatch, helper, path, metric
+):
+    body = '<html>RESPONSE_BODY_SENTINEL "token":"BODY_TOKEN_SENTINEL"</html>'
+    token = "DUMMY_TOKEN_SENTINEL"
+    requests, response = _serve_success_body(monkeypatch, body)
+
+    with pytest.raises(api.TrafficError) as exc_info:
+        helper("owner/repo", token)
+
+    message = str(exc_info.value)
+    displayed = "".join(
+        traceback.format_exception(
+            type(exc_info.value), exc_info.value, exc_info.value.__traceback__
+        )
+    )
+    assert path in message
+    assert "invalid JSON" in message
+    assert "RESPONSE_BODY_SENTINEL" not in message
+    assert "BODY_TOKEN_SENTINEL" not in message
+    assert token not in message
+    assert "RESPONSE_BODY_SENTINEL" not in displayed
+    assert "BODY_TOKEN_SENTINEL" not in displayed
+    assert token not in displayed
+    assert response.closed
+    assert len(requests) == 1
+    headers = {name.lower(): value for name, value in requests[0].header_items()}
+    assert headers["authorization"] == "Bearer " + token
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS)
+@pytest.mark.parametrize("root", ["null", '"RESPONSE_BODY_SENTINEL"'])
+def test_traffic_helpers_reject_wrong_root_containers(
+    monkeypatch, helper, path, metric, root
+):
+    requests, response = _serve_success_body(monkeypatch, root)
+
+    with pytest.raises(api.TrafficError) as exc_info:
+        helper("owner/repo", SENTINEL_TOKEN)
+
+    assert path in str(exc_info.value)
+    assert "expected" in str(exc_info.value)
+    assert "RESPONSE_BODY_SENTINEL" not in str(exc_info.value)
+    assert SENTINEL_TOKEN not in str(exc_info.value)
+    assert response.closed
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS[:2])
+def test_daily_helpers_reject_list_root(monkeypatch, helper, path, metric):
+    requests, response = _serve_success_body(monkeypatch, "[]")
+
+    with pytest.raises(api.TrafficError) as exc_info:
+        helper("owner/repo", SENTINEL_TOKEN)
+
+    assert path in str(exc_info.value)
+    assert "JSON object" in str(exc_info.value)
+    assert SENTINEL_TOKEN not in str(exc_info.value)
+    assert response.closed
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS[2:])
+def test_snapshot_helpers_reject_object_root(monkeypatch, helper, path, metric):
+    body = '{"response":"RESPONSE_BODY_SENTINEL"}'
+    requests, response = _serve_success_body(monkeypatch, body)
+
+    with pytest.raises(api.TrafficError) as exc_info:
+        helper("owner/repo", SENTINEL_TOKEN)
+
+    assert path in str(exc_info.value)
+    assert "JSON list" in str(exc_info.value)
+    assert "RESPONSE_BODY_SENTINEL" not in str(exc_info.value)
+    assert SENTINEL_TOKEN not in str(exc_info.value)
+    assert response.closed
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS[:2])
+@pytest.mark.parametrize(
+    "body",
+    [
+        "{}",
+        '{"other": []}',
+        '{"views": null}',
+        '{"views": {}}',
+        '{"views": "RESPONSE_BODY_SENTINEL"}',
+    ],
+)
+def test_daily_helpers_reject_missing_or_non_list_metric(
+    monkeypatch, helper, path, metric, body
+):
+    if metric == "clones":
+        body = body.replace("views", "clones")
+    requests, response = _serve_success_body(monkeypatch, body)
+
+    with pytest.raises(api.TrafficError) as exc_info:
+        helper("owner/repo", SENTINEL_TOKEN)
+
+    assert path in str(exc_info.value)
+    assert f"{metric}" in str(exc_info.value)
+    assert "RESPONSE_BODY_SENTINEL" not in str(exc_info.value)
+    assert SENTINEL_TOKEN not in str(exc_info.value)
+    assert response.closed
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize(("helper", "path", "metric"), TRAFFIC_HELPERS)
+@pytest.mark.parametrize(
+    "rows",
+    [[], [{"timestamp": "2026-10-10T00:00:00Z", "count": 3, "uniques": 2}]],
+)
+def test_traffic_helpers_accept_empty_and_nonempty_lists(
+    monkeypatch, helper, path, metric, rows
+):
+    body = _response_payload(metric, rows)
+    requests, response = _serve_success_body(monkeypatch, body)
+
+    result = helper("owner/repo", "dummy-token")
+
+    assert result == rows
+    assert response.closed
+    assert len(requests) == 1
+    assert requests[0].full_url == api.API + path