16344bbb3f
- Ajoute try/except autour du float() pour gérer les dates HTTP RFC 7231 - Cap à 30s pour éviter un blocage indéfini sur valeur énorme - Plancher à _RETRY_DELAY pour Retry-After: 0 ou négatif (FINDING-R2) - 4 nouveaux tests : date HTTP, valeur zéro, valeur énorme, health check partiel Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
430 lines
16 KiB
Python
430 lines
16 KiB
Python
"""Tests for GiteaClient API client."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from gitea_dashboard.client import GiteaClient
|
|
|
|
|
|
class TestGiteaClientInit:
|
|
"""Test client initialization and auth."""
|
|
|
|
def test_auth_header_is_set(self):
|
|
"""Session must carry Authorization header with token prefix."""
|
|
client = GiteaClient("http://gitea.local:3000", "my-secret-token")
|
|
assert client.session.headers["Authorization"] == "token my-secret-token"
|
|
|
|
def test_base_url_stored(self):
|
|
"""Base URL is stored without trailing slash."""
|
|
client = GiteaClient("http://gitea.local:3000/", "tok")
|
|
assert client.base_url == "http://gitea.local:3000"
|
|
|
|
def test_default_timeout_is_30(self):
|
|
"""Default timeout is 30 seconds."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
assert client.timeout == 30
|
|
|
|
def test_custom_timeout_is_stored(self):
|
|
"""Custom timeout is stored and used."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok", timeout=10)
|
|
assert client.timeout == 10
|
|
|
|
|
|
class TestGetPaginated:
|
|
"""Test internal pagination logic."""
|
|
|
|
def _make_client(self):
|
|
return GiteaClient("http://gitea.local:3000", "tok")
|
|
|
|
def test_single_page(self):
|
|
"""When response has fewer items than limit, stop after one request."""
|
|
client = self._make_client()
|
|
# Return 3 items (< 50 limit) -> single page
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_resp.json.return_value = [{"id": 1}, {"id": 2}, {"id": 3}]
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp) as mock_get:
|
|
result = client._get_paginated("/api/v1/user/repos")
|
|
|
|
assert result == [{"id": 1}, {"id": 2}, {"id": 3}]
|
|
# Called exactly once (single page)
|
|
mock_get.assert_called_once()
|
|
|
|
def test_two_pages(self):
|
|
"""When first page is full (limit items), fetch second page."""
|
|
client = self._make_client()
|
|
|
|
page1 = [{"id": i} for i in range(50)] # Exactly limit=50
|
|
page2 = [{"id": i} for i in range(50, 60)] # 10 items -> last page
|
|
|
|
mock_resp1 = MagicMock()
|
|
mock_resp1.raise_for_status = MagicMock()
|
|
mock_resp1.json.return_value = page1
|
|
|
|
mock_resp2 = MagicMock()
|
|
mock_resp2.raise_for_status = MagicMock()
|
|
mock_resp2.json.return_value = page2
|
|
|
|
with patch.object(client.session, "get", side_effect=[mock_resp1, mock_resp2]):
|
|
result = client._get_paginated("/api/v1/user/repos")
|
|
|
|
assert len(result) == 60
|
|
assert result == page1 + page2
|
|
|
|
def test_pagination_params_forwarded(self):
|
|
"""Extra params are merged with pagination params."""
|
|
client = self._make_client()
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_resp.json.return_value = []
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp) as mock_get:
|
|
client._get_paginated("/api/v1/repos/o/r/milestones", params={"state": "open"})
|
|
|
|
call_params = mock_get.call_args[1]["params"]
|
|
assert call_params["state"] == "open"
|
|
assert call_params["limit"] == 50
|
|
assert call_params["page"] == 1
|
|
|
|
|
|
class TestGetRepos:
|
|
"""Test get_repos method."""
|
|
|
|
def test_get_repos_calls_paginated(self):
|
|
"""get_repos delegates to _get_paginated with correct endpoint."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
with patch.object(client, "_get_paginated", return_value=[{"id": 1}]) as mock_pag:
|
|
result = client.get_repos()
|
|
|
|
mock_pag.assert_called_once_with("/api/v1/user/repos")
|
|
assert result == [{"id": 1}]
|
|
|
|
|
|
class TestGetLatestRelease:
|
|
"""Test get_latest_release method."""
|
|
|
|
def test_returns_release_on_success(self):
|
|
"""Returns release dict when repo has a release."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"tag_name": "v1.0", "published_at": "2026-01-01"}
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_latest_release("admin", "my-repo")
|
|
|
|
assert result == {"tag_name": "v1.0", "published_at": "2026-01-01"}
|
|
|
|
def test_returns_none_on_404(self):
|
|
"""Returns None when repo has no release (404)."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 404
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_latest_release("admin", "no-release-repo")
|
|
|
|
assert result is None
|
|
|
|
def test_raises_on_server_error(self):
|
|
"""HTTP 500 raises an exception instead of silently returning bad data."""
|
|
import requests as req
|
|
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 500
|
|
mock_resp.raise_for_status.side_effect = req.HTTPError("500 Server Error")
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
with pytest.raises(req.HTTPError):
|
|
client.get_latest_release("admin", "my-repo")
|
|
|
|
|
|
class TestGetMilestones:
|
|
"""Test get_milestones method."""
|
|
|
|
def test_get_milestones_calls_paginated_with_state_open(self):
|
|
"""get_milestones delegates to _get_paginated with state=open."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
milestones = [{"title": "v2.0", "open_issues": 3, "closed_issues": 2}]
|
|
|
|
with patch.object(client, "_get_paginated", return_value=milestones) as mock_pag:
|
|
result = client.get_milestones("admin", "my-repo")
|
|
|
|
mock_pag.assert_called_once_with(
|
|
"/api/v1/repos/admin/my-repo/milestones",
|
|
params={"state": "open"},
|
|
)
|
|
assert result == milestones
|
|
|
|
|
|
class TestGetWithRetry:
|
|
"""Test _get_with_retry method (retry on timeout)."""
|
|
|
|
def _make_client(self):
|
|
return GiteaClient("http://gitea.local:3000", "tok")
|
|
|
|
@patch("time.sleep")
|
|
def test_success_first_attempt(self, mock_sleep):
|
|
"""No timeout — returns response directly without sleeping."""
|
|
client = self._make_client()
|
|
mock_resp = MagicMock()
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result is mock_resp
|
|
mock_sleep.assert_not_called()
|
|
|
|
@patch("time.sleep")
|
|
def test_success_after_timeout(self, mock_sleep):
|
|
"""First call times out, second succeeds — one sleep of 1.0s."""
|
|
client = self._make_client()
|
|
mock_resp = MagicMock()
|
|
|
|
with patch.object(
|
|
client.session, "get", side_effect=[requests.Timeout("timeout"), mock_resp]
|
|
):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result is mock_resp
|
|
mock_sleep.assert_called_once_with(1.0)
|
|
|
|
@patch("time.sleep")
|
|
def test_all_timeouts(self, mock_sleep):
|
|
"""All 3 attempts timeout — raises Timeout, sleeps twice (1.0, 2.0)."""
|
|
client = self._make_client()
|
|
timeout_exc = requests.Timeout("timeout")
|
|
|
|
with patch.object(
|
|
client.session, "get", side_effect=[timeout_exc, timeout_exc, timeout_exc]
|
|
):
|
|
with pytest.raises(requests.Timeout):
|
|
client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert mock_sleep.call_count == 2
|
|
mock_sleep.assert_any_call(1.0)
|
|
mock_sleep.assert_any_call(2.0)
|
|
|
|
|
|
class TestGetWithRetry429:
|
|
"""Test _get_with_retry method (retry on HTTP 429 rate limiting)."""
|
|
|
|
def _make_client(self):
|
|
return GiteaClient("http://gitea.local:3000", "tok")
|
|
|
|
def _make_429_response(self, retry_after=None):
|
|
"""Create a mock 429 response."""
|
|
resp = MagicMock()
|
|
resp.status_code = 429
|
|
resp.headers = {"Retry-After": retry_after} if retry_after is not None else {}
|
|
resp.raise_for_status.side_effect = requests.HTTPError(
|
|
"429 Too Many Requests", response=resp
|
|
)
|
|
return resp
|
|
|
|
def _make_200_response(self):
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
return resp
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_on_429_with_retry_after(self, mock_sleep):
|
|
"""429 with Retry-After header: sleeps for the indicated duration, then succeeds."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response(retry_after="2")
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(client.session, "get", side_effect=[resp_429, resp_200]):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
mock_sleep.assert_called_once_with(2.0)
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_on_429_without_retry_after(self, mock_sleep):
|
|
"""429 without Retry-After header: uses linear backoff (1.0s for first retry)."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response()
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(client.session, "get", side_effect=[resp_429, resp_200]):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
mock_sleep.assert_called_once_with(1.0)
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_on_429_exhausted(self, mock_sleep):
|
|
"""3 consecutive 429 responses: raises HTTPError after exhausting retries."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response()
|
|
|
|
with patch.object(client.session, "get", return_value=resp_429):
|
|
with pytest.raises(requests.HTTPError):
|
|
client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert mock_sleep.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_after_http_date_falls_back_to_backoff(self, mock_sleep):
|
|
"""Retry-After contenant une date HTTP RFC 7231 (non-numerique) :
|
|
le parsing echoue silencieusement et on retombe sur le backoff lineaire."""
|
|
client = self._make_client()
|
|
# Valeur realiste envoyee par certains serveurs
|
|
resp_429 = self._make_429_response(retry_after="Wed, 21 Oct 2025 07:28:00 GMT")
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(client.session, "get", side_effect=[resp_429, resp_200]):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
# Backoff lineaire : attempt=0 → 1 * 1.0 = 1.0s
|
|
mock_sleep.assert_called_once_with(1.0)
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_after_zero_uses_floor(self, mock_sleep):
|
|
"""Retry-After: 0 ne provoque pas un retry immediat sans backoff.
|
|
Le plancher (_RETRY_DELAY = 1.0s) est applique."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response(retry_after="0")
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(client.session, "get", side_effect=[resp_429, resp_200]):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
mock_sleep.assert_called_once_with(1.0) # plancher _RETRY_DELAY
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_after_huge_value_capped_at_30s(self, mock_sleep):
|
|
"""Retry-After avec une valeur enorme est plafonne a 30s."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response(retry_after="3600") # 1 heure
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(client.session, "get", side_effect=[resp_429, resp_200]):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
mock_sleep.assert_called_once_with(30.0) # cap a 30s
|
|
|
|
@patch("time.sleep")
|
|
def test_retry_on_429_then_timeout(self, mock_sleep):
|
|
"""429 followed by Timeout: both retry types handled in same loop."""
|
|
client = self._make_client()
|
|
resp_429 = self._make_429_response()
|
|
resp_200 = self._make_200_response()
|
|
|
|
with patch.object(
|
|
client.session,
|
|
"get",
|
|
side_effect=[resp_429, requests.Timeout("timeout"), resp_200],
|
|
):
|
|
result = client._get_with_retry("http://gitea.local:3000/api/v1/test")
|
|
|
|
assert result.status_code == 200
|
|
assert mock_sleep.call_count == 2
|
|
|
|
|
|
class TestGetVersion:
|
|
"""Test get_version method."""
|
|
|
|
def test_get_version_success(self):
|
|
"""Returns version dict on success."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"version": "1.21.0"}
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_version()
|
|
|
|
assert result == {"version": "1.21.0"}
|
|
|
|
def test_get_version_connection_error(self):
|
|
"""ConnectionError propagates to caller."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
|
|
with patch.object(client.session, "get", side_effect=requests.ConnectionError("refused")):
|
|
with pytest.raises(requests.ConnectionError):
|
|
client.get_version()
|
|
|
|
|
|
class TestGetPaginatedEdgeCases:
|
|
"""Test edge cases for API responses."""
|
|
|
|
def _make_client(self):
|
|
return GiteaClient("http://gitea.local:3000", "tok")
|
|
|
|
def test_get_paginated_malformed_json(self):
|
|
"""Response with invalid JSON raises JSONDecodeError."""
|
|
import json
|
|
|
|
client = self._make_client()
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_resp.json.side_effect = json.JSONDecodeError("Expecting value", "", 0)
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
with pytest.raises(json.JSONDecodeError):
|
|
client._get_paginated("/api/v1/user/repos")
|
|
|
|
def test_get_repos_html_response(self):
|
|
"""HTML response (status 200 but HTML content) raises on json parsing."""
|
|
import json
|
|
|
|
client = self._make_client()
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_resp.json.side_effect = json.JSONDecodeError(
|
|
"Expecting value", "<html>Maintenance</html>", 0
|
|
)
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
with pytest.raises(json.JSONDecodeError):
|
|
client.get_repos()
|
|
|
|
|
|
class TestGetLatestCommit:
|
|
"""Test get_latest_commit method."""
|
|
|
|
def test_returns_first_commit(self):
|
|
"""Returns the first commit from the list."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
commit = {"sha": "abc123", "created": "2026-03-10T14:30:00Z"}
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = [commit]
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_latest_commit("admin", "my-repo")
|
|
|
|
assert result == commit
|
|
|
|
def test_empty_repo_returns_none(self):
|
|
"""Returns None when repo has no commits."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = []
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_latest_commit("admin", "empty-repo")
|
|
|
|
assert result is None
|
|
|
|
def test_404_returns_none(self):
|
|
"""Returns None when repo is not found (404)."""
|
|
client = GiteaClient("http://gitea.local:3000", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 404
|
|
|
|
with patch.object(client.session, "get", return_value=mock_resp):
|
|
result = client.get_latest_commit("admin", "missing-repo")
|
|
|
|
assert result is None
|