117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
import asyncio
|
|
|
|
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"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_application_shutdown_finishes_snapshot_before_closing_transport(monkeypatch):
|
|
calls = []
|
|
started = asyncio.Event()
|
|
|
|
async def active_snapshot():
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
calls.append("snapshot cancelled")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
|
|
|
async def stop_client():
|
|
assert main._live_snapshot_task is not None
|
|
state = "done" if main._live_snapshot_task.done() else "active"
|
|
calls.append(f"stop ({state})")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
|
|
|
|
try:
|
|
async with main.app.router.lifespan_context(main.app):
|
|
main._live_snapshot_task = asyncio.create_task(active_snapshot())
|
|
await started.wait()
|
|
finally:
|
|
task = main._live_snapshot_task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
main._live_snapshot_task = None
|
|
|
|
assert calls == ["start", "snapshot cancelled", "stop (done)"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_application_shutdown_cancels_available_work_scan_before_transport(monkeypatch):
|
|
calls = []
|
|
started = asyncio.Event()
|
|
|
|
async def active_scan():
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
calls.append("available scan cancelled")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
|
|
|
async def stop_client():
|
|
task = main._available_issue_snapshot_task
|
|
calls.append(f"stop ({'done' if task is not None and task.done() else 'active'})")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
|
|
|
|
try:
|
|
async with main.app.router.lifespan_context(main.app):
|
|
main._available_issue_snapshot_task = asyncio.create_task(active_scan())
|
|
await started.wait()
|
|
finally:
|
|
task = main._available_issue_snapshot_task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
main._available_issue_snapshot_task = None
|
|
|
|
assert calls == ["start", "available scan cancelled", "stop (done)"]
|