robot
85b3a4adc4
CI Pipeline / backend-test (push) Successful in 4m5s
CI Pipeline / frontend-test (push) Successful in 27s
CI Pipeline / frontend-build (push) Successful in 30s
CI Pipeline / backend-lint (push) Successful in 11s
CI Pipeline / backend-test (pull_request) Successful in 4m11s
CI Pipeline / frontend-test (pull_request) Successful in 26s
CI Pipeline / frontend-build (pull_request) Successful in 30s
CI Pipeline / backend-lint (pull_request) Successful in 8s
CI Pipeline / quality-gate (push) Successful in 2s
CI Pipeline / quality-gate (pull_request) Successful in 2s
- Add users.password_hash to _migrate_add_missing_columns pending list - Skip tables that don't exist yet (prevents crash on fresh DB) - Handle nullable columns (no DEFAULT clause) in ALTER TABLE - Fix SQLite column type (use sql_type instead of hardcoded TEXT) - Update migrate_roles.py to create schema columns/tables before seeding roles - Add unit tests for migration: adds column, nullable, idempotent, no-op
113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""Tests for startup DB migration logic.
|
|
|
|
Verifies that _migrate_add_missing_columns correctly adds missing columns
|
|
(password_hash, etc.) to existing tables without dropping/recreating them.
|
|
|
|
This test simulates a production database that was deployed before the
|
|
multi-admin PR — i.e., a ``users`` table that does NOT have a ``password_hash``
|
|
column — and confirms the migration adds it idempotently.
|
|
"""
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
|
|
from app.main import _migrate_add_missing_columns
|
|
|
|
|
|
class TestMigrateAddMissingColumns:
|
|
"""Test the startup migration that patches missing columns."""
|
|
|
|
@pytest.fixture
|
|
def legacy_engine(self):
|
|
"""Create a sync SQLite engine with a legacy ``users`` table (no password_hash)."""
|
|
engine = sa.create_engine("sqlite:///:memory:")
|
|
|
|
# Simulate a production DB that existed before the multi-admin PR:
|
|
# users table with the OLD schema (no password_hash column)
|
|
with engine.begin() as conn:
|
|
conn.execute(sa.text(
|
|
"CREATE TABLE users ("
|
|
"id INTEGER PRIMARY KEY, "
|
|
"email VARCHAR(255) UNIQUE NOT NULL, "
|
|
"display_name VARCHAR(100) NOT NULL, "
|
|
"avatar_url VARCHAR(500), "
|
|
"auth_provider VARCHAR(20) NOT NULL, "
|
|
"is_admin BOOLEAN, "
|
|
"created_at TIMESTAMP NOT NULL"
|
|
")"
|
|
))
|
|
# Insert a user with existing data
|
|
conn.execute(sa.text(
|
|
"INSERT INTO users (email, display_name, auth_provider, is_admin, created_at) "
|
|
"VALUES ('admin@local', 'Admin', 'google', 1, '2025-01-01 00:00:00')"
|
|
))
|
|
|
|
return engine
|
|
|
|
def test_password_hash_added_to_existing_users_table(self, legacy_engine):
|
|
"""The migration adds password_hash to a users table that lacks it."""
|
|
with legacy_engine.begin() as conn:
|
|
# Verify column does NOT exist before migration
|
|
info = sa.inspect(conn)
|
|
cols = [c["name"] for c in info.get_columns("users")]
|
|
assert "password_hash" not in cols, (
|
|
"Test setup failed: password_hash should not exist before migration"
|
|
)
|
|
|
|
# Run the migration
|
|
_migrate_add_missing_columns(conn)
|
|
|
|
# Verify column now exists
|
|
info = sa.inspect(conn)
|
|
cols = [c["name"] for c in info.get_columns("users")]
|
|
assert "password_hash" in cols
|
|
|
|
def test_password_hash_is_nullable(self, legacy_engine):
|
|
"""The migration creates password_hash as nullable — existing rows stay intact."""
|
|
with legacy_engine.begin() as conn:
|
|
_migrate_add_missing_columns(conn)
|
|
|
|
# Existing user should still be queryable, password_hash = NULL
|
|
result = conn.execute(sa.text(
|
|
"SELECT email, password_hash FROM users WHERE email = 'admin@local'"
|
|
))
|
|
row = result.fetchone()
|
|
assert row is not None
|
|
assert row[0] == "admin@local"
|
|
# password_hash should be NULL for the pre-existing user
|
|
assert row[1] is None
|
|
|
|
def test_migration_is_idempotent(self, legacy_engine):
|
|
"""Running the migration twice does not error — second run is a no-op."""
|
|
with legacy_engine.begin() as conn:
|
|
# First run — adds the column
|
|
_migrate_add_missing_columns(conn)
|
|
|
|
# Second run — should not raise
|
|
_migrate_add_missing_columns(conn)
|
|
|
|
# Column still exists
|
|
info = sa.inspect(conn)
|
|
cols = [c["name"] for c in info.get_columns("users")]
|
|
assert "password_hash" in cols
|
|
|
|
def test_migration_noop_when_column_already_exists(self, legacy_engine):
|
|
"""If password_hash already exists, the migration skips it cleanly."""
|
|
with legacy_engine.begin() as conn:
|
|
# Pre-create the column manually
|
|
conn.execute(sa.text(
|
|
"ALTER TABLE users ADD COLUMN password_hash VARCHAR(255)"
|
|
))
|
|
|
|
info = sa.inspect(conn)
|
|
cols = [c["name"] for c in info.get_columns("users")]
|
|
assert "password_hash" in cols
|
|
|
|
# Migration should not error and should not add a duplicate
|
|
_migrate_add_missing_columns(conn)
|
|
|
|
info = sa.inspect(conn)
|
|
cols = [c["name"] for c in info.get_columns("users")]
|
|
# password_hash appears exactly once
|
|
assert cols.count("password_hash") == 1
|