110 lines
4.8 KiB
Python
110 lines
4.8 KiB
Python
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.db.models import Base
|
|
from app.db.session import initialize_schema
|
|
|
|
|
|
def test_initialize_schema_creates_fresh_database() -> None:
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
try:
|
|
initialize_schema(engine)
|
|
columns = {col["name"] for col in inspect(engine).get_columns("sync_users")}
|
|
assert "notify_email_enabled" in columns
|
|
assert "notification_email" in columns
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_initialize_schema_adds_missing_columns_without_dropping_existing_rows() -> None:
|
|
"""Regression test: a database created before notify_email_enabled/
|
|
notification_email existed must gain those columns in place, keeping
|
|
every already-stored user row intact -- create_all() alone would not add
|
|
columns to a table that already exists."""
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
try:
|
|
# Simulate the pre-existing production schema by creating every
|
|
# table via the current models, then dropping the two new columns
|
|
# back off sync_users the only way sqlite allows: rebuild the table.
|
|
Base.metadata.create_all(engine)
|
|
with engine.begin() as conn:
|
|
conn.execute(text("ALTER TABLE sync_users RENAME TO sync_users_old"))
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE sync_users (
|
|
id INTEGER PRIMARY KEY,
|
|
name VARCHAR(120) NOT NULL,
|
|
enabled BOOLEAN NOT NULL,
|
|
health_state VARCHAR NOT NULL,
|
|
mywhoosh_state VARCHAR(32) NOT NULL,
|
|
garmin_state VARCHAR(32) NOT NULL,
|
|
action_reason TEXT,
|
|
mywhoosh_email_enc TEXT NOT NULL,
|
|
mywhoosh_password_enc TEXT NOT NULL,
|
|
garmin_email_enc TEXT NOT NULL,
|
|
garmin_password_enc TEXT NOT NULL,
|
|
created_at DATETIME,
|
|
updated_at DATETIME
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO sync_users (
|
|
id, name, enabled, health_state, mywhoosh_state, garmin_state,
|
|
action_reason, mywhoosh_email_enc, mywhoosh_password_enc,
|
|
garmin_email_enc, garmin_password_enc, created_at, updated_at
|
|
)
|
|
SELECT id, name, enabled, health_state, mywhoosh_state, garmin_state,
|
|
action_reason, mywhoosh_email_enc, mywhoosh_password_enc,
|
|
garmin_email_enc, garmin_password_enc, created_at, updated_at
|
|
FROM sync_users_old
|
|
"""
|
|
)
|
|
)
|
|
conn.execute(text("DROP TABLE sync_users_old"))
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO sync_users (
|
|
id, name, enabled, health_state, mywhoosh_state, garmin_state,
|
|
mywhoosh_email_enc, mywhoosh_password_enc, garmin_email_enc, garmin_password_enc
|
|
) VALUES (
|
|
1, 'Existing User', 1, 'healthy', 'connected', 'connected',
|
|
'enc-mw-email', 'enc-mw-pass', 'enc-garmin-email', 'enc-garmin-pass'
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
|
|
columns_before = {col["name"] for col in inspect(engine).get_columns("sync_users")}
|
|
assert "notify_email_enabled" not in columns_before
|
|
|
|
initialize_schema(engine)
|
|
|
|
columns_after = {col["name"] for col in inspect(engine).get_columns("sync_users")}
|
|
assert "notify_email_enabled" in columns_after
|
|
assert "notification_email" in columns_after
|
|
|
|
with engine.connect() as conn:
|
|
row = conn.execute(text("SELECT name, notify_email_enabled, notification_email FROM sync_users")).one()
|
|
assert row.name == "Existing User"
|
|
assert row.notify_email_enabled == 0
|
|
assert row.notification_email is None
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_initialize_schema_is_idempotent() -> None:
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
try:
|
|
initialize_schema(engine)
|
|
initialize_schema(engine)
|
|
columns = {col["name"] for col in inspect(engine).get_columns("sync_users")}
|
|
assert "notify_email_enabled" in columns
|
|
finally:
|
|
engine.dispose()
|