stackchain-dashboard/tests/test_dashboard_session_frontend.py

import json
import subprocess
from pathlib import Path

import pytest

from tests.dashboard_bundle import dashboard


ROOT = Path(__file__).resolve().parents[1]
SESSION_JS = ROOT / "frontend" / "session.js"


def run_session_scenario(scenario: str) -> dict:
    harness = f"""
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
const state = {{ requests: [], removed: [], deletedDatabases: [], deletionCompleted: false, deletedCaches: [], assigned: '', replaced: [], assignedAfterDeletion: false, workerMessages: [], confirmations: [], clearErrors: [], responseStatus: 200 }};
const storage = {{
  values: new Map([['stackchain.private', 'secret'], ['gitea.preference', 'keep']]),
  get length() {{ return this.values.size; }},
  key(index) {{ return Array.from(this.values.keys())[index] || null; }},
  removeItem(key) {{ state.removed.push(key); this.values.delete(key); }},
}};
const boundary = createSessionBoundary({{
  cookie: () => 'other=x; stackchain_csrf=csrf-proof; theme=dark',
  origin: 'https://forge.example',
  base: '/dashboard/',
  fetchImpl: async (url, options = {{}}) => {{
    state.requests.push({{ url: String(url), method: options.method || 'GET', headers: Object.fromEntries(new Headers(options.headers || {{}})) }});
    return new Response('{{}}', {{ status: state.responseStatus, headers: {{ 'Content-Type': 'application/json' }} }});
  }},
  localStorage: storage,
  sessionStorage: storage,
  indexedDB: {{ deleteDatabase: name => {{
    state.deletedDatabases.push(name);
    const request = {{ onsuccess: null, onerror: null, onblocked: null, error: null }};
    setTimeout(() => {{
      if (state.failDeletion) {{ request.error = new Error('database blocked'); request.onerror?.(); }}
      else {{ state.deletionCompleted = true; request.onsuccess?.(); }}
    }}, 20);
    return request;
  }} }},
  caches: {{ keys: async () => ['stackchain-dashboard-shell-v15', 'gitea-assets'], delete: async key => {{ state.deletedCaches.push(key); }} }},
  serviceWorker: {{ ready: Promise.resolve({{ active: {{ postMessage: (message, ports = []) => {{ state.workerMessages.push(message); ports[0]?.postMessage({{ok:true}}); }} }} }}) }},
  MessageChannel: class {{ constructor() {{
    const first = {{ onmessage: null, postMessage: data => queueMicrotask(() => second.onmessage?.({{data}})) }};
    const second = {{ onmessage: null, postMessage: data => queueMicrotask(() => first.onmessage?.({{data}})) }};
    this.port1 = first; this.port2 = second;
  }} }},
  location: {{
    assign: value => {{ state.assigned = value; state.assignedAfterDeletion = state.deletionCompleted; }},
    replace: value => state.replaced.push(value),
  }},
  onClearError: error => state.clearErrors.push(error.message),
  confirmAction: message => {{ state.confirmations.push(message); return true; }},
}});
(async () => {{ {scenario} }})().catch(error => {{ console.error(error); process.exit(1); }});
"""
    result = subprocess.run(["node", "-e", harness], text=True, capture_output=True, check=True)
    return json.loads(result.stdout)


def test_mutating_same_origin_fetch_receives_csrf_proof():
    result = run_session_scenario(
        """
await boundary.fetch('/dashboard/api/v1/notifications/7/read', { method: 'PATCH' });
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["requests"][0]["headers"]["x-csrf-token"] == "csrf-proof"


def test_same_origin_unauthorized_response_replaces_dashboard_once_without_clearing_private_work():
    result = run_session_scenario(
        """
state.responseStatus = 401;
await Promise.all([
  boundary.fetch('/dashboard/api/v1/live'),
  boundary.fetch('https://forge.example/dashboard/api/v1/context'),
]);
state.remaining = Array.from(storage.values.keys());
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["replaced"] == [
        "/dashboard/login?reason=session-expired"
    ]
    assert result["remaining"] == ["stackchain.private", "gitea.preference"]
    assert result["deletedDatabases"] == []
    assert result["deletedCaches"] == []


def test_cross_origin_unauthorized_response_does_not_expire_dashboard_session():
    result = run_session_scenario(
        """
state.responseStatus = 401;
await boundary.fetch('https://untrusted.example/api/private');
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["replaced"] == []


def test_sign_out_clears_only_dashboard_private_device_state():
    result = run_session_scenario(
        """
await boundary.signOut();
state.remaining = Array.from(storage.values.keys());
process.stdout.write(JSON.stringify(state));
"""
    )

    request = result["requests"][0]
    assert request["url"] == "/dashboard/api/v1/session"
    assert request["method"] == "DELETE"
    assert request["headers"]["x-csrf-token"] == "csrf-proof"
    assert "stackchain.private" in result["removed"]
    assert result["remaining"] == ["gitea.preference"]
    assert result["deletedDatabases"] == ["stackchain-background-outbox-v1"]
    assert result["deletedCaches"] == ["stackchain-dashboard-shell-v15"]
    assert result["assigned"] == "/dashboard/login"
    assert result["assignedAfterDeletion"] is True


def test_sign_out_stays_on_page_and_reports_failed_private_outbox_deletion():
    result = run_session_scenario(
        """
state.failDeletion = true;
try { await boundary.signOut(); } catch (error) { state.signOutError = error.message; }
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["assigned"] == ""
    assert result["clearErrors"] == ["Could not clear private queued work from this device."]
    assert result["signOutError"] == "Could not clear private queued work from this device."


def test_sign_out_all_devices_requires_confirmation_and_uses_global_endpoint():
    result = run_session_scenario(
        """
await boundary.signOutAllDevices();
state.remaining = Array.from(storage.values.keys());
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["confirmations"] == [
        "Sign out every device? You will need to sign in again everywhere."
    ]
    request = result["requests"][0]
    assert request["url"] == "/dashboard/api/v1/sessions"
    assert request["method"] == "DELETE"
    assert request["headers"]["x-csrf-token"] == "csrf-proof"
    assert result["remaining"] == ["gitea.preference"]
    assert result["assigned"] == "/dashboard/login"


def test_authenticated_dashboard_load_requests_queued_delivery_resume():
    result = run_session_scenario(
        """
await boundary.resumeQueuedWork();
process.stdout.write(JSON.stringify(state));
"""
    )

    assert result["workerMessages"] == [{"type": "stackchain-resume-outbox"}]


@pytest.mark.anyio
async def test_dashboard_loads_session_boundary_first_and_offers_sign_out():
    html = await dashboard()

    assert '<script src="static/session.js"></script>' in html
    assert html.index('static/session.js') < html.index('static/markdown.js')
    assert '<button id="sign-out" type="button">Sign out &amp; clear this device</button>' in html
    assert '<button id="sign-out-all" type="button">Sign out all devices</button>' in html
    assert '#sign-out-all { min-height:44px; }' in html