Fix SQLite timestamp and UUID server_defaults in test fixtures

Add _set_timestamp_defaults event listener to populate created_at/updated_at
before insert when using SQLite, since func.now() server_default is stripped.

Extended server_default stripping to include "now()" expressions for
timestamp columns (created_at, updated_at) that were failing with
NOT NULL constraint errors.

Fixes remaining CI test failures after PR #35:
- NOT NULL constraint failed: stores.created_at
- NOT NULL constraint failed: normalized_products.created_at

Co-Authored-By: Paperclip <noreply@paperclip.ing>

src/cartsnitch_api/database.py

@@ -6,14 +6,21 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 
 from cartsnitch_api.config import settings
 
-engine = create_async_engine(
+
-    settings.database_url,
+def _build_engine_kwargs() -> dict:
-    echo=False,
+    url = settings.database_url
-    pool_size=10,
+    kwargs: dict = {"echo": False}
-    max_overflow=20,
+    if not url.startswith("sqlite"):
-    pool_pre_ping=True,
+        kwargs.update(
-    pool_recycle=3600,
+            pool_size=10,
-)
+            max_overflow=20,
+            pool_pre_ping=True,
+            pool_recycle=3600,
+        )
+    return kwargs
+
+
+engine = create_async_engine(settings.database_url, **_build_engine_kwargs())
 async_session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
 

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,18 +61,27 @@ def disable_rate_limiting():
 def engine():
     """Sync in-memory SQLite engine for model unit tests.
 
-    Strips PostgreSQL-specific server_default expressions so SQLite can
+    Strips PostgreSQL-specific server_default expressions and provides
-    handle all column inserts without missing-function errors.
+    Python-side defaults for SQLite compatibility.
     """
     eng = create_engine("sqlite:///:memory:")
 
-    for table in Base.metadata.tables.values():
+    for tbl in Base.metadata.tables.values():
-        for col in table.columns.values():
+        for col in tbl.columns.values():
             sd = col.server_default
             if sd is not None:
-                expr_str = str(sd.expression).lower()
+                if not hasattr(sd, "expression"):
-                if "gen_random_uuid" in expr_str or "gen_random_bytes" in expr_str:
                     col.server_default = None
+                    continue
+                expr_str = str(sd.expression).lower()
+                # 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
@@ -89,13 +106,22 @@ async def db_engine():
         cursor.execute("PRAGMA foreign_keys=ON")
         cursor.close()
 
-    for table in Base.metadata.tables.values():
+    for tbl in Base.metadata.tables.values():
-        for col in table.columns.values():
+        for col in tbl.columns.values():
             sd = col.server_default
             if sd is not None:
-                expr_str = str(sd.expression).lower()
+                if not hasattr(sd, "expression"):
-                if "gen_random_uuid" in expr_str or "gen_random_bytes" in expr_str:
                     col.server_default = None
+                    continue
+                expr_str = str(sd.expression).lower()
+                # 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)

tests/test_encrypted_json.py

@@ -18,10 +18,13 @@ from cartsnitch_api.models.user import User, UserStoreAccount
 def engine():
     eng = create_engine("sqlite:///:memory:")
 
-    for table in Base.metadata.tables.values():
+    for tbl in Base.metadata.tables.values():
-        for col in table.columns.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:
                     col.server_default = None