import json import subprocess from pathlib import Path import pytest WORKER = Path(__file__).resolve().parents[1] / "frontend" / "service-worker.js" def run_worker_scenario(scenario: str) -> dict: harness = f""" const fs = require('fs'); const vm = require('vm'); const listeners = {{}}; const state = {{ added: [], deleted: [], claimed: false, skipped: false, fetches: [], puts: [], backgroundFlushes: 0, outboxPurges: 0, notifications: [], focused: [], opened: [], failFetch: false, stallFetch: false, lateFetch: false, fetchAborted: false, fetchStatus: 200, fetchRedirected: false, cachedBody: null }}; const storedResponses = new Map(); storedResponses.set( 'https://forge.example/dashboard/__offline-session-lease', new Response(JSON.stringify({{expires_at: Math.floor(Date.now() / 1000) + 3600}})), ); const cache = {{ addAll: async urls => {{ state.added = urls; }}, match: async request => {{ const key = String(request.url || request); if (storedResponses.has(key)) return storedResponses.get(key).clone(); return state.cachedBody === null ? null : new Response(state.cachedBody); }}, put: async (request, response) => {{ const key = String(request.url || request); state.puts.push(key); storedResponses.set(key, response.clone()); }}, }}; const context = {{ URL, Request, Response, Headers, AbortController, setTimeout, clearTimeout, console, self: {{ location: {{ href: 'https://forge.example/dashboard/service-worker.js', origin: 'https://forge.example' }}, __STACKCHAIN_NAVIGATION_TIMEOUT_MS: 15, addEventListener: (name, handler) => {{ listeners[name] = handler; }}, skipWaiting: async () => {{ state.skipped = true; }}, clients: {{ claim: async () => {{ state.claimed = true; }}, matchAll: async () => state.clientList || [], openWindow: async url => {{ state.opened.push(url); }}, }}, registration: {{showNotification: async (title, options) => state.notifications.push({{title,options}})}}, __issueSync: {{ flush: async () => {{ state.backgroundFlushes += 1; return state.flushResult; }}, purge: async () => {{ state.outboxPurges += 1; }}, getReceiptPreference: async login => state.receiptLogin === login, }}, }}, importScripts: () => {{}}, caches: {{ open: async () => cache, keys: async () => ['stackchain-dashboard-old', 'another-app-cache'], delete: async key => {{ state.deleted.push(key); return true; }}, match: async request => cache.match(request), }}, fetch: async (request, options = {{}}) => {{ state.fetches.push(String(request.url || request)); if (state.failFetch) throw new Error('offline'); if (state.stallFetch) return new Promise((resolve, reject) => {{ options.signal?.addEventListener('abort', () => {{ state.fetchAborted = true; reject(new Error('aborted')); }}, {{ once: true }}); }}); if (state.lateFetch) return new Promise(resolve => {{ options.signal?.addEventListener('abort', () => {{ state.fetchAborted = true; }}, {{ once: true }}); setTimeout(() => resolve(new Response('late network')), 50); }}); const response = new Response('network', {{ status: state.fetchStatus }}); Object.defineProperty(response, 'redirected', {{ value: state.fetchRedirected }}); return response; }}, }}; vm.createContext(context); vm.runInContext(fs.readFileSync({json.dumps(str(WORKER))}, 'utf8') + '\\nself.__testFetchJson = fetchJson; self.__testWithSessionCsrf = withSessionCsrf;', context); async function dispatch(name, request) {{ let pending; let response; listeners[name]({{ request, waitUntil: promise => {{ pending = promise; }}, respondWith: promise => {{ response = promise; }}, }}); if (pending) await pending; return response ? await response : null; }} async function dispatchSync(tag) {{ let pending; listeners.sync({{ tag, waitUntil: promise => {{ pending = promise; }} }}); if (pending) await pending; }} async function dispatchMessage(data, ports = []) {{ let pending; listeners.message({{ data, ports, waitUntil: promise => {{ pending = promise; }} }}); if (pending) await pending; }} async function dispatchNotificationClick(route) {{ let pending; listeners.notificationclick({{ notification: {{data: {{route}}, close: () => {{ state.notificationClosed = true; }}}}, waitUntil: promise => {{ pending = promise; }}, }}); if (pending) await pending; }} (async () => {{ {scenario} }})().catch(error => {{ console.error(error); process.exit(1); }}); """ completed = subprocess.run( ["node", "-e", harness], capture_output=True, check=True, text=True ) return json.loads(completed.stdout) def test_resumable_today_session_ships_in_a_new_offline_shell(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/my-work.js'" in source assert "BASE + 'static/dashboard.js'" in source assert "BASE + 'static/dashboard.css'" in source def test_duplicate_aware_capture_ships_in_a_new_offline_shell(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/create-issue-sheet.js'" in source assert "BASE + 'static/dashboard.js'" in source def test_exact_later_picker_ships_atomically_in_a_new_offline_shell(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/later-picker.js'" in source def test_navigation_deadline_ships_in_a_new_shell_cache(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/dashboard.css'" in source assert "BASE + 'static/dashboard.js'" in source assert "BASE + 'static/install-app.js'" in source def test_today_convergence_ships_in_a_new_shell_cache(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/today-sync.js'" in source def test_mobile_search_viewport_ships_in_a_new_offline_shell(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/mobile-search-viewport.js'" in source def test_update_ownership_flow_ships_atomically_in_a_new_offline_shell(): source = WORKER.read_text() assert "stackchain-dashboard-shell-v66" in source assert "BASE + 'static/update-ownership.js'" in source def test_background_sync_event_flushes_closed_app_issue_outbox_only_for_its_tag(): result = run_worker_scenario( """ await dispatchSync('stackchain-issue-outbox-v1'); await dispatchSync('another-app-sync'); process.stdout.write(JSON.stringify(state)); """ ) assert result["backgroundFlushes"] == 1 def test_background_delivery_preserves_structured_uncertain_error(): result = run_worker_scenario( """ context.fetch=async()=>new Response(JSON.stringify({detail:{code:'delivery_uncertain',message:'Verify it was not posted before retrying.'}}),{status:422,headers:{'Content-Type':'application/json'}}); const outcome=await context.self.__testFetchJson('/dashboard/api/v1/repos/o/r/issues',{method:'POST'}) .then(()=>({ok:true}),error=>({status:error.status,code:error.code,message:error.message})); process.stdout.write(JSON.stringify(outcome)); """ ) assert result == { "status": 422, "code": "delivery_uncertain", "message": "Verify it was not posted before retrying.", } def test_revoked_background_session_purges_worker_data_and_notifies_dashboard_clients(): result = run_worker_scenario( """ state.clientMessages=[]; state.clientList=[{postMessage:message=>state.clientMessages.push(message)}]; context.fetch=async()=>new Response(JSON.stringify({detail:'Authentication required',code:'session_revoked'}),{status:401,headers:{'Content-Type':'application/json'}}); const outcome=await context.self.__testFetchJson('/dashboard/api/v1/repos/o/r/issues',{method:'POST'}) .then(()=>({ok:true}),error=>({status:error.status,code:error.code})); process.stdout.write(JSON.stringify({state,outcome})); """ ) assert result["outcome"] == {"status": 401, "code": "session_revoked"} assert result["state"]["outboxPurges"] == 1 assert result["state"]["deleted"] == ["stackchain-dashboard-old"] assert result["state"]["clientMessages"] == [ {"type": "stackchain-session-revoked"} ] def test_authenticated_page_message_resumes_queued_background_delivery(): result = run_worker_scenario( """ await dispatchMessage({type:'stackchain-resume-outbox'}); process.stdout.write(JSON.stringify(state)); """ ) assert result["backgroundFlushes"] == 1 def test_background_mutation_abort_also_cancels_stalled_csrf_lookup(): result = run_worker_scenario( """ let mutations=0; context.fetch=(request, options={})=>{ if(String(request).includes('/api/v1/session')){ return new Promise((resolve,reject)=>options.signal?.addEventListener('abort',()=>reject(new Error('aborted')), {once:true})); } mutations+=1; return Promise.resolve(new Response('{}',{status:200,headers:{'Content-Type':'application/json'}})); }; const controller=new AbortController(); const request=context.self.__testWithSessionCsrf(() => context.self.__testFetchJson('/dashboard/api/v1/repos/o/r/issues',{method:'POST',signal:controller.signal}) ).then(()=> 'completed', error=>error.message); controller.abort(); const outcome=await Promise.race([request,new Promise(resolve=>setTimeout(()=>resolve('blocked'),40))]); process.stdout.write(JSON.stringify({outcome,mutations})); process.exit(0); """ ) assert result == {"outcome": "aborted", "mutations": 0} def test_device_purge_message_stops_worker_outbox_and_acknowledges_completion(): result = run_worker_scenario( """ const replies = []; await dispatchMessage({type:'stackchain-purge-outbox'}, [{postMessage: value => replies.push(value)}]); process.stdout.write(JSON.stringify({state,replies})); """ ) assert result["state"]["outboxPurges"] == 1 assert result["replies"] == [{"ok": True}] def test_opted_in_background_sync_notifies_privately_and_receipt_tap_focuses_route(): result = run_worker_scenario( """ state.receiptLogin = 'timmy'; state.flushResult = {login:'timmy', receipts:[ {id:'capture-1',status:'confirmed',kind:'issue',route:'#/my-work/issue/stackchain/api/44'}, {id:'bad-1',status:'attention',kind:'message',route:'#/my-work/drafts'}, ]}; state.clientList = [{url:'https://forge.example/dashboard/', navigate:async function(url){ this.url=url; }, focus:async function(){ state.focused.push(this.url); }}]; await dispatchSync('stackchain-issue-outbox-v1'); await dispatchNotificationClick('#/my-work/issue/stackchain/api/44'); process.stdout.write(JSON.stringify(state)); """ ) assert result["notifications"] == [ { "title": "Queued issue created", "options": { "body": "Tap to open it in Stackchain.", "tag": "stackchain-delivery-capture-1", "data": {"route": "#/my-work/issue/stackchain/api/44"}, }, }, { "title": "Queued work needs attention", "options": { "body": "Tap to review it in Drafts.", "tag": "stackchain-delivery-bad-1", "data": {"route": "#/my-work/drafts"}, }, }, ] assert result["focused"] == [ "https://forge.example/dashboard/#/my-work/issue/stackchain/api/44" ] assert result["opened"] == [] assert result["notificationClosed"] is True def test_background_mutations_obtain_session_bound_csrf_proof(): source = WORKER.read_text() assert "async function sessionCsrf(signal)" in source assert "BASE + 'api/v1/session'" in source assert "signal," in source assert "headers.set('X-CSRF-Token', csrf)" in source def test_one_session_bound_csrf_proof_is_reused_for_a_background_drain(): source = WORKER.read_text() assert "async function withSessionCsrf(work)" in source assert "batchedCsrfController = new AbortController()" in source assert "batchedCsrf = sessionCsrf(batchedCsrfController.signal)" in source assert "batchedCsrfController.abort()" in source assert "await (batchedCsrf || sessionCsrf(options.signal))" in source assert "if (options.signal?.aborted)" in source assert "batch: withSessionCsrf" in source def test_install_precaches_complete_subpath_scoped_app_shell(): result = run_worker_scenario( """ await dispatch('install'); process.stdout.write(JSON.stringify(state)); """ ) assert result["skipped"] is True assert result["added"][0] == "/dashboard/" assert set(result["added"]) == { "/dashboard/", "/dashboard/manifest.webmanifest", "/dashboard/static/dashboard.css", "/dashboard/static/dashboard.js", "/dashboard/static/icons/stackchain-192.png", "/dashboard/static/icons/stackchain-512.png", "/dashboard/static/session.js", "/dashboard/static/markdown.js", "/dashboard/static/commands.js", "/dashboard/static/search-preview.js", "/dashboard/static/widgets.js", "/dashboard/static/drafts.js", "/dashboard/static/unfiled-captures.js", "/dashboard/static/outbox-coordinator.js", "/dashboard/static/issue-outbox.js", "/dashboard/static/authored-outbox.js", "/dashboard/static/notification-read-outbox.js", "/dashboard/static/offline-work.js", "/dashboard/static/offline-today.js", "/dashboard/static/my-work.js", "/dashboard/static/card-planning.js", "/dashboard/static/today-work.js", "/dashboard/static/plan-today.js", "/dashboard/static/plan-today-preview.js", "/dashboard/static/today-sync.js", "/dashboard/static/update-ownership.js", "/dashboard/static/later-work.js", "/dashboard/static/later-sync.js", "/dashboard/static/detail-defer.js", "/dashboard/static/later-picker.js", "/dashboard/static/pick-work.js", "/dashboard/static/conversation.js", "/dashboard/static/issue-sheet.js", "/dashboard/static/create-issue-sheet.js", "/dashboard/static/create-and-start.js", "/dashboard/static/assign-and-start.js", "/dashboard/static/pull-sheet.js", "/dashboard/static/review-sheet.js", "/dashboard/static/work-route.js", "/dashboard/static/task-overlay-history.js", "/dashboard/static/context-poller.js", "/dashboard/static/mobile-task-dock.js", "/dashboard/static/mobile-work-entry.js", "/dashboard/static/mobile-launch.js", "/dashboard/static/install-app.js", "/dashboard/static/mobile-search-viewport.js", "/dashboard/static/mobile-composer-viewport.js", "/dashboard/static/background-issue-sync.js", } def test_activate_deletes_only_stale_stackchain_caches(): result = run_worker_scenario( """ await dispatch('activate'); process.stdout.write(JSON.stringify(state)); """ ) assert result["claimed"] is True assert result["deleted"] == ["stackchain-dashboard-old"] def test_offline_navigation_returns_cached_shell_for_share_target_url(): result = run_worker_scenario( """ state.failFetch = true; state.cachedBody = 'cached dashboard'; const response = await dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/?title=Shared&url=https%3A%2F%2Fexample.com', }); process.stdout.write(JSON.stringify({ body: await response.text(), state })); """ ) assert result["body"] == "cached dashboard" assert result["state"]["fetches"] == [ "https://forge.example/dashboard/?title=Shared&url=https%3A%2F%2Fexample.com" ] def test_expired_offline_lease_purges_private_worker_data_and_refuses_cached_shell(): result = run_worker_scenario( """ await dispatchMessage({type:'stackchain-session-lease', expiresAt:1}); state.failFetch = true; state.cachedBody = 'private cached dashboard'; const response = await dispatch('fetch', { method:'GET', mode:'navigate', url:'https://forge.example/dashboard/', }); process.stdout.write(JSON.stringify({ status: response.status, body: await response.text(), state, })); """ ) assert result["status"] == 401 assert result["body"] == "Your Stackchain session expired. Reconnect and sign in." assert result["state"]["outboxPurges"] == 1 assert result["state"]["deleted"] == ["stackchain-dashboard-old"] assert "private cached dashboard" not in result["body"] def test_stalled_navigation_is_aborted_and_returns_cached_shell_within_deadline(): result = run_worker_scenario( """ state.stallFetch = true; state.cachedBody = 'cached dashboard'; const response = await Promise.race([ dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }), new Promise(resolve => setTimeout(() => resolve(null), 80)), ]); process.stdout.write(JSON.stringify({ timedOut: response === null, body: response ? await response.text() : null, state, })); """ ) assert result["timedOut"] is False assert result["body"] == "cached dashboard" assert result["state"]["fetchAborted"] is True assert result["state"]["puts"] == [] def test_stalled_navigation_without_cached_shell_returns_deterministic_504(): result = run_worker_scenario( """ state.stallFetch = true; const response = await Promise.race([ dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }), new Promise(resolve => setTimeout(() => resolve(null), 80)), ]); process.stdout.write(JSON.stringify({ status: response?.status || null, body: response ? await response.text() : null, contentType: response?.headers.get('Content-Type') || null, state, })); """ ) assert result["status"] == 504 assert result["body"] == "Stackchain is offline and the dashboard is not cached yet. Reconnect and try again." assert result["contentType"] == "text/plain; charset=utf-8" assert result["state"]["fetchAborted"] is True def test_navigation_deadline_wins_when_fetch_ignores_abort_and_prevents_late_cache_write(): result = run_worker_scenario( """ state.lateFetch = true; state.cachedBody = 'cached dashboard'; const response = await Promise.race([ dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }), new Promise(resolve => setTimeout(() => resolve(null), 35)), ]); const body = response ? await response.text() : null; await new Promise(resolve => setTimeout(resolve, 60)); process.stdout.write(JSON.stringify({timedOut: response === null, body, state})); """ ) assert result["timedOut"] is False assert result["body"] == "cached dashboard" assert result["state"]["fetchAborted"] is True assert result["state"]["puts"] == [] def test_redirected_login_navigation_does_not_replace_cached_dashboard_shell(): result = run_worker_scenario( """ state.fetchRedirected = true; const response = await dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/?title=Shared', }); process.stdout.write(JSON.stringify({ status: response.status, state })); """ ) assert result["status"] == 200 assert result["state"]["puts"] == [] def test_cross_origin_navigation_is_not_intercepted_or_cached(): result = run_worker_scenario( """ const response = await dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://outside.example/page', }); process.stdout.write(JSON.stringify({ intercepted: response !== null, state })); """ ) assert result["intercepted"] is False assert result["state"]["fetches"] == [] assert result["state"]["puts"] == [] @pytest.mark.parametrize("status", [500, 502, 503, 504]) def test_server_outage_navigation_returns_working_cached_shell_without_replacing_it(status): result = run_worker_scenario( f""" state.fetchStatus = {status}; state.cachedBody = 'cached dashboard'; const response = await dispatch('fetch', {{ method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }}); process.stdout.write(JSON.stringify({{ status: response.status, body: await response.text(), state }})); """ ) assert result["status"] == 200 assert result["body"] == "cached dashboard" assert result["state"]["puts"] == [] def test_server_outage_without_cached_shell_preserves_network_error(): result = run_worker_scenario( """ state.fetchStatus = 503; const response = await dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }); process.stdout.write(JSON.stringify({ status: response.status, body: await response.text(), state })); """ ) assert result["status"] == 503 assert result["body"] == "network" assert result["state"]["puts"] == [] def test_client_error_navigation_is_not_hidden_by_cached_shell(): result = run_worker_scenario( """ state.fetchStatus = 401; state.cachedBody = 'cached dashboard'; const response = await dispatch('fetch', { method: 'GET', mode: 'navigate', url: 'https://forge.example/dashboard/', }); process.stdout.write(JSON.stringify({ status: response.status, body: await response.text(), state })); """ ) assert result["status"] == 401 assert result["body"] == "network" assert result["state"]["puts"] == []