122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_returns_fallback_when_gitea_exceeds_deadline(monkeypatch):
|
|
cancelled = asyncio.Event()
|
|
|
|
async def hanging_current_user():
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
cancelled.set()
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "CONTEXT_TIMEOUT_SECONDS", 0.01)
|
|
monkeypatch.setattr(main, "current_user", hanging_current_user)
|
|
monkeypatch.setattr(main, "repos", empty_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["retry-after"] == "1"
|
|
assert b'"repos":[]' in response.body
|
|
assert b'"error":"Gitea context request timed out after 0.01s"' in response.body
|
|
assert cancelled.is_set()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_normalizes_nullable_issue_collections(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
async def issue_collection():
|
|
return [{
|
|
"id": 50,
|
|
"number": 50,
|
|
"title": "Nullable metadata",
|
|
"state": "open",
|
|
"labels": None,
|
|
"assignees": None,
|
|
"html_url": "https://forge.example/issues/50",
|
|
}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty_collection)
|
|
monkeypatch.setattr(main, "issues", issue_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
issue = json.loads(response.body)["issues"][0]
|
|
|
|
assert issue["labels"] == []
|
|
assert issue["assignees"] == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_normalizes_nullable_repository_description(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
async def repo_collection():
|
|
return [{
|
|
"id": 54,
|
|
"name": "dashboard",
|
|
"full_name": "stackchain/dashboard",
|
|
"description": None,
|
|
"html_url": "https://forge.example/stackchain/dashboard",
|
|
}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", repo_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
repository = json.loads(response.body)["repos"][0]
|
|
|
|
assert repository["description"] == ""
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_normalizes_nullable_pull_request_author(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
async def pull_request_collection():
|
|
return [{
|
|
"id": 56,
|
|
"number": 56,
|
|
"title": "Nullable author",
|
|
"state": "open",
|
|
"user": None,
|
|
"html_url": "https://forge.example/pulls/56",
|
|
}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", pull_request_collection)
|
|
|
|
response = await main.context()
|
|
pull_request = json.loads(response.body)["pull_requests"][0]
|
|
|
|
assert pull_request["user"] == "" |