from pathlib import Path import httpx import pytest from src import main @pytest.mark.anyio async def test_work_page_endpoint_normalizes_requested_page_and_is_not_cacheable(monkeypatch): requested = [] async def page_loader(stream, page): requested.append((stream, page)) return { "stream": stream, "page": page, "total": 84, "has_more": False, "items": [{ "id": 51, "number": 51, "title": "Older issue", "state": "open", "labels": [], "assignees": [{"login": "timmy"}], "repository": {"full_name": "stackchain/api"}, "updated_at": "2026-08-07T10:00:00Z", "due_date": "2026-08-09T23:59:59Z", "milestone": {"id": 9, "title": "August RC", "state": "open"}, "html_url": "https://forge.example/stackchain/api/issues/51", }], } monkeypatch.setattr(main.gitea_proxy, "work_page", page_loader) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/api/v1/work/issue?page=2") assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert requested == [("issue", 2)] assert response.json() == { "stream": "issue", "page": 2, "total": 84, "has_more": False, "items": [{ "id": 51, "number": 51, "title": "Older issue", "state": "open", "labels": [], "assignees": ["timmy"], "repository": "stackchain/api", "work_reasons": [], "updated_at": "2026-08-07T10:00:00Z", "due_date": "2026-08-09T23:59:59Z", "milestone": {"id": 9, "title": "August RC"}, "url": "https://forge.example/stackchain/api/issues/51", }], } @pytest.mark.anyio async def test_work_page_endpoint_rejects_unknown_stream_before_upstream_io(monkeypatch): called = False async def page_loader(stream, page): nonlocal called called = True monkeypatch.setattr(main.gitea_proxy, "work_page", page_loader) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/api/v1/work/unknown?page=2") assert response.status_code == 422 assert called is False @pytest.mark.anyio async def test_filed_work_page_endpoint_preserves_authored_reason(monkeypatch): async def page_loader(stream, page): return { "stream": stream, "page": page, "total": 1, "has_more": False, "items": [{ "id": 870, "number": 870, "title": "Delegated", "state": "open", "labels": [], "assignees": [{"login": "alex"}], "work_reasons": ["created_by_me"], "repository": {"full_name": "stackchain/dashboard"}, "html_url": "https://forge.example/issues/870", }], } monkeypatch.setattr(main.gitea_proxy, "work_page", page_loader) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/api/v1/work/filed?page=2") assert response.status_code == 200 assert response.json()["items"][0]["work_reasons"] == ["created_by_me"] @pytest.mark.anyio async def test_authored_pull_page_endpoint_preserves_authorship_reason(monkeypatch): async def page_loader(stream, page): assert (stream, page) == ("authored", 2) return { "stream": stream, "page": page, "total": 51, "has_more": False, "items": [{ "id": 901, "number": 41, "title": "Ship mobile flow", "state": "open", "user": {"login": "timmy"}, "labels": [], "assignees": [], "work_reasons": ["authored_by_me"], "repository": {"full_name": "stackchain/dashboard"}, "html_url": "https://forge.example/pulls/41", }], } monkeypatch.setattr(main.gitea_proxy, "work_page", page_loader) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/api/v1/work/authored?page=2") assert response.status_code == 200 assert response.json()["items"][0]["work_reasons"] == ["authored_by_me"] def test_initial_context_preserves_authored_pull_reason(): payload = main._context_payload( {"id": 1, "login": "timmy"}, [], [], [{ "id": 901, "number": 41, "title": "Ship mobile flow", "state": "open", "user": {"login": "timmy"}, "labels": [], "assignees": [], "work_reasons": ["authored_by_me"], "repository": {"full_name": "stackchain/dashboard"}, "html_url": "https://forge.example/pulls/41", }], ) assert payload["pull_requests"][0]["work_reasons"] == ["authored_by_me"] @pytest.mark.anyio async def test_pwa_assets_expose_root_scoped_share_target_without_caching_api_data(): transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: manifest = await client.get("/manifest.webmanifest") worker = await client.get("/service-worker.js") assert manifest.status_code == 200 assert manifest.headers["content-type"].startswith("application/manifest+json") assert manifest.json()["id"] == "./" assert manifest.json()["start_url"] == "./" assert manifest.json()["scope"] == "./" assert manifest.json()["share_target"] == { "action": "./share-target", "method": "POST", "enctype": "multipart/form-data", "params": { "title": "title", "text": "text", "url": "url", "files": [{"name": "image", "accept": ["image/png", "image/jpeg", "image/webp"]}], }, } assert manifest.json()["shortcuts"] == [ { "name": "Continue work", "short_name": "Continue", "description": "Resume the highest-priority mobile work flow.", "url": "./?launch=continue", "icons": [{"src": "static/icons/stackchain-192.png", "sizes": "192x192", "type": "image/png"}], }, { "name": "New issue", "short_name": "New", "description": "Capture work now and file it online or offline.", "url": "./?launch=new", "icons": [{"src": "static/icons/stackchain-192.png", "sizes": "192x192", "type": "image/png"}], }, { "name": "Open Agenda", "short_name": "Agenda", "description": "Check every assigned deadline and continue the Agenda session.", "url": "./?launch=agenda", "icons": [{"src": "static/icons/stackchain-192.png", "sizes": "192x192", "type": "image/png"}], }, ] assert worker.status_code == 200 assert worker.headers["content-type"].startswith("application/javascript") assert worker.headers["service-worker-allowed"] == "/" assert "request.url.includes('/api/')" in worker.text assert "request.method !== 'GET'" in worker.text assert "new URL('./', self.location.href).pathname" in worker.text assert "acceptSharedContent(request)" in worker.text assert "sharedImageCapture.consume" in (Path(__file__).resolve().parents[1] / "frontend" / "dashboard.js").read_text() dashboard = (Path(__file__).resolve().parents[1] / "frontend" / "dashboard.js").read_text() assert "restoreContent:content=>" in dashboard assert "issueCapture.stageSharedContent(content)" in dashboard assert '' in (Path(__file__).resolve().parents[1] / "frontend" / "index.html").read_text() @pytest.mark.anyio async def test_dashboard_announces_offline_mode_and_refreshes_after_reconnect(): transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/") assert response.status_code == 200 assert 'id="offline-status"' in response.text assert 'role="status"' in response.text assert 'aria-live="polite"' in response.text bootstrap = (Path(__file__).resolve().parents[1] / "frontend" / "dashboard.js").read_text() assert "window.addEventListener('offline'" in bootstrap assert "window.addEventListener('online'" in bootstrap assert "contextPoller.refresh({ force: true })" in bootstrap @pytest.mark.anyio async def test_dashboard_shell_revalidates_instead_of_being_stored_as_fresh(): transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/") assert response.status_code == 200 assert response.headers["cache-control"] == "no-cache"