317 lines
9.5 KiB
Python
317 lines
9.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_skips_malformed_nested_issue_metadata(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
async def issue_collection():
|
|
return [{
|
|
"id": 85,
|
|
"number": 85,
|
|
"title": "Malformed nested metadata",
|
|
"state": "open",
|
|
"labels": [None, {"name": "bug"}, "invalid"],
|
|
"assignees": [None, {"login": "timmy"}, "invalid"],
|
|
"html_url": "https://forge.example/issues/85",
|
|
}]
|
|
|
|
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"] == ["bug"]
|
|
assert issue["assignees"] == ["timmy"]
|
|
|
|
|
|
@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"] == ""
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_normalizes_null_collection_payloads(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def null_collection():
|
|
return None
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", null_collection)
|
|
monkeypatch.setattr(main, "issues", null_collection)
|
|
monkeypatch.setattr(main, "pull_requests", null_collection)
|
|
|
|
response = await main.context()
|
|
context = json.loads(response.body)
|
|
|
|
assert context["repos"] == []
|
|
assert context["issues"] == []
|
|
assert context["pull_requests"] == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_normalizes_nullable_user_profile_fields(monkeypatch):
|
|
async def user():
|
|
return {
|
|
"id": 1,
|
|
"login": "timmy",
|
|
"full_name": None,
|
|
"email": None,
|
|
}
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
context = json.loads(response.body)
|
|
|
|
assert response.status_code == 200
|
|
assert context["user"]["full_name"] == ""
|
|
assert context["user"]["email"] == ""
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_returns_fallback_for_null_current_user_payload(monkeypatch):
|
|
async def null_user():
|
|
return None
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", null_user)
|
|
monkeypatch.setattr(main, "repos", empty_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
context = json.loads(response.body)
|
|
|
|
assert response.status_code == 200
|
|
assert context["repos"] == []
|
|
assert context["error"] == "Gitea current-user response was not an object"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_skips_malformed_repository_entries(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def repository_collection():
|
|
return [
|
|
{"id": 79, "name": "missing-required-fields"},
|
|
{
|
|
"id": 1,
|
|
"name": "dashboard",
|
|
"full_name": "stackchain/dashboard",
|
|
"description": "Operations dashboard",
|
|
"html_url": "https://forge.example/stackchain/dashboard",
|
|
},
|
|
]
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", repository_collection)
|
|
monkeypatch.setattr(main, "issues", empty_collection)
|
|
monkeypatch.setattr(main, "pull_requests", empty_collection)
|
|
|
|
response = await main.context()
|
|
repositories = json.loads(response.body)["repos"]
|
|
|
|
assert [repository["name"] for repository in repositories] == ["dashboard"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_skips_malformed_issue_entries(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def issue_collection():
|
|
return [
|
|
{"id": 81, "title": "missing-required-fields"},
|
|
{
|
|
"id": 1,
|
|
"number": 81,
|
|
"title": "Valid issue",
|
|
"state": "open",
|
|
"html_url": "https://forge.example/stackchain/dashboard/issues/81",
|
|
},
|
|
]
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
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()
|
|
issues = json.loads(response.body)["issues"]
|
|
|
|
assert [issue["title"] for issue in issues] == ["Valid issue"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_skips_malformed_pull_request_entries(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def pull_request_collection():
|
|
return [
|
|
{"id": 83, "title": "missing-required-fields"},
|
|
{
|
|
"id": 1,
|
|
"number": 83,
|
|
"title": "Valid pull request",
|
|
"state": "open",
|
|
"html_url": "https://forge.example/stackchain/stackchain-dashboard/pulls/83",
|
|
},
|
|
]
|
|
|
|
async def empty_collection():
|
|
return []
|
|
|
|
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_requests = json.loads(response.body)["pull_requests"]
|
|
|
|
assert [pull_request["title"] for pull_request in pull_requests] == [
|
|
"Valid pull request"
|
|
]
|