From 4d66aea6ed01f49212533c91451117ef1b7aaff5 Mon Sep 17 00:00:00 2001 From: sylvain Date: Tue, 10 Mar 2026 18:50:28 +0100 Subject: [PATCH] feat(client): add GiteaClient with auth and pagination (fixes #1) - GiteaClient with requests.Session and token auth header - _get_paginated for automatic pagination (limit=50) - get_repos, get_latest_release (None on 404), get_milestones - 9 unit tests with mocked requests.Session - Fix setuptools build backend in pyproject.toml Co-Authored-By: Claude Opus 4.6 --- pyproject.toml | 2 +- src/gitea_dashboard/client.py | 73 ++++++++++++++++++ tests/test_client.py | 136 ++++++++++++++++++++++++++++++++++ 3 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 src/gitea_dashboard/client.py create mode 100644 tests/test_client.py diff --git a/pyproject.toml b/pyproject.toml index fd4804f..2b3b52c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [build-system] requires = ["setuptools>=68.0", "wheel"] -build-backend = "setuptools.backends._legacy:_Backend" +build-backend = "setuptools.build_meta" [project] name = "gitea-dashboard" diff --git a/src/gitea_dashboard/client.py b/src/gitea_dashboard/client.py new file mode 100644 index 0000000..738d546 --- /dev/null +++ b/src/gitea_dashboard/client.py @@ -0,0 +1,73 @@ +"""Client API Gitea en lecture seule.""" + +from __future__ import annotations + +import requests + + +class GiteaClient: + """Client API Gitea en lecture seule. + + Utilise requests.Session pour reutiliser les connexions HTTP. + Authentification via header Authorization: token . + """ + + _PAGE_LIMIT = 50 + + def __init__(self, base_url: str, token: str) -> None: + """Initialise le client avec l'URL de base et le token API.""" + self.base_url = base_url.rstrip("/") + self.session = requests.Session() + self.session.headers["Authorization"] = f"token {token}" + + def _get_paginated(self, endpoint: str, params: dict | None = None) -> list[dict]: + """Requete GET avec pagination automatique. + + Boucle tant que len(page) == limit (50). + """ + all_items: list[dict] = [] + page = 1 + merged_params = dict(params) if params else {} + + while True: + merged_params["limit"] = self._PAGE_LIMIT + merged_params["page"] = page + url = f"{self.base_url}{endpoint}" + resp = self.session.get(url, params=merged_params) + resp.raise_for_status() + items = resp.json() + all_items.extend(items) + if len(items) < self._PAGE_LIMIT: + break + page += 1 + + return all_items + + def get_repos(self) -> list[dict]: + """Retourne tous les repos de l'utilisateur (pagination automatique). + + Endpoint: GET /api/v1/user/repos + """ + return self._get_paginated("/api/v1/user/repos") + + def get_latest_release(self, owner: str, repo: str) -> dict | None: + """Retourne la derniere release du repo, ou None si aucune. + + Endpoint: GET /api/v1/repos/{owner}/{repo}/releases/latest + Gere HTTP 404 en retournant None. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/releases/latest" + resp = self.session.get(url) + if resp.status_code == 404: + return None + return resp.json() + + def get_milestones(self, owner: str, repo: str) -> list[dict]: + """Retourne les milestones ouvertes du repo. + + Endpoint: GET /api/v1/repos/{owner}/{repo}/milestones?state=open + """ + return self._get_paginated( + f"/api/v1/repos/{owner}/{repo}/milestones", + params={"state": "open"}, + ) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..9c5e827 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,136 @@ +"""Tests for GiteaClient API client.""" + +from unittest.mock import MagicMock, patch + + +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" + + +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 + + +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