405 lines
12 KiB
Python
405 lines
12 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_log_issue_time_uses_canonical_shared_issue_endpoint_and_seconds():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(201, json={"id": 1})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
await gitea_proxy.log_issue_time("pull:stackchain/stackchain-dashboard:586:", 2520)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 1
|
|
assert requests[0].method == "POST"
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/stackchain-dashboard/issues/586/times"
|
|
assert requests[0].read() == b'{"time":2520}'
|
|
|
|
|
|
@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_gitea_transport_coalesces_identical_concurrent_gets():
|
|
calls = 0
|
|
release = asyncio.Event()
|
|
|
|
async def handler(request):
|
|
nonlocal calls
|
|
calls += 1
|
|
await release.wait()
|
|
return httpx.Response(200, json={"path": request.url.path})
|
|
|
|
gitea_proxy.start_client(
|
|
transport=httpx.MockTransport(handler),
|
|
max_concurrency=4,
|
|
admission_timeout=0.1,
|
|
)
|
|
try:
|
|
requests = [asyncio.create_task(gitea_proxy.fetch("user")) for _ in range(20)]
|
|
await asyncio.sleep(0)
|
|
release.set()
|
|
results = await asyncio.gather(*requests)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert calls == 1
|
|
assert results == [{"path": "/api/v1/user"}] * 20
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cancelling_one_coalesced_waiter_keeps_the_shared_get_running():
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def handler(request):
|
|
started.set()
|
|
await release.wait()
|
|
return httpx.Response(200, json={"path": request.url.path})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
owner = asyncio.create_task(gitea_proxy.fetch("user"))
|
|
try:
|
|
await started.wait()
|
|
waiter = asyncio.create_task(gitea_proxy.fetch("user"))
|
|
await asyncio.sleep(0)
|
|
waiter.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await waiter
|
|
release.set()
|
|
assert await owner == {"path": "/api/v1/user"}
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(owner, return_exceptions=True)
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_transport_rejects_reads_beyond_bounded_capacity():
|
|
active = 0
|
|
peak = 0
|
|
saturated = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def handler(request):
|
|
nonlocal active, peak
|
|
active += 1
|
|
peak = max(peak, active)
|
|
if active == 2:
|
|
saturated.set()
|
|
try:
|
|
await release.wait()
|
|
return httpx.Response(200, json={"path": request.url.path})
|
|
finally:
|
|
active -= 1
|
|
|
|
gitea_proxy.start_client(
|
|
transport=httpx.MockTransport(handler),
|
|
max_concurrency=3,
|
|
admission_timeout=0.02,
|
|
)
|
|
first = asyncio.create_task(gitea_proxy.fetch("first"))
|
|
second = asyncio.create_task(gitea_proxy.fetch("second"))
|
|
try:
|
|
await asyncio.wait_for(saturated.wait(), timeout=0.2)
|
|
with pytest.raises(gitea_proxy.GiteaOverloadedError):
|
|
await gitea_proxy.fetch("third")
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(first, second)
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert peak == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_transport_reserves_one_slot_for_mutations_and_bounds_them():
|
|
active = 0
|
|
peak = 0
|
|
reads_started = asyncio.Event()
|
|
mutation_started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def handler(request):
|
|
nonlocal active, peak
|
|
active += 1
|
|
peak = max(peak, active)
|
|
if active == 2:
|
|
reads_started.set()
|
|
if request.method == "POST":
|
|
mutation_started.set()
|
|
try:
|
|
await release.wait()
|
|
return httpx.Response(200, json={"method": request.method})
|
|
finally:
|
|
active -= 1
|
|
|
|
client = gitea_proxy.start_client(
|
|
transport=httpx.MockTransport(handler),
|
|
max_concurrency=3,
|
|
admission_timeout=0.02,
|
|
)
|
|
reads = [
|
|
asyncio.create_task(gitea_proxy.fetch("first")),
|
|
asyncio.create_task(gitea_proxy.fetch("second")),
|
|
]
|
|
mutation = None
|
|
try:
|
|
await asyncio.wait_for(reads_started.wait(), timeout=0.2)
|
|
mutation = asyncio.create_task(client.post("/api/v1/mutate", json={"value": 1}))
|
|
await asyncio.wait_for(mutation_started.wait(), timeout=0.2)
|
|
with pytest.raises(gitea_proxy.GiteaOverloadedError):
|
|
await asyncio.wait_for(
|
|
client.post("/api/v1/mutate", json={"value": 1}), timeout=0.1
|
|
)
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(*reads, *(task for task in [mutation] if task is not None))
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert peak == 3
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_transport_reads_bulkhead_configuration_from_environment(monkeypatch):
|
|
monkeypatch.setenv("GITEA_MAX_CONCURRENCY", "12")
|
|
monkeypatch.setenv("GITEA_ADMISSION_TIMEOUT_SECONDS", "0.75")
|
|
|
|
client = gitea_proxy.start_client(
|
|
transport=httpx.MockTransport(lambda request: httpx.Response(200, json={}))
|
|
)
|
|
try:
|
|
assert client.max_concurrency == 12
|
|
assert client.admission_timeout == 0.75
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_streaming_reads_share_the_same_bounded_read_capacity():
|
|
active = 0
|
|
saturated = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def handler(request):
|
|
nonlocal active
|
|
active += 1
|
|
if active == 2:
|
|
saturated.set()
|
|
try:
|
|
await release.wait()
|
|
return httpx.Response(200, content=b"diff")
|
|
finally:
|
|
active -= 1
|
|
|
|
gitea_proxy.start_client(
|
|
transport=httpx.MockTransport(handler),
|
|
max_concurrency=3,
|
|
admission_timeout=0.02,
|
|
)
|
|
streams = [
|
|
asyncio.create_task(gitea_proxy.fetch_text("first", 100)),
|
|
asyncio.create_task(gitea_proxy.fetch_text("second", 100)),
|
|
]
|
|
try:
|
|
await asyncio.wait_for(saturated.wait(), timeout=0.2)
|
|
with pytest.raises(gitea_proxy.GiteaOverloadedError):
|
|
await asyncio.wait_for(gitea_proxy.fetch("third"), timeout=0.1)
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(*streams)
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@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)"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_application_shutdown_drains_authored_mutation_before_transport(monkeypatch):
|
|
calls = []
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def authored_mutation():
|
|
started.set()
|
|
await release.wait()
|
|
calls.append("mutation persisted")
|
|
return {"id": 461}
|
|
|
|
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)
|
|
context = main.app.router.lifespan_context(main.app)
|
|
await context.__aenter__()
|
|
task = asyncio.create_task(authored_mutation())
|
|
main._authored_action_operations["shutdown-drain-461"] = (
|
|
("comment", "stackchain/dashboard", 461),
|
|
task,
|
|
0.0,
|
|
)
|
|
await started.wait()
|
|
|
|
shutdown = asyncio.create_task(context.__aexit__(None, None, None))
|
|
await asyncio.sleep(0)
|
|
|
|
assert not shutdown.done()
|
|
assert calls == ["start"]
|
|
|
|
release.set()
|
|
await shutdown
|
|
|
|
assert calls == ["start", "mutation persisted", "stop"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_application_shutdown_bounds_and_settles_stalled_authored_mutation(monkeypatch):
|
|
calls = []
|
|
started = asyncio.Event()
|
|
|
|
async def stalled_mutation():
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
calls.append("mutation cancelled")
|
|
|
|
monkeypatch.setattr(main, "AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", 0.01)
|
|
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)
|
|
context = main.app.router.lifespan_context(main.app)
|
|
await context.__aenter__()
|
|
task = asyncio.create_task(stalled_mutation())
|
|
main._authored_action_operations["stalled-shutdown-461"] = (
|
|
("review", "stackchain/dashboard", 461),
|
|
task,
|
|
0.0,
|
|
)
|
|
await started.wait()
|
|
|
|
await asyncio.wait_for(context.__aexit__(None, None, None), timeout=0.2)
|
|
|
|
assert task.cancelled()
|
|
assert main._authored_action_operations == {}
|
|
assert calls == ["start", "mutation cancelled", "stop"]
|