Fix SQLite timestamp, UUID, and User.id binding in test fixtures

Builds on the partial bd6b137 fix (which only stripped server_default
expressions) by also:

- Add _StringUUID TypeDecorator: lets Text/String/UUID columns accept
  uuid.UUID values on bind (SQLite has no native UUID type) and returns
  uuid.UUID on read so existing test assertions like
  isinstance(store.id, uuid.UUID) still pass.

- Replace UUID column types with _StringUUID before create_all so
  CREATE TABLE uses CHAR(36) instead of the native UUID type that
  SQLite can't bind.

- Extend before_insert listener to also set Text PK columns (User.id)
  and func.now()-stripped columns (ingested_at) to Python-side defaults
  so INSERTs without explicit values succeed under SQLite.

- Switch _create_test_user_and_session to use 32-char hex user/session
  ids so they match the format bound by the TypeDecorator on FK reads.

- Simplify test_encrypted_json.py to use the shared engine/session
  fixtures from conftest instead of duplicating its own broken engine.

Tests passing: tests/test_models.py (14), tests/test_encrypted_json.py (6).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/conftest.py

@@ -10,9 +10,10 @@ from datetime import UTC, datetime, timedelta
 
 import pytest
 from httpx import ASGITransport, AsyncClient
-from sqlalchemy import create_engine, event, text
+from sqlalchemy import String, TypeDecorator, Uuid, create_engine, event, text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.types import CHAR
 
 from cartsnitch_api.config import settings as cartsnitch_settings
 from cartsnitch_api.database import get_db
@@ -20,12 +21,100 @@ from cartsnitch_api.main import create_app
 from cartsnitch_api.models import Base
 
 
+class _StringUUID(TypeDecorator):
+    """TypeDecorator that lets Text/String/UUID columns accept uuid.UUID on bind.
+
+    SQLite has no native UUID type — passing a ``uuid.UUID`` raises
+    ``type 'UUID' is not supported``. This stores UUID values as their hex
+    string in the DB, accepts either uuid.UUID or str at bind time, and
+    returns uuid.UUID on read so existing test assertions like
+    ``isinstance(store.id, uuid.UUID)`` still work.
+    """
+
+    impl = CHAR(36)
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return None
+        if isinstance(value, uuid.UUID):
+            return str(value)
+        return str(value)
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return None
+        if isinstance(value, uuid.UUID):
+            return value
+        return uuid.UUID(value)
+
+
 def _set_timestamp_defaults(mapper, connection, target):
-    """Populate created_at/updated_at before insert for SQLite compatibility."""
+    """Populate created_at/updated_at and missing PK IDs for SQLite.
+
+    SQLite can't bind ``uuid.UUID`` objects to Text/String columns, and has
+    no server-side default for ``func.now()`` or ``gen_random_uuid()``. We
+    strip those server_defaults elsewhere; this listener fills in
+    Python-side timestamp defaults at insert time, generates IDs for PK
+    columns that have no default, and populates ``func.now()`` columns
+    whose server_default was stripped (e.g. ``ingested_at``). UUID values
+    for non-PK columns are converted by the ``_StringUUID`` TypeDecorator.
+    """
     now = datetime.now(UTC)
