Gitea-Tools/tests/test_api_reliability.py

"""Unit coverage for shared API pagination and failure handling (#67).

Covers gitea_auth.api_request failure conversion (timeouts, DNS/network,
502/503, malformed error payloads, malformed success JSON, no-secret leakage,
preserved success + 429 behavior) and gitea_auth.api_get_all pagination
(single/multi page, missing/malformed metadata, limit cap, max_pages, query
handling). Everything is mocked — no real network calls are made.
"""
import io
import unittest
import urllib.error
from unittest.mock import patch

import gitea_auth
import gitea_audit

FAKE_AUTH = "Basic ZmFrZTpmYWtl"  # not a real credential
URL = "https://gitea.example.com/api/v1/repos/o/r/issues"


class FakeResp:
    """Minimal context-manager stand-in for a urlopen response."""

    def __init__(self, body):
        self._body = body.encode("utf-8") if isinstance(body, str) else body

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def http_error(code, body="", headers=None):
    return urllib.error.HTTPError(
        url=URL, code=code, msg="err",
        hdrs=headers or {}, fp=io.BytesIO(body.encode("utf-8")),
    )


# ---------------------------------------------------------------------------
# api_request — success path preserved
# ---------------------------------------------------------------------------
class TestApiRequestSuccess(unittest.TestCase):

    @patch("gitea_auth.urllib.request.urlopen")
    def test_success_returns_parsed_json(self, mock_open):
        mock_open.return_value = FakeResp('{"number": 1, "title": "ok"}')
        self.assertEqual(
            gitea_auth.api_request("GET", URL, FAKE_AUTH),
            {"number": 1, "title": "ok"},
        )

    @patch("gitea_auth.urllib.request.urlopen")
    def test_empty_body_returns_none(self, mock_open):
        mock_open.return_value = FakeResp("")
        self.assertIsNone(gitea_auth.api_request("GET", URL, FAKE_AUTH))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_429_then_success_still_retries(self, mock_open):
        mock_open.side_effect = [http_error(429), FakeResp('{"ok": true}')]
        calls = []
        result = gitea_auth.api_request(
            "GET", URL, FAKE_AUTH,
            sleep_func=lambda d: calls.append(d), rand_func=lambda: 0.0,
        )
        self.assertEqual(result, {"ok": True})
        self.assertEqual(len(calls), 1)  # slept once between the two attempts


