210 lines
8.9 KiB
Python
210 lines
8.9 KiB
Python
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 '<script src="static/shared-image-capture.js"></script>' 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" |