44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_transport_is_reused_across_requests_and_closed():
|
|
client_ids = []
|
|
|
|
async def handler(request):
|
|
client_ids.append(id(gitea_proxy._client))
|
|
return httpx.Response(200, json={"path": request.url.path})
|
|
|
|
client = gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
first = await gitea_proxy.fetch("user")
|
|
second = await gitea_proxy.fetch("user/repos")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert first == {"path": "/api/v1/user"}
|
|
assert second == {"path": "/api/v1/user/repos"}
|
|
assert client_ids == [id(client), id(client)]
|
|
assert client.is_closed
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_application_lifespan_opens_and_closes_gitea_transport(monkeypatch):
|
|
calls = []
|
|
|
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
|
|
|
async def stop_client():
|
|
calls.append("stop")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
|
|
|
|
async with main.app.router.lifespan_context(main.app):
|
|
assert calls == ["start"]
|
|
|
|
assert calls == ["start", "stop"]
|