From bd6b137c68ef0b56eccfb022fddeb24241766505 Mon Sep 17 00:00:00 2001 From: Flea Flicker Date: Tue, 2 Jun 2026 02:53:46 +0000 Subject: [PATCH] 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 --- tests/conftest.py | 36 ++++++++++++++++++++++++++++-------- 1 file changed, 28 insertions(+), 8 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 8bc7d53..b3a226f 100644 --- a/tests/conftest.py +++ b/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(