fix(d2): datetime.now(UTC) everywhere + SQLITE-001 automation tests on ephemeral postgres

app/core/worker.py

@@ -346,7 +346,7 @@ async def cleanup_audit_log_job(ctx: dict[str, Any]) -> None:
     Iterates per-tenant for RLS compliance.
     """
     from sqlalchemy import text as sa_text, delete as sa_delete
-    from datetime import datetime, timedelta
+    from datetime import UTC, datetime, timedelta
 
     from app.core.db import get_worker_session_factory
     from app.models.audit import AuditLog
@@ -357,7 +357,7 @@ async def cleanup_audit_log_job(ctx: dict[str, Any]) -> None:
             tenant_result = await db.execute(sa_text("SELECT id FROM tenants"))
             tenant_ids = [row[0] for row in tenant_result]
 
-            cutoff = datetime.utcnow() - timedelta(days=365)
+            cutoff = datetime.now(UTC) - timedelta(days=365)
             total_deleted = 0
             for tenant_id in tenant_ids:
                 await db.execute(
@@ -389,7 +389,7 @@ async def cleanup_trash_job(ctx: dict[str, Any]) -> None:
     Default retention: 90 days in trash before permanent deletion.
     """
     from sqlalchemy import text as sa_text, delete as sa_delete
-    from datetime import datetime, timedelta
+    from datetime import UTC, datetime, timedelta
 
     from app.core.db import get_worker_session_factory
     from app.models.contact import Contact
@@ -401,7 +401,7 @@ async def cleanup_trash_job(ctx: dict[str, Any]) -> None:
             tenant_result = await db.execute(sa_text("SELECT id FROM tenants"))
             tenant_ids = [row[0] for row in tenant_result]
 
-            cutoff = datetime.utcnow() - timedelta(days=90)
+            cutoff = datetime.now(UTC) - timedelta(days=90)
             total_deleted = 0
 
             for tenant_id in tenant_ids:

app/plugins/builtins/automation/tests/test_automation.py

@@ -1,20 +1,44 @@
 """Tests for the Automation & Agents plugin.
 
-Uses pytest with async fixtures. Tests use SQLite in-memory database
-since PostgreSQL may not be available in the dev container.
+Uses pytest with async fixtures against an ephemeral PostgreSQL database
+(SQLITE-001 fix) — matches the project convention and exercises the real
+PGUUID/JSONB column types.
 """
 
 from __future__ import annotations
 
