This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user