From 30f92d18a2fb6fb14336015a0656c0de98fa1ef1 Mon Sep 17 00:00:00 2001 From: timmy Date: Thu, 6 Aug 2026 19:29:21 +0000 Subject: [PATCH] feat: unify live dashboard snapshot (#131) --- frontend/index.html | 29 ++-- src/gitea_proxy.py | 67 ++++++--- src/main.py | 201 +++++++++++++++----------- tests/test_api_cache_policy.py | 4 +- tests/test_api_paths.py | 3 +- tests/test_event_stream.py | 21 ++- tests/test_gitea_transport.py | 43 ++++++ tests/test_gitea_work_search.py | 23 +++ tests/test_live_snapshot.py | 144 ++++++++++++++++++ tests/test_shared_context_snapshot.py | 14 +- 10 files changed, 421 insertions(+), 128 deletions(-) create mode 100644 tests/test_gitea_transport.py create mode 100644 tests/test_live_snapshot.py diff --git a/frontend/index.html b/frontend/index.html index bf42ff9..d9e50cc 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -324,8 +324,8 @@ textarea { resize: vertical; min-height: 120px; } function setClock() { qs('#clock').textContent = fmt(new Date()); } setClock(); setInterval(setClock, 1000); - async function fetchContextSnapshot() { - const res = await fetch('api/v1/context', { headers: { Accept: 'application/json' } }); + async function fetchLiveSnapshot() { + const res = await fetch('api/v1/live', { headers: { Accept: 'application/json' } }); if (!res.ok) throw new Error('HTTP ' + res.status); return res.json(); } @@ -552,14 +552,13 @@ textarea { resize: vertical; min-height: 120px; } qs('#gitea-events-status').textContent = message; } - async function loadEventStream() { - setEventStreamStatus('Updating…'); - try { - const res = await fetch('api/v1/events', { headers: { Accept: 'application/json' } }); - if (!res.ok) throw new Error('HTTP ' + res.status); - paintEventStream(await res.json()); + function renderLiveSnapshot(snapshot) { + if (snapshot.context) renderContextSnapshot(snapshot.context); + else handleContextError(new Error('Context section unavailable')); + if (snapshot.events) { + paintEventStream(snapshot.events); setEventStreamStatus('Updated ' + fmt(new Date())); - } catch (e) { + } else { setEventStreamStatus('Update failed · showing last activity'); } } @@ -664,9 +663,12 @@ textarea { resize: vertical; min-height: 120px; } }); const contextPoller = createContextPoller({ - fetchContext: fetchContextSnapshot, - onSnapshot: renderContextSnapshot, - onError: handleContextError, + fetchContext: fetchLiveSnapshot, + onSnapshot: renderLiveSnapshot, + onError: error => { + handleContextError(error); + setEventStreamStatus('Update failed · showing last activity'); + }, isHidden: () => document.hidden, intervalMs: 8000, }); @@ -689,11 +691,8 @@ textarea { resize: vertical; min-height: 120px; } }); }); contextPoller.start(); - loadEventStream(); - setInterval(() => { if (!document.hidden) loadEventStream(); }, 5000); document.addEventListener('visibilitychange', () => { contextPoller.setVisible(!document.hidden); - if (!document.hidden) loadEventStream(); }); /* Widgets */ diff --git a/src/gitea_proxy.py b/src/gitea_proxy.py index 805c78b..f16ce0f 100644 --- a/src/gitea_proxy.py +++ b/src/gitea_proxy.py @@ -9,6 +9,7 @@ GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/") GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") REVIEW_DIFF_MAX_BYTES = 64 * 1024 REVIEW_DIFF_MAX_LINES = 400 +_client: httpx.AsyncClient | None = None def _auth() -> dict[str, str]: @@ -18,30 +19,47 @@ def _auth() -> dict[str, str]: return headers +def start_client(**kwargs) -> httpx.AsyncClient: + """Create the application-lifetime Gitea transport.""" + global _client + _client = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs) + return _client + + +def _get_client() -> httpx.AsyncClient: + if _client is None or _client.is_closed: + return start_client() + return _client + + +async def stop_client() -> None: + global _client + if _client is not None and not _client.is_closed: + await _client.aclose() + + async def fetch(path: str) -> Any: - async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client: - r = await client.get(f"/api/v1/{path}", headers=_auth()) - r.raise_for_status() - return r.json() + r = await _get_client().get(f"/api/v1/{path}", headers=_auth()) + r.raise_for_status() + return r.json() async def fetch_text(path: str, max_bytes: int) -> tuple[str, bool]: chunks: list[bytes] = [] size = 0 truncated = False - async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client: - async with client.stream( - "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"} - ) as response: - response.raise_for_status() - async for chunk in response.aiter_bytes(): - remaining = max_bytes - size - if len(chunk) > remaining: - chunks.append(chunk[:remaining]) - truncated = True - break - chunks.append(chunk) - size += len(chunk) + async with _get_client().stream( + "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"} + ) as response: + response.raise_for_status() + async for chunk in response.aiter_bytes(): + remaining = max_bytes - size + if len(chunk) > remaining: + chunks.append(chunk[:remaining]) + truncated = True + break + chunks.append(chunk) + size += len(chunk) return b"".join(chunks).decode("utf-8", errors="replace"), truncated @@ -100,11 +118,11 @@ async def issues() -> list[dict]: async def pull_requests() -> list[dict]: - assigned = await fetch( - "repos/issues/search?state=open&assigned=true&type=pulls&limit=50" - ) - review_requested = await fetch( - "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50" + assigned, review_requested = await asyncio.gather( + fetch("repos/issues/search?state=open&assigned=true&type=pulls&limit=50"), + fetch( + "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50" + ), ) merged: dict[int, dict] = {} for reason, pulls in ( @@ -201,8 +219,9 @@ async def pull_review_detail(repository: str, number: int) -> dict: } -async def activity_events() -> list[dict]: - user = await current_user() +async def activity_events(user: dict | None = None) -> list[dict]: + if user is None: + user = await current_user() events = await fetch(f"users/{user['login']}/activities/feeds?limit=20") if events is None: events = [] diff --git a/src/main.py b/src/main.py index ea6150f..77f163e 100644 --- a/src/main.py +++ b/src/main.py @@ -1,5 +1,6 @@ import asyncio import math +from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI, HTTPException @@ -7,6 +8,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles +from src import gitea_proxy from src.gitea_proxy import ( activity_events, current_user, @@ -20,12 +22,22 @@ from src.models import Issue, PullRequest, Repo, User from src.suggestion_engine import compute from src.views import router as frontend_router -app = FastAPI(title="Stackchain Dashboard") +@asynccontextmanager +async def lifespan(_app: FastAPI): + gitea_proxy.start_client() + try: + yield + finally: + await gitea_proxy.stop_client() + + +app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan) CONTEXT_TIMEOUT_SECONDS = 5.0 EVENT_STREAM_TIMEOUT_SECONDS = 5.0 READINESS_TIMEOUT_SECONDS = 5.0 REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend" +_live_snapshot_task: asyncio.Task | None = None class ContextPayloadError(ValueError): @@ -35,6 +47,52 @@ class ContextPayloadError(ValueError): class ReadinessPayloadError(ValueError): """Raised when Gitea returns a structurally invalid readiness payload.""" + +def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict: + user_model = User( + id=user_data["id"], + login=user_data["login"], + full_name=user_data.get("full_name") or "", + email=user_data.get("email") or "", + ) + repo_models = [ + Repo( + id=r["id"], name=r["name"], full_name=r["full_name"], + description=r.get("description") or "", url=r["html_url"], + updated_at=r.get("updated_at", ""), + ) + for r in (repo_data or [])[:50] + if isinstance(r, dict) + and all(field in r for field in ("id", "name", "full_name", "html_url")) + ] + issue_models = [ + Issue( + id=i["id"], number=i["number"], title=i["title"], state=i["state"], + labels=[label.get("name", "") for label in (i.get("labels") or []) if isinstance(label, dict)], + assignees=[assignee.get("login", "") for assignee in (i.get("assignees") or []) if isinstance(assignee, dict)], + repository=i["repository"].get("full_name", "") if isinstance(i.get("repository"), dict) else "", + updated_at=i.get("updated_at") or "", url=i["html_url"], + ) + for i in (issues_data or [])[:50] + if isinstance(i, dict) + and all(field in i for field in ("id", "number", "title", "state", "html_url")) + ] + pr_models = [ + PullRequest( + id=p["id"], number=p["number"], title=p["title"], state=p["state"], + user=p["user"].get("login", "") if isinstance(p.get("user"), dict) else "", + labels=[label.get("name", "") for label in (p.get("labels") or []) if isinstance(label, dict)], + assignees=[assignee.get("login", "") for assignee in (p.get("assignees") or []) if isinstance(assignee, dict)], + work_reasons=[reason for reason in (p.get("work_reasons") or []) if reason in ("assigned_to_me", "review_requested")], + repository=p["repository"].get("full_name", "") if isinstance(p.get("repository"), dict) else "", + updated_at=p.get("updated_at") or "", url=p["html_url"], + ) + for p in (prs_data or [])[:50] + if isinstance(p, dict) + and all(field in p for field in ("id", "number", "title", "state", "html_url")) + ] + return compute(user_model, repo_models, issue_models, pr_models).model_dump() + app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -50,7 +108,7 @@ app.include_router(frontend_router) @app.middleware("http") async def prevent_live_api_caching(request, call_next): response = await call_next(request) - if request.url.path in {"/api/v1/context", "/api/v1/events"} or ( + if request.url.path in {"/api/v1/context", "/api/v1/events", "/api/v1/live"} or ( request.url.path.startswith("/api/v1/repos/") and request.url.path.endswith("/review") ): @@ -145,87 +203,68 @@ async def context() -> JSONResponse: }, headers={ "Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS))) } if isinstance(e, TimeoutError) else None) - user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name") or "", email=user_data.get("email") or "") - repo_models = [ - Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description") or "", url=r["html_url"], updated_at=r.get("updated_at", "")) - for r in (repo_data or [])[:50] - if isinstance(r, dict) - and all(field in r for field in ("id", "name", "full_name", "html_url")) - ] - issue_models = [ - Issue( - id=i["id"], - number=i["number"], - title=i["title"], - state=i["state"], - labels=[ - label.get("name", "") - for label in (i.get("labels") or []) - if isinstance(label, dict) - ], - assignees=[ - assignee.get("login", "") - for assignee in (i.get("assignees") or []) - if isinstance(assignee, dict) - ], - repository=( - i["repository"].get("full_name", "") - if isinstance(i.get("repository"), dict) - else "" - ), - updated_at=i.get("updated_at") or "", - url=i["html_url"], + return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data)) + + +async def _load_context_for_user(user_data: dict) -> dict: + repo_data, issues_data, prs_data = await asyncio.gather( + repos(), issues(), pull_requests() + ) + if not all(field in user_data for field in ("id", "login")): + raise ContextPayloadError("Gitea current-user response did not include id and login") + if any( + data is not None and not isinstance(data, list) + for data in (repo_data, issues_data, prs_data) + ): + raise ContextPayloadError("Gitea collection response was not a list") + return _context_payload(user_data, repo_data, issues_data, prs_data) + + +async def _build_live_snapshot() -> dict: + user_data = await current_user() + if not isinstance(user_data, dict) or not user_data.get("login"): + raise ContextPayloadError("Gitea current-user response was invalid") + context_result, events_result = await asyncio.gather( + _load_context_for_user(user_data), + activity_events(user_data), + return_exceptions=True, + ) + context_ok = not isinstance(context_result, BaseException) + events_ok = not isinstance(events_result, BaseException) + return { + "context": context_result if context_ok else None, + "events": events_result if events_ok else None, + "sections": { + "context": "fresh" if context_ok else "temporarily unavailable", + "events": "fresh" if events_ok else "temporarily unavailable", + }, + } + + +@app.get("/api/v1/live") +async def live_snapshot() -> JSONResponse: + """Return a fresh, section-aware snapshot; join only an active identical load.""" + global _live_snapshot_task + if _live_snapshot_task is None or _live_snapshot_task.done(): + _live_snapshot_task = asyncio.create_task(_build_live_snapshot()) + task = _live_snapshot_task + try: + result = await asyncio.wait_for(task, timeout=CONTEXT_TIMEOUT_SECONDS) + return JSONResponse(result) + except TimeoutError: + return JSONResponse( + {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"}, + status_code=503, + headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))}, ) - for i in (issues_data or [])[:50] - if isinstance(i, dict) - and all( - field in i - for field in ("id", "number", "title", "state", "html_url") + except Exception: + return JSONResponse( + {"error": "Gitea live snapshot is temporarily unavailable"}, + status_code=503, ) - ] - pr_models = [ - PullRequest( - id=p["id"], - number=p["number"], - title=p["title"], - state=p["state"], - user=( - p["user"].get("login", "") - if isinstance(p.get("user"), dict) - else "" - ), - labels=[ - label.get("name", "") - for label in (p.get("labels") or []) - if isinstance(label, dict) - ], - assignees=[ - assignee.get("login", "") - for assignee in (p.get("assignees") or []) - if isinstance(assignee, dict) - ], - work_reasons=[ - reason - for reason in (p.get("work_reasons") or []) - if reason in ("assigned_to_me", "review_requested") - ], - repository=( - p["repository"].get("full_name", "") - if isinstance(p.get("repository"), dict) - else "" - ), - updated_at=p.get("updated_at") or "", - url=p["html_url"], - ) - for p in (prs_data or [])[:50] - if isinstance(p, dict) - and all( - field in p - for field in ("id", "number", "title", "state", "html_url") - ) - ] - ctx = compute(user_model, repo_models, issue_models, pr_models) - return JSONResponse(ctx.model_dump()) + finally: + if task.done() and _live_snapshot_task is task: + _live_snapshot_task = None @app.get("/api/v1/events") diff --git a/tests/test_api_cache_policy.py b/tests/test_api_cache_policy.py index c4a69fa..7fd8c75 100644 --- a/tests/test_api_cache_policy.py +++ b/tests/test_api_cache_policy.py @@ -12,7 +12,7 @@ async def test_live_gitea_api_responses_cannot_be_stored_by_shared_caches(monkey async def empty_collection(): return [] - async def empty_events(): + async def empty_events(user_data=None): return [] monkeypatch.setattr(main, "current_user", user) @@ -23,7 +23,7 @@ async def test_live_gitea_api_responses_cannot_be_stored_by_shared_caches(monkey transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: - for path in ("/api/v1/context", "/api/v1/events"): + for path in ("/api/v1/context", "/api/v1/events", "/api/v1/live"): response = await client.get(path) assert response.status_code == 200 diff --git a/tests/test_api_paths.py b/tests/test_api_paths.py index 6bdc3c7..4656e6a 100644 --- a/tests/test_api_paths.py +++ b/tests/test_api_paths.py @@ -15,6 +15,5 @@ def test_api_requests_resolve_inside_dashboard_subpath(): urljoin("https://forge.alexanderwhitestone.com/dashboard/", path) for path in api_paths } == { - "https://forge.alexanderwhitestone.com/dashboard/api/v1/context", - "https://forge.alexanderwhitestone.com/dashboard/api/v1/events", + "https://forge.alexanderwhitestone.com/dashboard/api/v1/live", } diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py index 5f48776..aa37730 100644 --- a/tests/test_event_stream.py +++ b/tests/test_event_stream.py @@ -17,8 +17,9 @@ def test_dashboard_has_realtime_gitea_event_stream_widget(): assert "Gitea event stream" in html assert "id=\"gitea-events\"" in html assert "function paintEventStream" in html - assert "setInterval(() => { if (!document.hidden) loadEventStream(); }, 5000)" in html - assert "if (!document.hidden) loadEventStream();" in html + assert "fetch('api/v1/live'" in html + assert "paintEventStream(snapshot.events)" in html + assert "loadEventStream" not in html assert "event.actor?.login" in html assert "event.repo?.full_name" in html @@ -27,7 +28,6 @@ def test_event_stream_reports_when_activity_was_refreshed(): html = DASHBOARD.read_text() assert 'id="gitea-events-status"' in html - assert "setEventStreamStatus('Updating…')" in html assert "setEventStreamStatus('Updated ' + fmt(new Date()))" in html @@ -121,6 +121,21 @@ async def test_activity_events_fetches_feed_for_authenticated_user(monkeypatch): assert paths == ["users/timmy/activities/feeds?limit=20"] +@pytest.mark.anyio +async def test_activity_events_reuses_supplied_authenticated_user(monkeypatch): + async def unexpected_current_user(): + raise AssertionError("current user must not be fetched twice") + + async def fake_fetch(path): + assert path == "users/timmy/activities/feeds?limit=20" + return [] + + monkeypatch.setattr(gitea_proxy, "current_user", unexpected_current_user) + monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch) + + assert await gitea_proxy.activity_events({"login": "timmy"}) == [] + + @pytest.mark.anyio async def test_activity_events_skips_malformed_feed_entries(monkeypatch): async def fake_current_user(): diff --git a/tests/test_gitea_transport.py b/tests/test_gitea_transport.py new file mode 100644 index 0000000..4985cab --- /dev/null +++ b/tests/test_gitea_transport.py @@ -0,0 +1,43 @@ +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"] diff --git a/tests/test_gitea_work_search.py b/tests/test_gitea_work_search.py index 979ec24..83fcb12 100644 --- a/tests/test_gitea_work_search.py +++ b/tests/test_gitea_work_search.py @@ -55,6 +55,29 @@ async def test_pull_requests_merge_assignment_and_review_responsibilities(monkey assert pulls[1]["work_reasons"] == ["review_requested"] +@pytest.mark.anyio +async def test_pull_request_searches_start_concurrently(monkeypatch): + started = [asyncio.Event(), asyncio.Event()] + release = asyncio.Event() + + async def fake_fetch(path): + index = 0 if "assigned=true" in path else 1 + started[index].set() + await release.wait() + return [] + + monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch) + task = asyncio.create_task(gitea_proxy.pull_requests()) + try: + await asyncio.wait_for( + asyncio.gather(*(event.wait() for event in started)), timeout=1 + ) + finally: + release.set() + + assert await task == [] + + @pytest.mark.anyio async def test_requested_review_guard_uses_only_dedicated_review_search(monkeypatch): requested_paths = [] diff --git a/tests/test_live_snapshot.py b/tests/test_live_snapshot.py new file mode 100644 index 0000000..4f11111 --- /dev/null +++ b/tests/test_live_snapshot.py @@ -0,0 +1,144 @@ +import asyncio +import json + +import pytest + +from src import main + + +@pytest.fixture(autouse=True) +def reset_live_snapshot_task(): + main._live_snapshot_task = None + yield + main._live_snapshot_task = None + + +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"}] + + 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) + + 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["sections"] == {"context": "fresh", "events": "fresh"} + + +@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) + + 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", + } + assert "secret" 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) + + result = payload(await main.live_snapshot()) + + assert result["context"] is None + assert result["events"] == [{"type": "push"}] + assert result["sections"] == { + "context": "temporarily unavailable", + "events": "fresh", + } + + +@pytest.mark.anyio +async def test_live_snapshot_coalesces_only_simultaneous_requests(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) + + first = asyncio.create_task(main.live_snapshot()) + await asyncio.sleep(0) + second = asyncio.create_task(main.live_snapshot()) + await asyncio.sleep(0) + assert user_calls == 1 + + release.set() + await asyncio.gather(first, second) + await main.live_snapshot() + + assert user_calls == 2 diff --git a/tests/test_shared_context_snapshot.py b/tests/test_shared_context_snapshot.py index 1e1006e..5a2eb4f 100644 --- a/tests/test_shared_context_snapshot.py +++ b/tests/test_shared_context_snapshot.py @@ -14,4 +14,16 @@ async def test_one_context_snapshot_updates_every_context_backed_panel(): assert "setInterval(tickWidgets, 2000)" not in html assert "updateRepoMix(qs('#repo-mix'), fetch)" not in html assert "setInterval(load, 8000)" not in html - assert "document.addEventListener('visibilitychange'" in html \ No newline at end of file + assert "document.addEventListener('visibilitychange'" in html + + +@pytest.mark.anyio +async def test_one_live_snapshot_updates_work_and_activity_on_one_timer(): + html = await dashboard() + + assert "fetch('api/v1/live'" in html + assert "renderLiveSnapshot(snapshot)" in html + assert "renderContextSnapshot(snapshot.context)" in html + assert "paintEventStream(snapshot.events)" in html + assert "loadEventStream" not in html + assert "5000" not in html \ No newline at end of file