mail notification

tests/db/test_session.py

@@ -0,0 +1,109 @@
+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()