200 lines
8.2 KiB
Python
200 lines
8.2 KiB
Python
"""
|
|
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"
|