102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
import datetime
|
|
|
|
import pytest
|
|
import requests
|
|
from pydantic import ValidationError
|
|
|
|
from src.gitea_api import GiteaApi, GiteaApiError
|
|
|
|
|
|
def repo_data(repo_id=42):
|
|
return {
|
|
"id": repo_id,
|
|
"ssh_url": "ssh://git@example.com/alex/repo.git",
|
|
"clone_url": "https://example.com/alex/repo.git",
|
|
"name": "repo",
|
|
"updated_at": datetime.datetime.now(datetime.UTC).isoformat(),
|
|
"owner": {
|
|
"id": 23,
|
|
"login": "alex",
|
|
"email": "alex@example.com",
|
|
},
|
|
}
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status_code=200, data=None, json_error=None):
|
|
self.status_code = status_code
|
|
self._data = data
|
|
self._json_error = json_error
|
|
|
|
def json(self):
|
|
if self._json_error:
|
|
raise self._json_error
|
|
return self._data
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, responses=None, request_error=None):
|
|
self.headers = {}
|
|
self.responses = list(responses or [])
|
|
self.request_error = request_error
|
|
self.calls = []
|
|
|
|
def get(self, url, params, timeout):
|
|
self.calls.append((url, params, timeout))
|
|
if self.request_error:
|
|
raise self.request_error
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def test_get_repositories_fetches_all_pages_with_timeout(monkeypatch):
|
|
session = FakeSession(
|
|
[
|
|
FakeResponse(data=[repo_data(repo_id=1)]),
|
|
FakeResponse(data=[repo_data(repo_id=2)]),
|
|
FakeResponse(data=[]),
|
|
]
|
|
)
|
|
monkeypatch.setattr(requests, "session", lambda: session)
|
|
|
|
repos = GiteaApi(
|
|
endpoint="https://example.com",
|
|
api_token="api-token",
|
|
).get_repositories(page_size=1)
|
|
|
|
assert [repo.repo_id for repo in repos] == [1, 2]
|
|
assert session.headers == {"Authorization": "token api-token"}
|
|
assert session.calls[0][2] == (5, 30)
|
|
|
|
|
|
def test_get_repositories_raises_on_http_error(monkeypatch):
|
|
session = FakeSession([FakeResponse(status_code=500, data=[])])
|
|
monkeypatch.setattr(requests, "session", lambda: session)
|
|
|
|
with pytest.raises(GiteaApiError, match="status 500"):
|
|
GiteaApi(
|
|
endpoint="https://example.com",
|
|
api_token="api-token",
|
|
).get_repositories()
|
|
|
|
|
|
def test_get_repositories_raises_on_network_error(monkeypatch):
|
|
session = FakeSession(request_error=requests.Timeout("timed out"))
|
|
monkeypatch.setattr(requests, "session", lambda: session)
|
|
|
|
with pytest.raises(GiteaApiError, match="request failed"):
|
|
GiteaApi(
|
|
endpoint="https://example.com",
|
|
api_token="api-token",
|
|
).get_repositories()
|
|
|
|
|
|
def test_get_repositories_raises_on_invalid_payload(monkeypatch):
|
|
session = FakeSession([FakeResponse(data=[{"id": 1}])])
|
|
monkeypatch.setattr(requests, "session", lambda: session)
|
|
|
|
with pytest.raises((GiteaApiError, ValidationError)):
|
|
GiteaApi(
|
|
endpoint="https://example.com",
|
|
api_token="api-token",
|
|
).get_repositories()
|