Simon/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-06-28 14:56:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

run fmt

Browse Source
This commit is contained in:
Georges-Antoine Assi
2025-12-01 20:02:34 -05:00
parent 64a37e1f10
commit 874a8ec85d
2 changed files with 64 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
backend/alembic/versions/0057_multi_notes.py
View File

@@ -9,6 +9,7 @@ Create Date: 2025-09-29 14:20:28.990148
import sqlalchemy as sa
from alembic import op
from sqlalchemy import text
from utils.database import is_mariadb, is_mysql
# revision identifiers, used by Alembic.
@@ -60,7 +61,7 @@ def upgrade() -> None:
connection = op.get_bind()
# For MariaDB compatibility, we limit the content index length
if is_mysql(connection) or is_mariadb(connection):
if is_mysql(connection) or is_mariadb(connection):
connection.execute(
text("CREATE INDEX idx_rom_notes_content ON rom_notes (content(100))")
)
@@ -86,7 +87,7 @@ def upgrade() -> None:
# Migrate existing notes from rom_user to rom_notes table
# Both note_raw_markdown and note_is_public columns exist from previous migrations
result = connection.execute(
text(
"""
@@ -122,9 +123,9 @@ def upgrade() -> None:
def downgrade() -> None:
"""Drop the rom_notes table and restore note columns to rom_user."""
connection = op.get_bind()
# Add back the old columns to rom_user
op.add_column(
"rom_user",
@@ -138,7 +139,7 @@ def downgrade() -> None:
)
# Migrate notes back to rom_user (take first note per user/rom)
result = connection.execute(
text(
"""
@@ -168,10 +169,12 @@ def downgrade() -> None:
)
# Drop indexes and table
if is_mysql(connection) or is_mariadb(connection):
if is_mysql(connection) or is_mariadb(connection):
# The content index was created with a specific length using raw SQL for MySQL/MariaDB,
# so we drop it with raw SQL as well.
connection.execute(text("DROP INDEX IF EXISTS idx_rom_notes_content ON rom_notes"))
connection.execute(
text("DROP INDEX IF EXISTS idx_rom_notes_content ON rom_notes")
)
else:
# For other databases, the content index was created with op.create_index.
op.drop_index("idx_rom_notes_content", table_name="rom_notes")

98
backend/tests/handler/database/test_rom_note_migration.py
View File

@@ -3,30 +3,29 @@ Unit tests for rom_notes table migration.
Test file: tests/handler/database/test_rom_note_migration.py
"""
import os
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, text, inspect
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.pool import NullPool
import os
@pytest.fixture(params=["postgresql", "mysql"])
def db_engine(request):
"""Create test database engines for different dialects."""
dialect = request.param
if dialect == "postgresql":
database_url = os.getenv(
"TEST_POSTGRES_URL",
"postgresql://romm:romm@localhost:5432/romm"
"TEST_POSTGRES_URL", "postgresql://romm:romm@localhost:5432/romm"
)
else: # mysql
database_url = os.getenv(
"TEST_MYSQL_URL",
"mysql+pymysql://romm:romm@127.0.0.1/romm"
"TEST_MYSQL_URL", "mysql+pymysql://romm:romm@127.0.0.1/romm"
)
engine = create_engine(database_url, poolclass=NullPool)
yield engine, dialect
engine.dispose()
@@ -46,7 +45,7 @@ def get_migration_revisions(config):
# You'll need to update these to match your actual revision IDs
# You can find them with: alembic history
ROM_NOTES_REVISION = "0057_multi_notes" # Your upgrade revision
PREVIOUS_REVISION = "0056_gamelist_xml" # The revision before rom_notes
PREVIOUS_REVISION = "0056_gamelist_xml" # The revision before rom_notes
return ROM_NOTES_REVISION, PREVIOUS_REVISION
@@ -54,56 +53,67 @@ def test_upgrade_creates_table_and_indexes(alembic_config):
"""Test that upgrade creates rom_notes table with all indexes."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
# Start from previous revision
command.downgrade(config, PREV_REV)
# Verify table doesn't exist yet
inspector = inspect(engine)
assert "rom_notes" not in inspector.get_table_names()
# Upgrade to rom_notes revision
command.upgrade(config, ROM_NOTES_REV)
# Verify table exists
inspector = inspect(engine)
assert "rom_notes" in inspector.get_table_names()
# Verify columns
columns = {col["name"] for col in inspector.get_columns("rom_notes")}
expected = {"id", "rom_id", "user_id", "title", "content", "is_public", "created_at", "updated_at"}
expected = {
"id",
"rom_id",
"user_id",
"title",
"content",
"is_public",
"created_at",
"updated_at",
}
assert expected.issubset(columns), f"Missing columns: {expected - columns}"
# Verify indexes
indexes = {idx["name"] for idx in inspector.get_indexes("rom_notes")}
expected_indexes = {
"idx_rom_notes_public",
"idx_rom_notes_rom_user",
"idx_rom_notes_title",
"idx_rom_notes_content"
"idx_rom_notes_content",
}
assert expected_indexes.issubset(indexes), f"Missing indexes: {expected_indexes - indexes}"
assert expected_indexes.issubset(
indexes
), f"Missing indexes: {expected_indexes - indexes}"
def test_downgrade_removes_table(alembic_config):
"""Test that downgrade removes rom_notes table."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
# Ensure we're at rom_notes revision
command.upgrade(config, ROM_NOTES_REV)
# Verify table exists
inspector = inspect(engine)
assert "rom_notes" in inspector.get_table_names()
# Downgrade to previous revision
command.downgrade(config, PREV_REV)
# Verify table is removed
inspector = inspect(engine)
assert "rom_notes" not in inspector.get_table_names()
# Verify old columns are restored to rom_user
columns = {col["name"] for col in inspector.get_columns("rom_user")}
assert "note_raw_markdown" in columns
@@ -114,23 +124,23 @@ def test_upgrade_removes_old_columns_from_rom_user(alembic_config):
"""Test that upgrade removes old note columns from rom_user."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
# Start from previous revision (where old columns exist)
command.downgrade(config, PREV_REV)
inspector = inspect(engine)
columns_before = {col["name"] for col in inspector.get_columns("rom_user")}
# Old columns should exist
assert "note_raw_markdown" in columns_before
assert "note_is_public" in columns_before
# Upgrade
command.upgrade(config, ROM_NOTES_REV)
inspector = inspect(engine)
columns_after = {col["name"] for col in inspector.get_columns("rom_user")}
# Old columns should be removed (if your upgrade does this)
# Comment out these assertions if your upgrade keeps the old columns
# assert "note_raw_markdown" not in columns_after
@@ -141,22 +151,22 @@ def test_indexes_are_dialect_specific(alembic_config):
"""Test that content index works with dialect-specific syntax."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
# Upgrade to rom_notes revision
command.upgrade(config, ROM_NOTES_REV)
inspector = inspect(engine)
indexes = {idx["name"] for idx in inspector.get_indexes("rom_notes")}
# Verify content index exists
assert "idx_rom_notes_content" in indexes
# Test that we can query using content column
with engine.connect() as conn:
# This should work regardless of dialect
result = conn.execute(
text("SELECT COUNT(*) as cnt FROM rom_notes WHERE content = :content"),
{"content": "test"}
{"content": "test"},
)
count = result.scalar()
assert count is not None # Should be 0, but no error
@@ -166,22 +176,22 @@ def test_foreign_keys_and_constraints(alembic_config):
"""Test that foreign keys and constraints are properly created."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
command.upgrade(config, ROM_NOTES_REV)
inspector = inspect(engine)
# Check foreign keys
foreign_keys = inspector.get_foreign_keys("rom_notes")
assert len(foreign_keys) >= 2
fk_columns = {fk["constrained_columns"][0] for fk in foreign_keys}
assert "rom_id" in fk_columns and "user_id" in fk_columns
# Check primary key
pk = inspector.get_pk_constraint("rom_notes")
assert "id" in pk["constrained_columns"]
# Check NOT NULL constraints
columns = {col["name"]: col for col in inspector.get_columns("rom_notes")}
assert columns["rom_id"]["nullable"] is False
@@ -193,22 +203,22 @@ def test_full_migration_cycle(alembic_config):
"""Test complete upgrade -> downgrade -> upgrade cycle."""
config, engine, dialect = alembic_config
ROM_NOTES_REV, PREV_REV = get_migration_revisions(config)
# Start from previous
command.downgrade(config, PREV_REV)
inspector = inspect(engine)
assert "rom_notes" not in inspector.get_table_names()
# Upgrade
command.upgrade(config, ROM_NOTES_REV)
inspector = inspect(engine)
assert "rom_notes" in inspector.get_table_names()
# Downgrade
command.downgrade(config, PREV_REV)
inspector = inspect(engine)
assert "rom_notes" not in inspector.get_table_names()
# Upgrade again
command.upgrade(config, ROM_NOTES_REV)
inspector = inspect(engine)