fix: add migration logic for warnings column and update tests

Critical fixes identified in code review:

1. Add warnings column migration to _migrate_schema()
   - Checks if warnings column exists in jobs table
   - Adds column via ALTER TABLE if missing
   - Ensures existing databases get new column on upgrade

2. Document CHECK constraint limitation
   - Added docstring explaining ALTER TABLE cannot add CHECK constraints
   - Notes that "downloading_data" status requires fresh DB or manual migration

3. Add comprehensive migration tests
   - test_migration_adds_warnings_column: Verifies warnings column migration
   - test_migration_adds_simulation_run_id_column: Tests existing migration
   - Both tests include cleanup to prevent cross-test contamination

4. Update test fixtures and expectations
   - Updated clean_db fixture to delete from all 9 tables
   - Fixed table count assertions (6 -> 9 tables)
   - Updated expected columns in schema tests

All 21 database tests now pass.
This commit is contained in:
2025-11-01 23:17:25 -04:00
parent 711ae5df73
commit baa44c208a
3 changed files with 158 additions and 5 deletions

View File

@@ -286,7 +286,13 @@ def cleanup_dev_database(db_path: str = "data/trading_dev.db", data_path: str =
def _migrate_schema(cursor: sqlite3.Cursor) -> None:
"""Migrate existing database schema to latest version."""
"""
Migrate existing database schema to latest version.
Note: Cannot add CHECK constraints to existing columns via ALTER TABLE.
The "downloading_data" status in jobs table requires a fresh database
or manual migration if upgrading from an older schema version.
"""
# Check if positions table exists and has simulation_run_id column
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='positions'")
if cursor.fetchone():
@@ -299,6 +305,18 @@ def _migrate_schema(cursor: sqlite3.Cursor) -> None:
ALTER TABLE positions ADD COLUMN simulation_run_id TEXT
""")
# Check if jobs table exists and has warnings column
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='jobs'")
if cursor.fetchone():
cursor.execute("PRAGMA table_info(jobs)")
columns = [row[1] for row in cursor.fetchall()]
if 'warnings' not in columns:
# Add warnings column to existing jobs table
cursor.execute("""
ALTER TABLE jobs ADD COLUMN warnings TEXT
""")
def _create_indexes(cursor: sqlite3.Cursor) -> None:
"""Create database indexes for query performance."""