# ---------------------------------------------------------------------------
# api_request — failure handling
# ---------------------------------------------------------------------------
class TestApiRequestFailures(unittest.TestCase):

    @patch("gitea_auth.urllib.request.urlopen")
    def test_timeout_converted_to_runtimeerror(self, mock_open):
        mock_open.side_effect = TimeoutError("timed out")
        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        # Fixed message only (#699) — still a RuntimeError subclass.
        self.assertIsInstance(ctx.exception, RuntimeError)
        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")

    @patch("gitea_auth.urllib.request.urlopen")
    def test_dns_network_failure_converted(self, mock_open):
        mock_open.side_effect = urllib.error.URLError("Name or service not known")
        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")

    @patch("gitea_auth.urllib.request.urlopen")
    def test_502_upstream_message(self, mock_open):
        mock_open.side_effect = http_error(502, "bad gateway")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        msg = str(ctx.exception)
        self.assertEqual(msg, "Gitea upstream unavailable")
        self.assertEqual(ctx.exception.reason_code, "upstream_unavailable")
        self.assertEqual(ctx.exception.http_status, 502)

    @patch("gitea_auth.urllib.request.urlopen")
    def test_503_upstream_message(self, mock_open):
        mock_open.side_effect = http_error(503, "")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(str(ctx.exception), "Gitea upstream unavailable")
        self.assertEqual(ctx.exception.http_status, 503)

    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_error_payload_does_not_crash(self, mock_open):
        # Non-JSON garbage error body must still yield a clean typed error
        # with a fixed message (no body echo — #699).
        mock_open.side_effect = http_error(500, "<html>garbage</html>")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
        self.assertNotIn("<html>", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_success_json_raises_clean_error(self, mock_open):
        mock_open.return_value = FakeResp("not json{")
        with self.assertRaises(RuntimeError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertIn("malformed JSON response", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_no_secret_leak_in_error_body(self, mock_open):
        mock_open.side_effect = http_error(
            400, "failed: token supersecret123 rejected")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        msg = str(ctx.exception)
        self.assertNotIn("supersecret123", msg)
        # Fixed message — body never appears (stronger than redaction).
        self.assertEqual(msg, "Gitea HTTP request failed")

    @patch("gitea_auth.urllib.request.urlopen")
    def test_auth_header_never_in_error(self, mock_open):
        mock_open.side_effect = http_error(400, "bad request")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIn(FAKE_AUTH, str(ctx.exception))


# ---------------------------------------------------------------------------
# api_get_all — pagination
# ---------------------------------------------------------------------------
class TestApiGetAll(unittest.TestCase):

    @patch("gitea_auth.api_request")
    def test_single_page(self, mock_req):
        mock_req.return_value = [{"id": 1}, {"id": 2}]  # short page (< page_size)
        result = gitea_auth.api_get_all(URL, FAKE_AUTH, page_size=50)
        self.assertEqual(result, [{"id": 1}, {"id": 2}])
        self.assertEqual(mock_req.call_count, 1)

    @patch("gitea_auth.api_request")
    def test_multi_page(self, mock_req):
        mock_req.side_effect = [
            [{"id": 1}, {"id": 2}],  # full page
            [{"id": 3}, {"id": 4}],  # full page
            [{"id": 5}],             # short page -> stop
        ]
        result = gitea_auth.api_get_all(URL, FAKE_AUTH, page_size=2)
        self.assertEqual([r["id"] for r in result], [1, 2, 3, 4, 5])
        self.assertEqual(mock_req.call_count, 3)

    @patch("gitea_auth.api_request")
    def test_missing_metadata_none_page_ends(self, mock_req):
        mock_req.return_value = None  # empty/malformed metadata -> treated as end
        self.assertEqual(gitea_auth.api_get_all(URL, FAKE_AUTH), [])
        self.assertEqual(mock_req.call_count, 1)

    @patch("gitea_auth.api_request")
    def test_malformed_metadata_non_list_raises(self, mock_req):
        mock_req.return_value = {"message": "not a list"}
        with self.assertRaises(RuntimeError) as ctx:
            gitea_auth.api_get_all(URL, FAKE_AUTH)
        self.assertIn("expected a list page", str(ctx.exception))

    @patch("gitea_auth.api_request")
    def test_limit_caps_results(self, mock_req):
        mock_req.side_effect = [[{"id": 1}, {"id": 2}], [{"id": 3}, {"id": 4}]]
        result = gitea_auth.api_get_all(URL, FAKE_AUTH, page_size=2, limit=3)
        self.assertEqual([r["id"] for r in result], [1, 2, 3])

    @patch("gitea_auth.api_request")
    def test_max_pages_safety_cap(self, mock_req):
        mock_req.side_effect = [
            [{"id": 1}, {"id": 2}], [{"id": 3}, {"id": 4}], [{"id": 5}, {"id": 6}],
        ]
        result = gitea_auth.api_get_all(URL, FAKE_AUTH, page_size=2, max_pages=2)
        self.assertEqual(len(result), 4)
        self.assertEqual(mock_req.call_count, 2)

    @patch("gitea_auth.api_request")
    def test_query_params_appended_and_preserved(self, mock_req):
        mock_req.return_value = []  # first (empty) page ends immediately
        gitea_auth.api_get_all(URL + "?state=open", FAKE_AUTH, page_size=2)
        called_url = mock_req.call_args[0][1]
        self.assertIn("state=open", called_url)
        self.assertIn("page=1", called_url)
        self.assertIn("limit=2", called_url)


if __name__ == "__main__":
    unittest.main()