stackchain-dashboard/tests/test_service_worker.py

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: [], failFetch: false, fetchStatus: 200, cachedBody: null }};
const cache = {{
  addAll: async urls => {{ state.added = urls; }},
  match: async request => state.cachedBody === null ? null : new Response(state.cachedBody),
  put: async (request, response) => {{ state.puts.push(String(request.url || request)); }},
}};
const context = {{
  URL, Request, Response,
  console,
  self: {{
    location: {{ href: 'https://forge.example/dashboard/service-worker.js', origin: 'https://forge.example' }},
    addEventListener: (name, handler) => {{ listeners[name] = handler; }},
    skipWaiting: async () => {{ state.skipped = true; }},
    clients: {{ claim: async () => {{ state.claimed = true; }} }},
  }},
  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 => {{
    state.fetches.push(String(request.url || request));
    if (state.failFetch) throw new Error('offline');
    return new Response('network', {{ status: state.fetchStatus }});
  }},
}};
vm.createContext(context);
vm.runInContext(fs.readFileSync({json.dumps(str(WORKER))}, 'utf8'), 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 () => {{
{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_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/icons/stackchain-192.png",
        "/dashboard/static/icons/stackchain-512.png",
        "/dashboard/static/markdown.js",
        "/dashboard/static/commands.js",
        "/dashboard/static/search-preview.js",
        "/dashboard/static/widgets.js",
        "/dashboard/static/drafts.js",
        "/dashboard/static/offline-work.js",
        "/dashboard/static/my-work.js",
        "/dashboard/static/pick-work.js",
        "/dashboard/static/conversation.js",
        "/dashboard/static/issue-sheet.js",
        "/dashboard/static/create-issue-sheet.js",
        "/dashboard/static/pull-sheet.js",
        "/dashboard/static/review-sheet.js",
        "/dashboard/static/work-route.js",
        "/dashboard/static/context-poller.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_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"] == []