tests/conftest.py

@@ -19,6 +19,14 @@ from cartsnitch_api.database import get_db
 from cartsnitch_api.main import create_app
 from cartsnitch_api.models import Base
 
+
+def _set_timestamp_defaults(mapper, connection, target):
+    """Populate created_at/updated_at before insert for SQLite compatibility."""
+    now = datetime.now(UTC)
+    for col in [c for c in mapper.columns if c.key in ("created_at", "updated_at")]:
+        if getattr(target, col.key, None) is None:
+            setattr(target, col.key, now)
+
 TEST_JWT_SECRET = secrets.token_urlsafe(32)
 TEST_SERVICE_KEY = secrets.token_urlsafe(32)
 TEST_FERNET_KEY = "7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8="
@@ -53,22 +61,28 @@ def disable_rate_limiting():
 def engine():
     """Sync in-memory SQLite engine for model unit tests.
 
-    Strips ALL PostgreSQL-specific server_default expressions so SQLite can
-    handle all column inserts without missing-function errors.
+    Strips PostgreSQL-specific server_default expressions and provides
+    Python-side defaults for SQLite compatibility.
     """
     eng = create_engine("sqlite:///:memory:")
 
-    for table in Base.metadata.tables.values():
-        for col in table.columns.values():
+    for tbl in Base.metadata.tables.values():
+        for col in tbl.columns.values():
             sd = col.server_default
             if sd is not None:
                 if not hasattr(sd, "expression"):
                     col.server_default = None
                     continue
                 expr_str = str(sd.expression).lower()
-                if "gen_random_uuid" in expr_str or "gen_random_bytes" in expr_str:
+                # Strip PostgreSQL-specific defaults
+                if any(x in expr_str for x in ["gen_random_uuid", "gen_random_bytes", "now()"]):
                     col.server_default = None
 
+    # Register event listener to populate timestamps on insert
+    for cls in Base.registry._class_registry.values():
+        if hasattr(cls, "__mapper__"):
+            event.listen(cls, "before_insert", _set_timestamp_defaults)
+
     Base.metadata.create_all(eng)
     yield eng
     eng.dispose()
@@ -92,17 +106,23 @@ async def db_engine():
         cursor.execute("PRAGMA foreign_keys=ON")
         cursor.close()
 
-    for table in Base.metadata.tables.values():
-        for col in table.columns.values():
+    for tbl in Base.metadata.tables.values():
+        for col in tbl.columns.values():
             sd = col.server_default
             if sd is not None:
                 if not hasattr(sd, "expression"):
                     col.server_default = None
                     continue
                 expr_str = str(sd.expression).lower()
-                if "gen_random_uuid" in expr_str or "gen_random_bytes" in expr_str:
+                # Strip PostgreSQL-specific defaults
+                if any(x in expr_str for x in ["gen_random_uuid", "gen_random_bytes", "now()"]):
                     col.server_default = None
 
+    # Register event listener to populate timestamps on insert
+    for cls in Base.registry._class_registry.values():
+        if hasattr(cls, "__mapper__"):
+            event.listen(cls, "before_insert", _set_timestamp_defaults)
+
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
         await conn.execute(