-    for col in [c for c in mapper.columns if c.key in ("created_at", "updated_at")]:
+    for col in mapper.columns:
-        if getattr(target, col.key, None) is None:
+        key = col.key
-            setattr(target, col.key, now)
+        if key in ("created_at", "updated_at"):
+            if getattr(target, key, None) is None:
+                setattr(target, key, now)
+            continue
+        if col.primary_key and getattr(target, key, None) is None:
+            setattr(target, key, str(uuid.uuid4()))
+            continue
+        if getattr(col, "_sqlite_default_now", False) and getattr(target, key, None) is None:
+            setattr(target, key, now)
+
+
+def _adapt_columns_for_sqlite():
+    """Strip Postgres-only server_defaults and adapt UUID columns for SQLite.
+
+    Must be called BEFORE ``Base.metadata.create_all`` so the DDL reflects
+    the adapted column types.
+    """
+    for tbl in Base.metadata.tables.values():
+        for col in tbl.columns.values():
+            # Strip PostgreSQL-specific function server_defaults (gen_random_uuid,
+            # gen_random_bytes, now()) but keep simple string-literal defaults
+            # like ``server_default="false"`` since they work in SQLite.
+            sd = col.server_default
+            if sd is not None:
+                sd_text = str(sd.arg) if hasattr(sd, "arg") else str(sd)
+                sd_text = sd_text.lower()
+                if any(x in sd_text for x in ["gen_random_uuid", "gen_random_bytes", "now()"]):
+                    col.server_default = None
+                    if "now()" in sd_text and not col.nullable:
+                        col._sqlite_default_now = True  # type: ignore[attr-defined]
+
+            # Replace UUID column types with a SQLite-compatible TypeDecorator
+            if isinstance(col.type, Uuid):
+                col.type = _StringUUID()
+
+            # Text/String PK columns without a default need the _StringUUID type
+            # so the before_insert listener can generate hex-string IDs.
+            if col.primary_key and col.default is None and col.server_default is None:
+                if not isinstance(col.type, _StringUUID):
+                    col.type = _StringUUID()
+
+            # FK columns that may receive uuid.UUID values from test code
+            if col.foreign_keys and not col.primary_key and isinstance(col.type, String):
+                col.type = _StringUUID()
+
+
+def _register_event_listeners():
+    """Attach before_insert listener to every mapped class."""
+    for cls in Base.registry._class_registry.values():
+        if hasattr(cls, "__mapper__"):
+            event.listen(cls, "before_insert", _set_timestamp_defaults)
+
 
 TEST_JWT_SECRET = secrets.token_urlsafe(32)
 TEST_SERVICE_KEY = secrets.token_urlsafe(32)
@@ -61,28 +150,13 @@ def disable_rate_limiting():
 def engine():
     """Sync in-memory SQLite engine for model unit tests.
 
-    Strips PostgreSQL-specific server_default expressions and provides
+    Strips PostgreSQL-specific server_default expressions, replaces UUID
-    Python-side defaults for SQLite compatibility.
+    column types with a SQLite-compatible TypeDecorator, and registers a
+    before_insert event listener to populate timestamps.
     """
     eng = create_engine("sqlite:///:memory:")
-
+    _adapt_columns_for_sqlite()
-    for tbl in Base.metadata.tables.values():
+    _register_event_listeners()
-        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()
-                # 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()
@@ -106,22 +180,8 @@ async def db_engine():
         cursor.execute("PRAGMA foreign_keys=ON")
         cursor.close()
 
-    for tbl in Base.metadata.tables.values():
+    _adapt_columns_for_sqlite()
-        for col in tbl.columns.values():
+    _register_event_listeners()
-            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()
-                # 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)
@@ -212,11 +272,11 @@ async def _create_test_user_and_session(
     Returns (user_dict, session_token).  Better-Auth stores the raw token
     in the DB, so we insert it as-is.
     """
-    user_id = str(uuid.uuid4())
+    user_id = uuid.uuid4().hex
     email = user_overrides.get("email", "test@example.com")
     display_name = user_overrides.get("display_name", "Test User")
     session_token = secrets.token_urlsafe(32)
-    session_id = str(uuid.uuid4())
+    session_id = uuid.uuid4().hex
     now = datetime.now(UTC).isoformat()
     expires = (datetime.now(UTC) + timedelta(days=7)).isoformat()
 

tests/test_encrypted_json.py

@@ -5,42 +5,13 @@ import json
 import pytest
 from cryptography.fernet import Fernet
 from pydantic import ValidationError
-from sqlalchemy import column, create_engine, table, text
+from sqlalchemy import column, table, text
-from sqlalchemy.orm import sessionmaker
 
 from cartsnitch_api.config import settings
-from cartsnitch_api.models import Base
 from cartsnitch_api.models.store import Store
 from cartsnitch_api.models.user import User, UserStoreAccount
 
 
-@pytest.fixture
-def engine():
-    eng = create_engine("sqlite:///:memory:")
-
-    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:
-                    col.server_default = None
-
-    Base.metadata.create_all(eng)
-    yield eng
-    eng.dispose()
-
-
-@pytest.fixture
-def session(engine):
-    factory = sessionmaker(bind=engine)
-    with factory() as sess:
-        yield sess
-
-
 @pytest.fixture
 def store(session):
     s = Store(name="Test Store", slug="test-store")