""" Step 16 — Single Deployment + Startup DB Migration Tests Green paths: - test_fresh_install_initializes: Fresh DB gets the schema, schema_version stamped, defaults present - test_data_persists_across_reload: Data survives a reload (stable OPFS location) - test_opfs_directory_unversioned: OPFS pool directory carries no version suffix - test_full_migration_flow: Pending migration -> forced backup download -> Update -> app loads with data intact (uses the v1 no-op migration by rewinding schema_version to 0) Red paths: - test_db_newer_than_app: dbVersion > SCHEMA_VERSION shows the blocking DB_NEWER_THAN_APP error overlay without touching the data - test_migration_screen_no_bypass: Update stays disabled before a backup download, and a reload lands back on the migration screen with data untouched Deferred: - Migration-failure rollback test: cannot be exercised realistically until a real v2 migration script exists (faking it would require test-only scripts in production code). """ import re from playwright.sync_api import Page, expect def _wait_for_db(page: Page, app_url: str): """Navigate and wait for the database to be ready.""" page.goto(app_url) page.wait_for_function( "document.documentElement.getAttribute('data-db-ready') === 'true'", timeout=15000 ) def _set_db_version(page: Page, version: int): """Rewrite the schema_version table so it contains only `version`.""" page.evaluate(f""" (async () => {{ await window.__db.run("DELETE FROM schema_version"); await window.__db.run("INSERT INTO schema_version (version) VALUES ({version})"); }})() """) def _create_exercise(page: Page, title: str) -> str: return page.evaluate(f""" window.__repos.exercises.create({{ title: '{title}', description: '', created_by: 'test' }}) """) def _download_via_button(page: Page, trigger_selector: str): """Click a button that opens the save-filename dialog, accept the default name, return the Download.""" page.locator(trigger_selector).click() page.wait_for_selector('[data-testid="save-file-confirm"]') with page.expect_download() as download_info: page.locator('[data-testid="save-file-confirm"]').click() return download_info.value def test_fresh_install_initializes(page: Page, app_url: str): """A fresh install creates the schema, stamps SCHEMA_VERSION, and seeds defaults.""" _wait_for_db(page, app_url) versions = page.evaluate( 'window.__db.selectAll("SELECT version FROM schema_version ORDER BY version")' ) assert len(versions) == 1, f"Expected a single schema_version row, got {versions}" assert versions[0]["version"] == 1 uuid = page.evaluate("async () => await window.__repos.settings.get('user_uuid')") assert uuid is not None, "Default settings missing after fresh install" def test_data_persists_across_reload(page: Page, app_url: str): """Data persists across reload now that the OPFS location is version-independent.""" _wait_for_db(page, app_url) ex_id = _create_exercise(page, "Step16 Persistent Exercise") page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-ready') === 'true'", timeout=15000 ) exercise = page.evaluate(f"window.__repos.exercises.getById('{ex_id}')") assert exercise is not None, "Exercise lost after reload" assert exercise["title"] == "Step16 Persistent Exercise" def test_opfs_directory_unversioned(page: Page, app_url: str): """The OPFS pool directory is named .trainus with no app-version suffix.""" _wait_for_db(page, app_url) dir_names = page.evaluate(""" (async () => { const root = await navigator.storage.getDirectory(); const names = []; for await (const name of root.keys()) names.push(name); return names; })() """) assert ".trainus" in dir_names, f"Expected '.trainus' OPFS directory, found: {dir_names}" versioned = [n for n in dir_names if re.match(r"^\.trainus-", n)] assert versioned == [], f"Found versioned OPFS directories: {versioned}" def test_full_migration_flow(page: Page, app_url: str): """Pending migration forces a backup download before Update, then migrates and keeps data.""" _wait_for_db(page, app_url) ex_id = _create_exercise(page, "Step16 Migration Exercise") _set_db_version(page, 0) page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-migration') === 'pending'", timeout=15000 ) expect(page.locator('[data-testid="db-migration"]')).to_be_visible() # Update is locked until a backup has been downloaded expect(page.locator('[data-testid="migration-update"]')).to_be_disabled() download = _download_via_button(page, '[data-testid="migration-backup"]') filename = download.suggested_filename assert re.match( r"^trainus-\d+\.\d+\.\d+-backup-\d{8}-\d{2}h\d{2}m\d{2}s\.sqlite3$", filename ), f"Unexpected backup filename: {filename}" expect(page.locator('[data-testid="migration-update"]')).to_be_enabled() page.locator('[data-testid="migration-update"]').click() page.wait_for_function( "document.documentElement.getAttribute('data-db-ready') === 'true'", timeout=15000 ) max_version = page.evaluate( 'window.__db.selectOne("SELECT MAX(version) AS v FROM schema_version")' ) assert max_version["v"] == 1, f"schema_version not migrated: {max_version}" exercise = page.evaluate(f"window.__repos.exercises.getById('{ex_id}')") assert exercise is not None, "Data lost during migration" last_backup = page.evaluate("async () => await window.__repos.settings.get('last_backup_at')") assert last_backup is not None, "last_backup_at was not set by the forced backup" def test_db_newer_than_app(page: Page, app_url: str): """A DB created by a newer app version blocks with DB_NEWER_THAN_APP and stays untouched.""" _wait_for_db(page, app_url) ex_id = _create_exercise(page, "Step16 Newer DB Exercise") page.evaluate('window.__db.run("INSERT INTO schema_version (version) VALUES (2)")') page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-error') !== null", timeout=15000 ) error = page.evaluate("document.documentElement.getAttribute('data-db-error')") assert error == "DB_NEWER_THAN_APP", f"Unexpected error code: {error}" expect(page.locator('[data-testid="db-error"]')).to_be_visible() # Data untouched: removing the fake newer version makes the app load again page.evaluate('window.__db.run("DELETE FROM schema_version WHERE version = 2")') page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-ready') === 'true'", timeout=15000 ) exercise = page.evaluate(f"window.__repos.exercises.getById('{ex_id}')") assert exercise is not None, "Data lost while DB was flagged as newer than the app" def test_migration_screen_no_bypass(page: Page, app_url: str): """The migration screen cannot be bypassed: Update stays disabled and survives reloads.""" _wait_for_db(page, app_url) ex_id = _create_exercise(page, "Step16 NoBypass Exercise") _set_db_version(page, 0) page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-migration') === 'pending'", timeout=15000 ) expect(page.locator('[data-testid="migration-update"]')).to_be_disabled() # Reloading without downloading a backup lands back on the migration screen page.reload() page.wait_for_function( "document.documentElement.getAttribute('data-db-migration') === 'pending'", timeout=15000 ) expect(page.locator('[data-testid="db-migration"]')).to_be_visible() expect(page.locator('[data-testid="migration-update"]')).to_be_disabled() # No data was touched while paused version = page.evaluate( 'window.__db.selectOne("SELECT MAX(version) AS v FROM schema_version")' ) assert version["v"] == 0, f"schema_version changed without user action: {version}" exercise = page.evaluate(f"window.__repos.exercises.getById('{ex_id}')") assert exercise is not None, "Data touched while migration was pending"