Exercise FastAPI endpoints through TestClient and mock FreshRSS transport

main.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from urllib.parse import quote
 from fastapi import FastAPI, HTTPException, Query
 import requests
 import humanize
@@ -82,7 +83,11 @@ def freshrss_unread(
         "n": n,
     }
     category_label = category if isinstance(category, str) and category else None
-    stream_id = f"user/-/label/{category_label}" if category_label else "user/-/state/com.google/reading-list"
+    stream_id = (
+        f"user/-/label/{quote(category_label, safe='')}"
+        if category_label
+        else "user/-/state/com.google/reading-list"
+    )
     # Using the same host as before but with the right endpoint
     url = f"{FRESHRSS_HOST}/api/greader.php/reader/api/0/stream/contents/{stream_id}"
     try:

tests/test_main.py

@@ -3,8 +3,9 @@
 from pathlib import Path
 
 import pytest
-from fastapi import HTTPException
 import requests
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
 
 ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
@@ -33,13 +34,40 @@ def raise_for_status(self):
             raise self._raise_error
 
 
-def import_app(monkeypatch, env=None):
-    for name in REQUIRED_ENV:
-        monkeypatch.delenv(name, raising=False)
-    for name, value in (env or REQUIRED_ENV).items():
+@pytest.fixture
+def main_module(monkeypatch):
+    """Import the application only after installing a deterministic environment."""
+    for name, value in REQUIRED_ENV.items():
         monkeypatch.setenv(name, value)
     sys.modules.pop("main", None)
-    return importlib.import_module("main")
+    module = importlib.import_module("main")
+    yield module
+    sys.modules.pop("main", None)
+
+
+@pytest.fixture
+def test_client(main_module):
+    with TestClient(main_module.app) as client:
+        yield client
+
+
+def install_freshrss_transport(monkeypatch, main_module, *, payload=None):
+    """Stub requests at the FreshRSS boundary, leaving the HTTP app intact."""
+    calls = {"post": [], "get": []}
+
+    def fake_post(url, data, timeout):
+        calls["post"].append({"url": url, "data": data, "timeout": timeout})
+        return FakeResponse(text="SID=ignored\nAuth= transport-token \n")
+
+    def fake_get(url, headers, params, timeout):
+        calls["get"].append(
+            {"url": url, "headers": headers, "params": params, "timeout": timeout}
+        )
+        return FakeResponse(payload=payload or {"items": []})
+
+    monkeypatch.setattr(main_module.requests, "post", fake_post)
+    monkeypatch.setattr(main_module.requests, "get", fake_get)
+    return calls
 
 
 def test_import_requires_freshrss_environment(monkeypatch):
@@ -50,157 +78,156 @@ def test_import_requires_freshrss_environment(monkeypatch):
     with pytest.raises(RuntimeError) as excinfo:
         importlib.import_module("main")
 
-    message = str(excinfo.value)
-    assert "FRESHRSS_HOST" in message
-    assert "FRESHRSS_USER" in message
-    assert "FRESHRSS_PASS" in message
-
+    assert all(name in str(excinfo.value) for name in REQUIRED_ENV)
 
-def test_health_endpoint_returns_ok(monkeypatch):
-    main = import_app(monkeypatch)
 
-    assert main.health() == {"status": "ok"}
-    assert any(route.path == "/health" for route in main.app.routes)
+def test_health_endpoint_returns_json_over_http(test_client):
+    response = test_client.get("/health")
 
+    assert response.status_code == 200
+    assert response.headers["content-type"] == "application/json"
+    assert response.json() == {"status": "ok"}
 
-def test_get_greader_token_logs_in_once_and_caches_token(monkeypatch):
-    main = import_app(monkeypatch)
-    calls = []
 
-    def fake_post(url, data, timeout):
-        calls.append({"url": url, "data": data, "timeout": timeout})
-        return FakeResponse(text="SID=ignored\nAuth= cached-token \n")
+def test_unread_uses_default_query_and_authenticates_once(
+    test_client, monkeypatch, main_module
+):
+    calls = install_freshrss_transport(monkeypatch, main_module)
 
-    monkeypatch.setattr(main.requests, "post", fake_post)
+    first = test_client.get("/freshrss/unread")
+    second = test_client.get("/freshrss/unread")
 
