diff --git a/backend/app/main.py b/backend/app/main.py index 71fcf5d..438e0cd 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -45,6 +45,7 @@ def _migrate_add_missing_columns(sync_conn): import sqlalchemy as sa # Column definitions: (table, name, type, default) + # default=None → nullable column (no DEFAULT clause) pending = [ ("station_config", "stream_url", sa.String(500), ""), ("station_config", "stream_metadata_url", sa.String(500), ""), @@ -75,6 +76,8 @@ def _migrate_add_missing_columns(sync_conn): ("station_config", "color_light", sa.String(7), "#f0ebe3"), ("station_config", "color_medium", sa.String(7), "#8a8580"), ("station_config", "color_black", sa.String(7), "#1a1816"), + # Multi-admin support: password_hash (nullable — existing rows stay NULL) + ("users", "password_hash", sa.String(255), None), ] # Determine dialect @@ -83,6 +86,10 @@ def _migrate_add_missing_columns(sync_conn): inspector = sa.inspect(sync_conn) for table_name, col_name, col_type, default in pending: + # Skip tables that don't exist yet (e.g., station_config on fresh DB) + if not inspector.has_table(table_name): + continue + # Check if column already exists existing = inspector.get_columns(table_name) if any(c["name"] == col_name for c in existing): @@ -99,13 +106,23 @@ def _migrate_add_missing_columns(sync_conn): sql_type = "TEXT" if dialect == "sqlite": - sync_conn.execute( - sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" TEXT NOT NULL DEFAULT '{default}'") - ) + if default is None: + sync_conn.execute( + sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" {sql_type}") + ) + else: + sync_conn.execute( + sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" {sql_type} NOT NULL DEFAULT '{default}'") + ) else: - sync_conn.execute( - sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" {sql_type} NOT NULL DEFAULT '{default}'") - ) + if default is None: + sync_conn.execute( + sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" {sql_type}") + ) + else: + sync_conn.execute( + sa.text(f"ALTER TABLE {table_name} ADD COLUMN \"{col_name}\" {sql_type} NOT NULL DEFAULT '{default}'") + ) print(f" ✓ Added {table_name}.{col_name}") diff --git a/backend/migrate_roles.py b/backend/migrate_roles.py index 9010bfb..8a9f2b7 100644 --- a/backend/migrate_roles.py +++ b/backend/migrate_roles.py @@ -1,5 +1,6 @@ """ -Migration seed: create the 'admin' role and assign it to existing is_admin=True users. +Migration seed: add missing schema columns/tables, then create the 'admin' role +and assign it to existing is_admin=True users. Run once after deploying the updated schema. Idempotent — safe to run multiple times. @@ -15,9 +16,84 @@ from app.database import async_session from app.user_models import User, Role, user_roles +async def ensure_schema(session) -> None: + """Add missing columns/tables so the User model matches the live database. + + - Adds ``password_hash`` to ``users`` (nullable — existing rows stay NULL). + - Creates ``roles`` and ``user_roles`` tables if they do not exist. + + All steps use ``IF NOT EXISTS`` / ``NOT EXISTS`` guards so the function is + fully idempotent. + """ + conn = await session.connection() + + # 1) Add password_hash column if missing + result = await conn.execute( + text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = 'users' AND column_name = 'password_hash'" + ) + ) + if not result.scalar(): + await conn.execute( + text("ALTER TABLE users ADD COLUMN password_hash VARCHAR(255)") + ) + print(" ✓ Added 'password_hash' column to 'users'") + else: + print(" ✓ 'users.password_hash' already exists") + + # 2) Create 'roles' table if missing + result = await conn.execute( + text( + "SELECT 1 FROM information_schema.tables " + "WHERE table_name = 'roles'" + ) + ) + if not result.scalar(): + await conn.execute( + text( + "CREATE TABLE IF NOT EXISTS roles (" + "id SERIAL PRIMARY KEY, " + "name VARCHAR(50) UNIQUE NOT NULL, " + "description TEXT" + ")" + ) + ) + print(" ✓ Created 'roles' table") + else: + print(" ✓ 'roles' table already exists") + + # 3) Create 'user_roles' table if missing + result = await conn.execute( + text( + "SELECT 1 FROM information_schema.tables " + "WHERE table_name = 'user_roles'" + ) + ) + if not result.scalar(): + await conn.execute( + text( + "CREATE TABLE IF NOT EXISTS user_roles (" + "user_id INTEGER REFERENCES users(id) ON DELETE CASCADE, " + "role_id INTEGER REFERENCES roles(id) ON DELETE CASCADE, " + "PRIMARY KEY (user_id, role_id)" + ")" + ) + ) + print(" ✓ Created 'user_roles' table") + else: + print(" ✓ 'user_roles' table already exists") + + await session.commit() + + async def migrate() -> None: - """Ensure 'admin' role exists and assign to existing is_admin=True users.""" + """Ensure schema is up to date, 'admin' role exists, and legacy admins get the role.""" async with async_session() as session: + print("Ensuring schema is up to date...") + await ensure_schema(session) + + print("Running role migration...") # 1) Create 'admin' role if it doesn't exist result = await session.execute(select(Role).where(Role.name == "admin")) admin_role = result.scalar_one_or_none() diff --git a/backend/tests/test_migration.py b/backend/tests/test_migration.py new file mode 100644 index 0000000..b51099c --- /dev/null +++ b/backend/tests/test_migration.py @@ -0,0 +1,112 @@ +"""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