"""FlowDeck — PWA offline client module tests (v6.0.0). Covers static/js/offline.js (IndexedDB + sync queue), its inclusion in the templates and the editor's offline save hook. The server-side endpoints are exercised in tests/test_sync.py. """ import re from pathlib import Path ROOT = Path(__file__).resolve().parent.parent OFFLINE_JS = ROOT / "static" / "js" / "offline.js" def _token(user_id, login): from app.auth.session import SessionManager return SessionManager.create_session({"id": user_id, "login": login, "full_name": login.title(), "is_admin": 1}) def _create_user_ws(client, login="offlinetest", ws_name="Offline WS"): from app.db import get_conn with get_conn() as conn: cur = conn.execute( "INSERT INTO users (login, full_name, email) VALUES (?, ?, ?)", (login, login.title(), f"{login}@test.com"), ) user_id = cur.lastrowid conn.execute( "INSERT INTO workspaces (name, owner_id) VALUES (?, ?)", (ws_name, user_id), ) conn.commit() return _token(user_id, login) # ── static asset ─────────────────────────────────────────────────────────── def test_offline_js_served(client): resp = client.get("/static/js/offline.js") assert resp.status_code == 200 assert "javascript" in resp.headers.get("content-type", "") def test_offline_js_exposes_public_api(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "window.FlowOffline" in src for api in ("enqueue", "flush", "delta", "status", "savePageOffline", "getQueue", "pendingCount", "onChange"): assert api in src, f"missing FlowOffline API: {api}" def test_offline_js_indexeddb_schema(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "flowdeck-offline" in src for store in ("pages_offline", "collections_offline", "sync_queue", "sync_meta"): assert store in src, f"missing IndexedDB store: {store}" def test_offline_js_targets_sync_endpoints(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "/api/v2/sync/batch" in src assert "/api/v2/sync/delta" in src assert "/api/v2/sync/status" in src # mutation shape expected by the server engine for field in ("mutation_id", "client_timestamp", "device_id"): assert field in src, f"missing mutation field: {field}" def test_offline_js_background_sync(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "sync-flowdeck" in src assert "registerBackgroundSync" in src # ── template wiring ──────────────────────────────────────────────────────── def test_base_includes_offline_module(client): base = (ROOT / "app" / "templates" / "base.html").read_text(encoding="utf-8") assert "/static/js/offline.js" in base assert "/sw.js" in base # served on a base.html page as well session = _create_user_ws(client, login="basemod") resp = client.get("/local-workspace", cookies={"flowdeck_session": session}) assert resp.status_code == 200 assert "/static/js/offline.js" in resp.text def test_workspace_offline_banner_shows_pending(client): session = _create_user_ws(client) resp = client.get("/local-workspace", cookies={"flowdeck_session": session}) assert resp.status_code == 200 body = resp.text assert "offline-banner" in body assert "pendingSyncCount" in body assert "FlowOffline" in body def test_editor_save_has_offline_hook(): src = (ROOT / "static" / "js" / "page_editor_scripts.js").read_text(encoding="utf-8") assert "FlowOffline" in src assert "savePageOffline" in src assert "navigator.onLine" in src def test_sw_precache_includes_offline_js(): sw = (ROOT / "static" / "sw.js").read_text(encoding="utf-8") urls = re.findall(r"'(/static/js/offline\.js[^']*)'", sw) assert urls, "offline.js missing from SW precache" for url in urls: assert (ROOT / url.split("?")[0].lstrip("/")).is_file() # ── phase 6: hardening + sync UI ─────────────────────────────────────────── def test_offline_js_batch_limits_and_timeout(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "MAX_BATCH" in src assert "FETCH_TIMEOUT_MS" in src assert "AbortController" in src assert "fetchWithTimeout" in src assert "purgeOld" in src assert "QUEUE_RETENTION_MS" in src def test_offline_js_dirty_markers_and_sync_event(): src = OFFLINE_JS.read_text(encoding="utf-8") assert "getDirtyPageIds" in src assert "markDirtySidebar" in src assert "fd-dirty" in src assert "flowdeck:sync-done" in src def test_base_has_sync_badge_and_toast_wiring(client): base = (ROOT / "app" / "templates" / "base.html").read_text(encoding="utf-8") assert "fd-sync-badge" in base assert "fd-sync-spinner" in base assert "fd-dirty" in base assert "flowdeck:sync-done" in base # rendered on an authenticated base.html page session = _create_user_ws(client, login="syncbadge") resp = client.get("/local-workspace", cookies={"flowdeck_session": session}) assert resp.status_code == 200 assert "fd-sync-badge" in resp.text # ── phase 7: E2E spec (runs against a live server via Playwright) ─────────── def test_e2e_pwa_offline_spec_present(): spec = ROOT / "e2e" / "pwa_offline.spec.js" assert spec.is_file() src = spec.read_text(encoding="utf-8") assert "FlowOffline" in src assert "/api/v2/sync/delta" in src assert "setOffline(true)" in src assert "setOffline(false)" in src assert "navigator.serviceWorker" in src assert "manifest" in src # skips gracefully when the server is not running assert "test.skip" in src # ── phase 8: release docs + version ──────────────────────────────────────── def test_version_at_least_6_0_0(): """VERSION and app/main.py agree and are >= 6.0.0 (PWA release baseline).""" version = (ROOT / "VERSION").read_text(encoding="utf-8").strip() assert tuple(int(x) for x in version.split(".")) >= (6, 0, 0) assert f'version="{version}"' in (ROOT / "app" / "main.py").read_text(encoding="utf-8") def test_help_documents_offline_mode(client): session = _create_user_ws(client, login="helpuser") resp = client.get("/help", cookies={"flowdeck_session": session}) assert resp.status_code == 200 body = resp.text assert "Offline mode (PWA)" in body assert "IndexedDB" in body def test_changelog_and_roadmap_mention_v6(client): changelog = (ROOT / "CHANGELOG.md").read_text(encoding="utf-8") roadmap = (ROOT / "ROADMAP.md").read_text(encoding="utf-8") assert "## v6.0.0" in changelog assert "PWA" in changelog assert "v6.0.0 — PWA" in roadmap