-    assert main.get_greader_token() == "cached-token"
-    assert main.get_greader_token() == "cached-token"
-    assert calls == [
+    assert first.status_code == second.status_code == 200
+    assert first.json() == second.json() == []
+    assert calls["post"] == [
         {
             "url": "https://freshrss.example.test/api/greader.php/accounts/ClientLogin",
             "data": {"Email": "reader", "Passwd": "secret"},
             "timeout": 10,
         }
     ]
-
-
-def test_get_greader_token_rejects_failed_login(monkeypatch):
-    main = import_app(monkeypatch)
-    monkeypatch.setattr(main.requests, "post", lambda *args, **kwargs: FakeResponse(status_code=403, text="nope"))
-
-    with pytest.raises(HTTPException) as excinfo:
-        main.get_greader_token()
-
-    assert excinfo.value.status_code == 502
-    assert "FreshRSS login failed with status 403" == excinfo.value.detail
-
-
-def test_get_greader_token_rejects_missing_auth_line(monkeypatch):
-    main = import_app(monkeypatch)
-    monkeypatch.setattr(main.requests, "post", lambda *args, **kwargs: FakeResponse(text="SID=only"))
-
-    with pytest.raises(HTTPException) as excinfo:
-        main.get_greader_token()
-
-    assert excinfo.value.status_code == 502
-    assert excinfo.value.detail == "Auth token not found in FreshRSS response"
-
-
-def test_get_greader_token_wraps_request_failures(monkeypatch):
-    main = import_app(monkeypatch)
-
-    def fake_post(*args, **kwargs):
-        raise requests.Timeout("slow upstream")
-
-    monkeypatch.setattr(main.requests, "post", fake_post)
-
-    with pytest.raises(HTTPException) as excinfo:
-        main.get_greader_token()
-
-    assert excinfo.value.status_code == 502
-    assert excinfo.value.detail == "FreshRSS login request failed"
-
-
-def test_freshrss_unread_fetches_reading_list_and_shapes_items(monkeypatch):
-    main = import_app(monkeypatch)
-    monkeypatch.setattr(main, "get_greader_token", lambda: "token-123")
-    captured = {}
-
-    def fake_get(url, headers, params, timeout):
-        captured.update({"url": url, "headers": headers, "params": params, "timeout": timeout})
-        return FakeResponse(
-            payload={
-                "items": [
-                    {
-                        "title": "Release shipped",
-                        "origin": {"title": "GitHub Releases"},
-                        "published": 1700000000,
-                        "alternate": [{"href": "https://example.test/release"}],
-                    },
-                    {
-                        "title": "Missing timestamp is ignored",
-                        "origin": {"title": "Bad Feed"},
-                    },
-                ]
-            }
-        )
-
-    monkeypatch.setattr(main.requests, "get", fake_get)
-
-    result = main.freshrss_unread(n=5)
-
-    assert captured == {
+    assert len(calls["get"]) == 2
+    assert calls["get"][0] == {
         "url": "https://freshrss.example.test/api/greader.php/reader/api/0/stream/contents/user/-/state/com.google/reading-list",
-        "headers": {"Authorization": "GoogleLogin auth=token-123"},
-        "params": {"xt": "user/-/state/com.google/read", "output": "json", "n": 5},
+        "headers": {"Authorization": "GoogleLogin auth=transport-token"},
+        "params": {
+            "xt": "user/-/state/com.google/read",
+            "output": "json",
+            "n": 10,
+        },
         "timeout": 10,
     }
-    assert len(result) == 1
-    assert result[0]["title"] == "Release shipped"
-    assert result[0]["feed"] == "GitHub Releases"
-    assert result[0]["published"] == 1700000000
-    assert result[0]["url"] == "https://example.test/release"
-    assert result[0]["display"].startswith("Release shipped • ")
-
 
-def test_freshrss_unread_scopes_to_category_and_handles_missing_url(monkeypatch):
-    main = import_app(monkeypatch)
-    monkeypatch.setattr(main, "get_greader_token", lambda: "token-123")
-    captured = {}
 
-    def fake_get(url, headers, params, timeout):
-        captured.update({"url": url, "params": params})
-        return FakeResponse(
-            payload={
-                "items": [
-                    {
-                        "title": "Category item",
-                        "origin": {},
-                        "published": 1700000000,
-                        "alternate": [],
-                    }
-                ]
-            }
-        )
-
-    monkeypatch.setattr(main.requests, "get", fake_get)
+def test_unread_accepts_explicit_query_and_encoded_category(
+    test_client, monkeypatch, main_module
+):
+    calls = install_freshrss_transport(monkeypatch, main_module)
+
+    response = test_client.get(
+        "/freshrss/unread", params={"n": "25", "category": "Tech & Science/News"}
+    )
+
+    assert response.status_code == 200
+    request = calls["get"][0]
+    assert request["params"]["n"] == 25
+    assert request["url"].endswith("/user/-/label/Tech%20%26%20Science%2FNews")
+
+
+@pytest.mark.parametrize("value", ["not-a-number", "1.5", "", "0", "-1"])
+def test_unread_rejects_invalid_counts_without_contacting_upstream(
+    value, test_client, monkeypatch, main_module
+):
+    calls = install_freshrss_transport(monkeypatch, main_module)
+
+    response = test_client.get("/freshrss/unread", params={"n": value})
+
+    assert response.status_code == 422
+    assert response.json()["detail"]
+    assert calls == {"post": [], "get": []}
+
+
+def test_unread_response_json_structure(test_client, monkeypatch, main_module):
+    calls = install_freshrss_transport(
+        monkeypatch,
+        main_module,
+        payload={
+            "items": [
+                {
+                    "title": "Release shipped",
+                    "origin": {"title": "GitHub Releases"},
+                    "published": 1700000000,
+                    "alternate": [{"href": "https://example.test/release"}],
+                },
+                {"title": "No timestamp", "origin": {"title": "Bad Feed"}},
+            ]
+        },
+    )
+
+    response = test_client.get("/freshrss/unread", params={"n": 5})
+
+    assert response.status_code == 200
+    assert calls["get"][0]["params"]["n"] == 5
+    body = response.json()
+    assert len(body) == 1
+    assert set(body[0]) == {"title", "feed", "published", "url", "display"}
+    assert body[0] == {
+        "title": "Release shipped",
+        "feed": "GitHub Releases",
+        "published": 1700000000,
+        "url": "https://example.test/release",
+        "display": body[0]["display"],
+    }
+    assert body[0]["display"].startswith("Release shipped • ")
+
+
+def test_unread_serializes_upstream_failure_as_502(
+    test_client, monkeypatch, main_module
+):
+    monkeypatch.setattr(
+        main_module.requests,
+        "post",
+        lambda *args, **kwargs: FakeResponse(text="Auth=token\n"),
+    )
+    monkeypatch.setattr(
+        main_module.requests,
+        "get",
+        lambda *args, **kwargs: FakeResponse(
+            raise_error=requests.HTTPError("500 Server Error")
+        ),
+    )
+
+    response = test_client.get("/freshrss/unread")
+
+    assert response.status_code == 502
+    assert response.json() == {"detail": "FreshRSS unread request failed"}
+
+
+# These focused unit tests make token parsing failures easier to diagnose than an
+# endpoint-level assertion while still mocking only the requests transport.
+def test_get_greader_token_rejects_failed_login(monkeypatch, main_module):
+    monkeypatch.setattr(
+        main_module.requests,
+        "post",
+        lambda *args, **kwargs: FakeResponse(status_code=403, text="nope"),
+    )
 
-    result = main.freshrss_unread(n=3, category="Tech")
+    with pytest.raises(HTTPException) as excinfo:
+        main_module.get_greader_token()
 
-    assert captured["url"].endswith("/stream/contents/user/-/label/Tech")
-    assert captured["params"]["n"] == 3
-    assert result[0]["feed"] is None
-    assert result[0]["url"] == ""
+    assert excinfo.value.status_code == 502
+    assert excinfo.value.detail == "FreshRSS login failed with status 403"
 
 
-def test_freshrss_unread_wraps_upstream_http_errors(monkeypatch):
-    main = import_app(monkeypatch)
-    monkeypatch.setattr(main, "get_greader_token", lambda: "token-123")
-    upstream_error = requests.HTTPError("500 Server Error")
-    monkeypatch.setattr(main.requests, "get", lambda *args, **kwargs: FakeResponse(raise_error=upstream_error))
+def test_get_greader_token_rejects_missing_auth_line(monkeypatch, main_module):
+    monkeypatch.setattr(
+        main_module.requests,
+        "post",
+        lambda *args, **kwargs: FakeResponse(text="SID=only"),
+    )
 
     with pytest.raises(HTTPException) as excinfo:
-        main.freshrss_unread()
+        main_module.get_greader_token()
 
     assert excinfo.value.status_code == 502
-    assert excinfo.value.detail == "FreshRSS unread request failed"
+    assert excinfo.value.detail == "Auth token not found in FreshRSS response"