775 lines
25 KiB
Python
775 lines
25 KiB
Python
import asyncio
|
|
import json
|
|
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.live_snapshot_store import LiveSnapshotStore
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_live_snapshot_task(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_live_snapshot_store",
|
|
LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()),
|
|
)
|
|
main._live_snapshot_task = None
|
|
main._live_snapshot_value = None
|
|
main._live_snapshot_created_at = None
|
|
main._live_section_created_at = {
|
|
section: None for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_section_failure_count = {
|
|
section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_section_retry_at = {
|
|
section: None for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_snapshot_refreshing_sections = set()
|
|
main._live_section_revisions = {
|
|
section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
yield
|
|
main._live_snapshot_task = None
|
|
main._live_snapshot_value = None
|
|
main._live_snapshot_created_at = None
|
|
main._live_section_created_at = {
|
|
section: None for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_section_failure_count = {
|
|
section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_section_retry_at = {
|
|
section: None for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_snapshot_refreshing_sections = set()
|
|
main._live_section_revisions = {
|
|
section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
|
|
|
|
def payload(response):
|
|
return json.loads(response.body)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_fetches_user_once_and_updates_work_and_activity(monkeypatch):
|
|
calls = {"user": 0}
|
|
|
|
async def user():
|
|
calls["user"] += 1
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def events(authenticated_user):
|
|
assert authenticated_user["login"] == "timmy"
|
|
return [{"type": "push"}]
|
|
|
|
async def updates():
|
|
return {
|
|
"items": [{"id": 42, "title": "Mentioned you"}],
|
|
"page": 1,
|
|
"total": 125,
|
|
"has_more": True,
|
|
}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", updates)
|
|
|
|
response = await main.live_snapshot()
|
|
result = payload(response)
|
|
|
|
assert calls["user"] == 1
|
|
assert result["context"]["user"]["login"] == "timmy"
|
|
assert result["events"] == [{"type": "push"}]
|
|
assert result["notifications"] == [{"id": 42, "title": "Mentioned you"}]
|
|
assert result["notification_pagination"] == {
|
|
"page": 1, "total": 125, "has_more": True
|
|
}
|
|
assert result["sections"] == {
|
|
"context": "fresh", "events": "fresh", "notifications": "fresh"
|
|
}
|
|
assert set(result["revisions"]) == {"context", "events", "notifications"}
|
|
assert len(set(result["revisions"].values())) == 1
|
|
assert result["revisions"]["context"].endswith(".1")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def events(_authenticated_user):
|
|
return [{"type": "push"}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
|
|
first = payload(await main.live_snapshot())
|
|
second = payload(await main.live_snapshot(
|
|
context_revision=first["revisions"]["context"],
|
|
events_revision=first["revisions"]["events"],
|
|
notifications_revision=first["revisions"]["notifications"],
|
|
))
|
|
|
|
assert "context" not in second
|
|
assert "events" not in second
|
|
assert "notifications" not in second
|
|
assert "notification_pagination" not in second
|
|
assert second["sections"] == first["sections"]
|
|
assert second["revisions"] == first["revisions"]
|
|
assert "freshness" in second
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", lambda _user: empty())
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
worker_a = payload(await main.live_snapshot())
|
|
worker_a_token = worker_a["revisions"]["context"]
|
|
|
|
main._live_snapshot_value = None
|
|
main._live_snapshot_created_at = None
|
|
main._live_section_created_at = {
|
|
section: None for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_section_revisions = {
|
|
section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
|
|
}
|
|
main._live_snapshot_store = LiveSnapshotStore(
|
|
tmp_path / "other-worker.sqlite3", clock=lambda: main._live_snapshot_clock()
|
|
)
|
|
|
|
worker_b = payload(await main.live_snapshot(context_revision=worker_a_token))
|
|
|
|
assert worker_a_token.endswith(".1")
|
|
assert worker_b["revisions"]["context"].endswith(".1")
|
|
assert worker_b["revisions"]["context"] != worker_a_token
|
|
assert worker_b["context"]["user"]["login"] == "timmy"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_rejects_malformed_revision_before_upstream_work(monkeypatch):
|
|
async def unexpected_user():
|
|
raise AssertionError("malformed revision reached upstream work")
|
|
|
|
monkeypatch.setattr(main, "current_user", unexpected_user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get("/api/v1/live?context_revision=not-a-token")
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_keeps_fresh_context_when_activity_fails(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def failing_events(authenticated_user):
|
|
raise ConnectionError("secret upstream detail")
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", failing_events)
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
|
|
result = payload(await main.live_snapshot())
|
|
|
|
assert result["context"]["user"]["login"] == "timmy"
|
|
assert result["events"] is None
|
|
assert result["sections"] == {
|
|
"context": "fresh",
|
|
"events": "temporarily unavailable",
|
|
"notifications": "fresh",
|
|
}
|
|
assert "secret" not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_keeps_work_and_activity_when_notifications_fail(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def events(_authenticated_user):
|
|
return [{"type": "push"}]
|
|
|
|
async def failing_updates():
|
|
raise ConnectionError("private notification failure")
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", failing_updates)
|
|
|
|
result = payload(await main.live_snapshot())
|
|
|
|
assert result["context"]["user"]["login"] == "timmy"
|
|
assert result["events"] == [{"type": "push"}]
|
|
assert result["notifications"] is None
|
|
assert result["sections"]["notifications"] == "temporarily unavailable"
|
|
assert "private" not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def failing_repos():
|
|
raise ConnectionError("work unavailable")
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def events(authenticated_user):
|
|
return [{"type": "push"}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", failing_repos)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
|
|
result = payload(await main.live_snapshot())
|
|
|
|
assert result["context"] is None
|
|
assert result["events"] == [{"type": "push"}]
|
|
assert result["sections"] == {
|
|
"context": "temporarily unavailable",
|
|
"events": "fresh",
|
|
"notifications": "fresh",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_reuses_completed_snapshot_inside_freshness_window(monkeypatch):
|
|
user_calls = 0
|
|
release = asyncio.Event()
|
|
|
|
async def user():
|
|
nonlocal user_calls
|
|
user_calls += 1
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def blocked_repos():
|
|
await release.wait()
|
|
return []
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
async def events(authenticated_user):
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", blocked_repos)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
|
|
first = asyncio.create_task(main.live_snapshot())
|
|
await asyncio.sleep(0)
|
|
second = asyncio.create_task(main.live_snapshot())
|
|
for _ in range(100):
|
|
if user_calls:
|
|
break
|
|
await asyncio.to_thread(lambda: None)
|
|
assert user_calls == 1
|
|
|
|
release.set()
|
|
await asyncio.gather(first, second)
|
|
await main.live_snapshot()
|
|
|
|
assert user_calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(monkeypatch):
|
|
now = 100.0
|
|
refresh_started = asyncio.Event()
|
|
release_refresh = asyncio.Event()
|
|
builds = 0
|
|
|
|
async def snapshot(_sections=None):
|
|
nonlocal builds
|
|
builds += 1
|
|
if builds == 2:
|
|
refresh_started.set()
|
|
await release_refresh.wait()
|
|
return {
|
|
"context": {"generation": builds},
|
|
"events": [],
|
|
"notifications": [],
|
|
"sections": {},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
|
|
|
|
first = payload(await main.live_snapshot())
|
|
now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
|
|
stale_request = asyncio.create_task(main.live_snapshot())
|
|
await refresh_started.wait()
|
|
await asyncio.sleep(0)
|
|
returned_immediately = stale_request.done()
|
|
release_refresh.set()
|
|
stale = payload(await stale_request)
|
|
|
|
assert returned_immediately is True
|
|
assert first["context"]["generation"] == 1
|
|
assert stale["context"]["generation"] == 1
|
|
assert stale["freshness"]["stale"] is True
|
|
assert stale["freshness"]["revalidating"] is True
|
|
assert builds == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_failed_revalidation_enters_cooldown_and_keeps_last_snapshot(monkeypatch):
|
|
now = 100.0
|
|
builds = 0
|
|
|
|
async def snapshot(_sections=None):
|
|
nonlocal builds
|
|
builds += 1
|
|
if builds > 1:
|
|
raise ConnectionError("private upstream failure")
|
|
return {
|
|
"context": {"generation": 1},
|
|
"events": [{"id": 1}],
|
|
"notifications": [],
|
|
"sections": {
|
|
"context": "fresh",
|
|
"events": "fresh",
|
|
"notifications": "fresh",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
|
|
|
|
await main.live_snapshot()
|
|
now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
|
|
stale = payload(await main.live_snapshot())
|
|
assert main._live_snapshot_task is not None
|
|
await asyncio.gather(main._live_snapshot_task, return_exceptions=True)
|
|
degraded = payload(await main.live_snapshot())
|
|
|
|
assert stale["freshness"]["revalidating"] is True
|
|
assert degraded["context"] == {"generation": 1}
|
|
assert degraded["freshness"]["stale"] is True
|
|
assert degraded["freshness"]["degraded"] is True
|
|
assert degraded["freshness"]["last_refresh_failed"] is True
|
|
assert degraded["freshness"]["revalidating"] is False
|
|
assert degraded["freshness"]["retry_in_seconds"] > 0
|
|
assert builds == 2
|
|
assert "private" not in json.dumps(degraded)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_partial_refresh_updates_fresh_sections_and_retains_failed_sections(monkeypatch):
|
|
now = 100.0
|
|
builds = 0
|
|
|
|
async def snapshot(_sections=None):
|
|
nonlocal builds
|
|
builds += 1
|
|
if builds == 1:
|
|
return {
|
|
"context": {"generation": 1},
|
|
"events": [{"id": "last-known"}],
|
|
"notifications": [{"id": 7}],
|
|
"notification_pagination": {"page": 1, "total": 1, "has_more": False},
|
|
"sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
|
|
}
|
|
return {
|
|
"context": {"generation": 2},
|
|
"events": None,
|
|
"notifications": [{"id": 8}],
|
|
"notification_pagination": {"page": 1, "total": 1, "has_more": False},
|
|
"sections": {
|
|
"context": "fresh",
|
|
"events": "temporarily unavailable",
|
|
"notifications": "fresh",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
|
|
|
|
await main.live_snapshot()
|
|
now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
|
|
await main.live_snapshot()
|
|
assert main._live_snapshot_task is not None
|
|
await main._live_snapshot_task
|
|
result = payload(await main.live_snapshot())
|
|
|
|
assert result["context"] == {"generation": 2}
|
|
assert result["events"] == [{"id": "last-known"}]
|
|
assert result["notifications"] == [{"id": 8}]
|
|
assert result["sections"] == {
|
|
"context": "fresh",
|
|
"events": "stale",
|
|
"notifications": "fresh",
|
|
}
|
|
assert result["freshness"]["degraded"] is True
|
|
assert result["freshness"]["revalidating"] is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_cooldown_does_not_stop_due_work_and_activity_refreshes(monkeypatch):
|
|
now = 100.0
|
|
calls = {"context": 0, "events": 0, "notifications": 0}
|
|
event_refresh_times = []
|
|
notification_outage = False
|
|
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def work():
|
|
calls["context"] += 1
|
|
return []
|
|
|
|
async def empty_work():
|
|
return []
|
|
|
|
async def events(_authenticated_user):
|
|
calls["events"] += 1
|
|
event_refresh_times.append(now)
|
|
return [{"generation": calls["events"]}]
|
|
|
|
async def updates():
|
|
calls["notifications"] += 1
|
|
if notification_outage:
|
|
raise ConnectionError("notifications unavailable")
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", work)
|
|
monkeypatch.setattr(main, "issues", empty_work)
|
|
monkeypatch.setattr(main, "pull_requests", empty_work)
|
|
monkeypatch.setattr(main, "activity_events", events)
|
|
monkeypatch.setattr(main, "notifications", updates)
|
|
|
|
await main.live_snapshot()
|
|
notification_outage = True
|
|
|
|
for elapsed in (
|
|
main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1,
|
|
main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS,
|
|
):
|
|
now += elapsed
|
|
await main.live_snapshot()
|
|
assert main._live_snapshot_task is not None
|
|
await main._live_snapshot_task
|
|
|
|
# Notification backoff is now 10 seconds. The healthy sections become due
|
|
# one second before that cooldown expires and must refresh independently.
|
|
now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
|
|
response = payload(await main.live_snapshot())
|
|
assert main._live_snapshot_task is not None
|
|
await main._live_snapshot_task
|
|
refreshed = payload(await main.live_snapshot())
|
|
|
|
assert calls == {"context": 3, "events": 3, "notifications": 3}
|
|
assert event_refresh_times == [100.0, 109.0, 123.0]
|
|
assert response["freshness"]["sections"]["context"]["revalidating"] is True
|
|
assert refreshed["events"] == [{"generation": 3}]
|
|
assert refreshed["sections"]["notifications"] == "stale"
|
|
assert refreshed["freshness"]["sections"]["notifications"]["retry_in_seconds"] == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state(monkeypatch):
|
|
now = 100.0
|
|
builds = 0
|
|
retry_started = asyncio.Event()
|
|
release_retry = asyncio.Event()
|
|
|
|
async def snapshot(_sections=None):
|
|
nonlocal builds
|
|
builds += 1
|
|
if builds == 1:
|
|
return {
|
|
"context": {"generation": 1}, "events": [], "notifications": [],
|
|
"sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
|
|
}
|
|
if builds == 2:
|
|
return {
|
|
"context": None, "events": [], "notifications": [],
|
|
"sections": {
|
|
"context": "temporarily unavailable",
|
|
"events": "fresh",
|
|
"notifications": "fresh",
|
|
},
|
|
}
|
|
retry_started.set()
|
|
await release_retry.wait()
|
|
return {
|
|
"context": {"generation": 3}, "events": [], "notifications": [],
|
|
"sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
|
|
|
|
await main.live_snapshot()
|
|
now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
|
|
await main.live_snapshot()
|
|
assert main._live_snapshot_task is not None
|
|
await main._live_snapshot_task
|
|
now += main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS
|
|
|
|
first = asyncio.create_task(main.live_snapshot())
|
|
second = asyncio.create_task(main.live_snapshot())
|
|
await asyncio.wait_for(retry_started.wait(), timeout=1.0)
|
|
assert builds == 3
|
|
assert payload(await first)["freshness"]["revalidating"] is True
|
|
assert payload(await second)["freshness"]["revalidating"] is True
|
|
|
|
release_retry.set()
|
|
refresh_task = main._live_snapshot_task
|
|
assert refresh_task is not None
|
|
await refresh_task
|
|
recovered = payload(await main.live_snapshot())
|
|
|
|
assert recovered["context"] == {"generation": 3}
|
|
assert recovered["freshness"]["degraded"] is False
|
|
assert recovered["freshness"]["last_refresh_failed"] is False
|
|
assert recovered["freshness"]["retry_in_seconds"] == 0
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cancelling_one_waiter_does_not_cancel_the_shared_snapshot(monkeypatch):
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
snapshot_calls = 0
|
|
|
|
async def blocked_snapshot(_sections=None):
|
|
nonlocal snapshot_calls
|
|
snapshot_calls += 1
|
|
started.set()
|
|
await release.wait()
|
|
return {
|
|
"context": {},
|
|
"events": [],
|
|
"notifications": [],
|
|
"sections": {},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "_build_live_snapshot", blocked_snapshot)
|
|
|
|
disconnected = asyncio.create_task(main.live_snapshot())
|
|
await started.wait()
|
|
survivor = asyncio.create_task(main.live_snapshot())
|
|
await asyncio.sleep(0)
|
|
|
|
disconnected.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await disconnected
|
|
|
|
assert main._live_snapshot_task is not None
|
|
assert not main._live_snapshot_task.done()
|
|
|
|
release.set()
|
|
response = await survivor
|
|
|
|
assert response.status_code == 200
|
|
assert snapshot_calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_shared_snapshot_deadline_cancels_upstream_work_for_all_waiters(monkeypatch):
|
|
started = asyncio.Event()
|
|
cancelled = asyncio.Event()
|
|
|
|
async def snapshot_that_exceeds_deadline(_sections=None):
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
cancelled.set()
|
|
|
|
monkeypatch.setattr(main, "_build_live_snapshot", snapshot_that_exceeds_deadline)
|
|
monkeypatch.setattr(main, "CONTEXT_TIMEOUT_SECONDS", 0.01)
|
|
|
|
first = asyncio.create_task(main.live_snapshot())
|
|
await started.wait()
|
|
second = asyncio.create_task(main.live_snapshot())
|
|
|
|
try:
|
|
responses = await asyncio.gather(first, second)
|
|
await asyncio.wait_for(cancelled.wait(), timeout=0.1)
|
|
|
|
assert [response.status_code for response in responses] == [503, 503]
|
|
assert [payload(response)["error"] for response in responses] == [
|
|
"Gitea live snapshot timed out after 0.01s",
|
|
"Gitea live snapshot timed out after 0.01s",
|
|
]
|
|
finally:
|
|
task = main._live_snapshot_task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cold_refresh_cooldown_remains_a_retryable_503(monkeypatch):
|
|
now = 100.0
|
|
|
|
async def unavailable(_sections=None):
|
|
raise ConnectionError("private upstream failure")
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
|
|
monkeypatch.setattr(main, "_build_live_snapshot", unavailable)
|
|
|
|
first = await main.live_snapshot()
|
|
second = await main.live_snapshot()
|
|
|
|
assert first.status_code == 503
|
|
assert second.status_code == 503
|
|
assert payload(second) == {
|
|
"error": "Gitea live snapshot is temporarily unavailable"
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cancelled_refresh_releases_shared_lease(monkeypatch):
|
|
started = asyncio.Event()
|
|
|
|
async def blocked(_sections=None):
|
|
started.set()
|
|
await asyncio.Event().wait()
|
|
|
|
monkeypatch.setattr(main, "_build_live_snapshot", blocked)
|
|
request = asyncio.create_task(main.live_snapshot())
|
|
await started.wait()
|
|
assert main._live_snapshot_task is not None
|
|
|
|
main._live_snapshot_task.cancel()
|
|
await asyncio.gather(request, main._live_snapshot_task, return_exceptions=True)
|
|
|
|
replacement = main._live_snapshot_store.try_acquire_refresh(
|
|
{"context"}, lease_seconds=1
|
|
)
|
|
assert replacement is not None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_shared_store_outage_returns_retryable_503(monkeypatch):
|
|
def unavailable():
|
|
raise sqlite3.OperationalError("private database detail")
|
|
|
|
monkeypatch.setattr(main._live_snapshot_store, "load", unavailable)
|
|
|
|
response = await main.live_snapshot()
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["retry-after"] == "1"
|
|
assert payload(response) == {
|
|
"error": "Gitea live snapshot state is temporarily unavailable"
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_independent_workers_reuse_shared_live_snapshot(monkeypatch, tmp_path):
|
|
calls = 0
|
|
|
|
async def user():
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", lambda _user: empty())
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
path = tmp_path / "worker-shared.sqlite3"
|
|
monkeypatch.setattr(main, "_live_snapshot_store", LiveSnapshotStore(path, clock=lambda: 100.0))
|
|
|
|
first = payload(await main.live_snapshot())
|
|
main._live_snapshot_value = None
|
|
main._live_snapshot_created_at = None
|
|
main._live_section_created_at = {section: None for section in main.LIVE_SNAPSHOT_SECTIONS}
|
|
main._live_snapshot_store = LiveSnapshotStore(path, clock=lambda: 100.0)
|
|
second = payload(await main.live_snapshot())
|
|
|
|
assert calls == 1
|
|
assert second["context"] == first["context"]
|
|
assert second["revisions"] == first["revisions"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_live_snapshot_persists_reboot_stable_wall_timestamps(monkeypatch, tmp_path):
|
|
epoch = 1_700_000_000.0
|
|
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "_live_snapshot_clock", lambda: epoch, raising=False)
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_live_snapshot_store",
|
|
LiveSnapshotStore(tmp_path / "wall.sqlite3", clock=lambda: main._live_snapshot_clock()),
|
|
)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
monkeypatch.setattr(main, "activity_events", lambda _user: empty())
|
|
monkeypatch.setattr(main, "notifications", empty)
|
|
|
|
response = await main.live_snapshot()
|
|
|
|
assert response.status_code == 200
|
|
assert set(main._live_snapshot_store.load().created_at.values()) == {epoch}
|