From db527ea78ae564ba3d688f7748df66910426c9b0 Mon Sep 17 00:00:00 2001 From: Skulldorom <51134009+Skulldorom@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:43:05 +0300 Subject: [PATCH] Test API endpoints through TestClient --- main.py | 7 +- tests/test_main.py | 295 +++++++++++++++++++++++++-------------------- 2 files changed, 167 insertions(+), 135 deletions(-) diff --git a/main.py b/main.py index 627c74a..cb25e5e 100644 --- a/main.py +++ b/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: diff --git a/tests/test_main.py b/tests/test_main.py index 8e624e9..59fe711 100644 --- a/tests/test_main.py +++ b/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"