From 471f96b65438914cc8a8d07f3853ceb9baa86c47 Mon Sep 17 00:00:00 2001 From: Barcode Betty Date: Tue, 2 Jun 2026 03:37:40 +0000 Subject: [PATCH] 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 --- tests/conftest.py | 146 ++++++++++++++++++++++++----------- tests/test_encrypted_json.py | 31 +------- 2 files changed, 104 insertions(+), 73 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index b3a226f..1342238 100644 --- a/tests/conftest.py +++ b/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")]: - if getattr(target, col.key, None) is None: - setattr(target, col.key, now) + for col in mapper.columns: + key = col.key + 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 - Python-side defaults for SQLite compatibility. + Strips PostgreSQL-specific server_default expressions, replaces UUID + column types with a SQLite-compatible TypeDecorator, and registers a + before_insert event listener to populate timestamps. """ 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() - # 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) - + _adapt_columns_for_sqlite() + _register_event_listeners() 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(): - 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) + _adapt_columns_for_sqlite() + _register_event_listeners() 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() diff --git a/tests/test_encrypted_json.py b/tests/test_encrypted_json.py index 08b16d7..a19ab94 100644 --- a/tests/test_encrypted_json.py +++ b/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.orm import sessionmaker +from sqlalchemy import column, table, text 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")