Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -28,18 +28,31 @@


def upgrade() -> None:
# SQLite checks foreign keys on DROP TABLE too — an implicit "as if every row were
# deleted" pass — and app.database's connect listener runs every connection here
# with PRAGMA foreign_keys=ON. Batch mode's recreate-the-table strategy for adding
# a constraint needs to DROP `asset`, which real assettag/suggestion rows
# referencing it then block. A fresh test database never catches this: schema
# migrations run before any fixture has inserted a row, so there's nothing yet to
# violate. Migrations run under maintainer control, not the app-runtime mutations
# this pragma is meant to protect — safe to disable for just this rebuild.
op.execute('PRAGMA foreign_keys=OFF')
with op.batch_alter_table('asset', schema=None) as batch_op:
batch_op.add_column(sa.Column('parent_asset_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True))
batch_op.add_column(sa.Column('in_point', sa.Float(), nullable=True))
batch_op.add_column(sa.Column('out_point', sa.Float(), nullable=True))
batch_op.create_index('ix_asset_parent_asset_id', ['parent_asset_id'], unique=False)
batch_op.create_foreign_key('fk_asset_parent_asset_id_asset', 'asset', ['parent_asset_id'], ['id'])
op.execute('PRAGMA foreign_keys=ON')


def downgrade() -> None:
# Same reasoning as upgrade() — this recreates `asset` too.
op.execute('PRAGMA foreign_keys=OFF')
with op.batch_alter_table('asset', schema=None) as batch_op:
batch_op.drop_constraint('fk_asset_parent_asset_id_asset', type_='foreignkey')
batch_op.drop_index('ix_asset_parent_asset_id')
batch_op.drop_column('out_point')
batch_op.drop_column('in_point')
batch_op.drop_column('parent_asset_id')
op.execute('PRAGMA foreign_keys=ON')
47 changes: 47 additions & 0 deletions backend/tests/test_migrations.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,3 +166,50 @@ def test_the_migrated_schema_enforces_case_insensitive_tag_names(alembic_config)
"INSERT INTO tag (id, user_id, name, created_at)"
" VALUES ('3', 'other', 'nato', '2026-01-01')"
)


def test_add_clip_columns_survives_real_foreign_key_references(alembic_config):
"""The landmine this guards against was real, not theoretical (M7's first deploy).

`56ac14e89a0c` -> `7d4b9c1a6f28` batch-recreates `asset` to add its own
self-referencing foreign key. SQLite enforces foreign keys on DROP TABLE too — an
implicit "as if every row were deleted" check — so recreating `asset` while
`PRAGMA foreign_keys=ON` (every connection here runs with it on, see
app.database's connect listener) fails the moment another table holds a row that
actually references one, which is exactly what `assettag` does in any populated
database. A fresh database never catches this: schema migrations always run before
any fixture inserts a row, so there is nothing yet to violate — which is exactly
how this shipped clean and broke on the first real deploy. This seeds a real
cross-reference first, the way production always has one, so the migration is
proven against the case that matters rather than the empty case every other test
in this file uses.
"""
config, db_path = alembic_config
command.upgrade(config, "56ac14e89a0c")

with sqlite3.connect(db_path) as conn:
conn.execute("PRAGMA foreign_keys=ON")
conn.execute(
"INSERT INTO asset (id, user_id, name, asset_type, source, storage_key,"
" size_bytes, field_provenance, upload_date, modified_date, metadata_modified_date)"
" VALUES ('a1', 'u', 'Real asset', 'video', 'local_upload', 'k1',"
" 100, '{}', '2026-01-01', '2026-01-01', '2026-01-01')"
)
conn.execute(
"INSERT INTO tag (id, user_id, name, created_at)"
" VALUES ('t1', 'u', 'Tag', '2026-01-01')"
)
conn.execute(
"INSERT INTO assettag (asset_id, tag_id, created_at) VALUES ('a1', 't1', '2026-01-01')"
)
conn.commit()

command.upgrade(config, "head")

with sqlite3.connect(db_path) as conn:
assert conn.execute("SELECT id FROM asset WHERE id = 'a1'").fetchone() is not None
assert conn.execute(
"SELECT * FROM assettag WHERE asset_id = 'a1' AND tag_id = 't1'"
).fetchone() is not None
conn.execute("PRAGMA foreign_keys=ON")
assert conn.execute("PRAGMA foreign_key_check").fetchall() == []
Loading