Spark
5aa0c7d9a7
Modèles (chemin DB, migration 0.1.0, ordre, réconciliation), classification, client Gitea (pagination, en-tête, écritures), API (CSRF, write-back, codes d'erreur, 502 propre), auth Basic et fail-closed. Tous verts; pyflakes propre.
99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
"""Gitea client: pagination, auth header, write operations (audit I2/M9)."""
|
|
|
|
import httpx
|
|
|
|
from app.gitea_client import PAGE_SIZE, GiteaClient
|
|
|
|
|
|
def _client(handler, token=""):
|
|
return GiteaClient(
|
|
base_url="http://gitea.test",
|
|
token=token,
|
|
cache_ttl=0,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
|
|
async def test_pagination_follows_until_short_page():
|
|
seen_pages = []
|
|
|
|
def handler(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"id": 7})
|
|
page = int(request.url.params.get("page", "1"))
|
|
seen_pages.append(page)
|
|
count = PAGE_SIZE if page == 1 else 25
|
|
start = (page - 1) * PAGE_SIZE
|
|
data = [{"id": start + i} for i in range(count)]
|
|
return httpx.Response(200, json={"data": data}, headers={"X-Total-Count": "75"})
|
|
|
|
client = _client(handler)
|
|
repos = await client.search_repos()
|
|
assert len(repos) == 75
|
|
assert seen_pages.count(1) >= 1 and 2 in seen_pages
|
|
|
|
|
|
async def test_pagination_without_total_header():
|
|
def handler(request):
|
|
page = int(request.url.params.get("page", "1"))
|
|
count = PAGE_SIZE if page == 1 else 10
|
|
return httpx.Response(200, json=[{"number": i} for i in range(count)])
|
|
|
|
client = _client(handler)
|
|
issues = await client.get_repo_issues("o", "r", state="open")
|
|
assert len(issues) == PAGE_SIZE + 10
|
|
|
|
|
|
async def test_search_uses_authenticated_user_id_not_hardcoded():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"id": 42})
|
|
seen["uid"] = request.url.params.get("uid")
|
|
return httpx.Response(200, json={"data": []})
|
|
|
|
client = _client(handler)
|
|
await client.search_repos()
|
|
assert seen["uid"] == "42"
|
|
|
|
|
|
async def test_authorization_header_only_with_token():
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["auth"] = request.headers.get("authorization")
|
|
return httpx.Response(200, json={"version": "1.27.3"})
|
|
|
|
await _client(handler, token="").ping()
|
|
assert captured["auth"] is None
|
|
|
|
await _client(handler, token="sekret").ping()
|
|
assert captured["auth"] == "token sekret"
|
|
|
|
|
|
async def test_set_issue_state_patches_gitea():
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["method"] = request.method
|
|
captured["path"] = request.url.path
|
|
captured["json"] = request.content.decode()
|
|
return httpx.Response(200, json={"state": "closed"})
|
|
|
|
client = _client(handler, token="x")
|
|
await client.set_issue_state("Projets", "demo", 3, "closed")
|
|
assert captured["method"] == "PATCH"
|
|
assert captured["path"].endswith("/repos/Projets/demo/issues/3")
|
|
assert '"closed"' in captured["json"]
|
|
|
|
|
|
async def test_create_issue_posts_title():
|
|
def handler(request):
|
|
assert request.method == "POST"
|
|
return httpx.Response(201, json={"number": 9, "title": "Nouvelle"})
|
|
|
|
client = _client(handler, token="x")
|
|
issue = await client.create_issue("Projets", "demo", "Nouvelle")
|
|
assert issue["number"] == 9
|