+import os
 import uuid
 from collections.abc import AsyncGenerator
 from datetime import UTC, datetime, timedelta
 
 import pytest
 import pytest_asyncio
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from app.core.db import Base
+import app.models  # noqa: F401  — registers core models
+import app.models.outbox  # noqa: F401  — event_outbox is NOT re-exported by app.models
+
+# Register ALL plugin models so create_all can resolve cross-plugin FKs
+# (e.g. entity_attachments.dms_file_id -> files) — same pattern as
+# scripts/sync_plugin_schema.py.
+import importlib
+import pkgutil
+
+import app.plugins.builtins as _builtins_pkg
+
+for _importer, _modname, _ispkg in pkgutil.iter_modules(_builtins_pkg.__path__):
+    if not _ispkg:
+        continue
+    try:
+        importlib.import_module(f"app.plugins.builtins.{_modname}.models")
+    except ImportError:
+        pass  # plugin without models module
+    except Exception:  # pragma: no cover - defensive
+        pass
+
 from app.plugins.builtins.automation.models import (
     AgentRun,
     AutomationRun,
@@ -25,36 +49,86 @@ from app.plugins.builtins.automation.services import (
     CronJobService,
 )
 
+
+def _ephemeral_db_url() -> str:
+    """Derive an ephemeral test DB URL from DATABASE_URL/.env.test."""
+    base_url = os.environ.get(
+        "DATABASE_URL",
+        "postgresql+asyncpg://leocrm_test:test123@localhost:5432/leocrm_test",
+    )
+    return f"{base_url.rsplit('/', 1)[0]}/automation_test_{uuid.uuid4().hex[:8]}"
+
+
 # ─── Fixtures ───
 
 
 @pytest_asyncio.fixture
 async def db() -> AsyncGenerator[AsyncSession, None]:
-    """Create an in-memory SQLite database for testing."""
-    engine = create_async_engine(
-        "sqlite+aiosqlite:///:memory:",
-        echo=False,
-    )
+    """Create an ephemeral PostgreSQL database for this test run."""
+    db_url = _ephemeral_db_url()
+    admin_url = db_url.rsplit("/", 1)[0] + "/postgres"
+
+    from sqlalchemy.ext.asyncio import create_async_engine as _cae
+
+    admin_engine = _cae(admin_url, isolation_level="AUTOCOMMIT")
+    async with admin_engine.connect() as conn:
+        await conn.execute(text(f'CREATE DATABASE "{db_url.rsplit("/", 1)[1]}"'))
+    await admin_engine.dispose()
+
+    # Plugin models use the pgvector Vector type — enable the extension in
+    # the fresh database before create_all runs (must connect to the target
+    # DB itself; CREATE EXTENSION has no ON DATABASE clause).
+    ext_engine = _cae(db_url, isolation_level="AUTOCOMMIT")
+    async with ext_engine.connect() as conn:
+        await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+    await ext_engine.dispose()
+
+    engine = create_async_engine(db_url, echo=False)
 
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
     async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
-    async with async_session() as session:
-        yield session
-
-    await engine.dispose()
+    try:
+        async with async_session() as session:
+            yield session
+    finally:
+        await engine.dispose()
+        admin_engine2 = _cae(admin_url, isolation_level="AUTOCOMMIT")
+        async with admin_engine2.connect() as conn:
+            await conn.execute(text(f'DROP DATABASE IF EXISTS "{db_url.rsplit("/", 1)[1]}"'))
+        await admin_engine2.dispose()
 
 
-@pytest.fixture
-def tenant_id() -> uuid.UUID:
-    return uuid.uuid4()
+@pytest_asyncio.fixture
+async def tenant_id(db: AsyncSession) -> uuid.UUID:
+    """Create a real tenant row — PostgreSQL enforces FKs, unlike SQLite."""
+    from app.models.tenant import Tenant
+
+    tid = uuid.uuid4()
+    db.add(Tenant(id=tid, name="Test Org", slug=f"test-{tid.hex[:8]}"))
+    await db.commit()
+    return tid
 
 
-@pytest.fixture
-def user_id() -> uuid.UUID:
-    return uuid.uuid4()
+@pytest_asyncio.fixture
+async def user_id(db: AsyncSession, tenant_id: uuid.UUID) -> uuid.UUID:
+    """Create a real user row belonging to the test tenant."""
+    from app.models.user import User
+
+    uid = uuid.uuid4()
+    db.add(
+        User(
+            id=uid,
+            email=f"test-{uid.hex[:8]}@example.com",
+            name="Test User",
+            password_hash="not-a-real-hash",
+            is_active=True,
+        )
+    )
+    await db.commit()
+    return uid
 
 
 # ─── AgentService Tests ───
@@ -425,11 +499,17 @@ class TestDryRunMode:
         assert automation.dry_run is True
 
     @pytest.mark.asyncio
-    async def test_dry_run_flag_in_run(self, db: AsyncSession, tenant_id: uuid.UUID):
+    async def test_dry_run_flag_in_run(self, db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID):
         """Test that dry_run flag is stored in AutomationRun."""
+        # PostgreSQL enforces the FK to automations — create a real one first
+        data = {"name": "dry-run-flag", "description": "", "trigger_type": "manual",
+                "trigger_config": {}, "conditions": [], "actions": [],
+                "is_active": True, "dry_run": True}
+        automation = await AutomationService.create(db, tenant_id, data, user_id=user_id)
+
         run = AutomationRun(
             tenant_id=tenant_id,
-            automation_id=uuid.uuid4(),
+            automation_id=automation.id,
             status="dry_run",
             started_at=datetime.now(UTC),
             dry_run=True,
@@ -475,7 +555,9 @@ class TestRateLimiting:
         )
         recent_runs = result.scalar() or 0
         assert recent_runs == 2
-        assert recent_runs < agent.max_executions_per_hour  # 2 < 2 is False, so limit would be hit
+        # With max_executions_per_hour=2 and 2 runs in the window, the limit
+        # is reached — the next execution must be blocked.
+        assert recent_runs >= agent.max_executions_per_hour
 
 
 # ─── Budget Limit Tests ───
@@ -512,7 +594,8 @@ class TestBudgetLimit:
             .where(AgentRun.agent_id == agent.id)
         )
         total_cost = float(cost_result.scalar() or 0.0)
-        assert total_cost == 0.6
+        # FLOAT column accumulates binary rounding (0.6000000000000001)
+        assert total_cost == pytest.approx(0.6)
         assert total_cost >= agent.budget_limit_usd  # 0.6 >= 0.5, budget exceeded
 
 

app/plugins/builtins/mcp_client/routes.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import logging
 import uuid
-from datetime import datetime
+from datetime import UTC, datetime
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
@@ -170,7 +170,7 @@ async def list_server_tools(
         tools_resp = await client.list_tools()
         # Update last_connected_at
         await db.execute(
-            update(McpServerConfigModel).where(McpServerConfigModel.id == sid).values(last_connected_at=datetime.utcnow())
+            update(McpServerConfigModel).where(McpServerConfigModel.id == sid).values(last_connected_at=datetime.now(UTC))
         )
         await db.commit()
         return tools_resp

app/routes/audit.py

@@ -6,7 +6,7 @@ import csv
 import io
 import json
 import uuid
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi.responses import StreamingResponse
@@ -167,7 +167,7 @@ async def audit_retention_cleanup(
     Default retention: 365 days.
     """
     tenant_id = uuid.UUID(current_user["tenant_id"])
-    cutoff = datetime.utcnow() - timedelta(days=retention_days)
+    cutoff = datetime.now(UTC) - timedelta(days=retention_days)
 
     q = delete(AuditLog).where(
         AuditLog.tenant_id == tenant_id,

app/services/backup_service.py

@@ -156,7 +156,7 @@ async def create_backup(
         # Update backup record
         backup.status = "completed"
         backup.size_bytes = size_bytes
-        backup.completed_at = datetime.utcnow()
+        backup.completed_at = datetime.now(UTC)
         await db.flush()
         await db.refresh(backup)
 

app/services/webhook_service.py

@@ -9,6 +9,7 @@ import json
 import logging
 import socket
 import uuid
+from datetime import UTC, datetime
 from typing import Any
 from urllib.parse import urlparse
 
@@ -218,7 +219,7 @@ async def send_webhook(
     body = {
         "event": event_name,
         "payload": payload,
-        "timestamp": __import__("datetime").datetime.utcnow().isoformat() + "Z",
+        "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
     }
     body_bytes = json.dumps(body, default=str).encode("utf-8")