add proper testing

Author: enitrat

API_DOCUMENTATION.md

@@ -360,11 +360,11 @@ The Query Insights API exposes raw interaction logs and lightweight analytics fo
 
 Fetch paginated user queries within a specific window.
 
-- `start_date` *(ISO 8601, required)* — inclusive lower bound.
-- `end_date` *(ISO 8601, required)* — inclusive upper bound.
-- `agent_id` *(optional)* — filter by agent id when provided.
-- `limit` *(default `100`)* — maximum rows returned.
-- `offset` *(default `0`)* — pagination offset.
+- `start_date` _(ISO 8601, required)_ — inclusive lower bound.
+- `end_date` _(ISO 8601, required)_ — inclusive upper bound.
+- `agent_id` _(optional)_ — filter by agent id when provided.
+- `limit` _(default `100`)_ — maximum rows returned.
+- `offset` _(default `0`)_ — pagination offset.
 
 **Response** `200 OK`
 
@@ -388,7 +388,7 @@ Fetch paginated user queries within a specific window.
 
 Trigger an asynchronous analysis job. The response returns immediately with the job identifier; the analysis runs in the background.
 
-**Request**
+#### Request
 
 ```json
 {

TASK_PRD.md

@@ -141,6 +141,9 @@ strict_optional = true
 testpaths = ["tests"]
 pythonpath = ["src"]
 asyncio_mode = "auto"
+markers = [
+  "db: marks tests that require a database (run by default, use -m 'not db' to skip)",
+]
 filterwarnings = [
   "ignore::DeprecationWarning",
   "ignore::PendingDeprecationWarning",

python/pyproject.toml

@@ -17,6 +17,33 @@
 logger = structlog.get_logger(__name__)
 
 
+def _normalize_row(row: dict | None, fields_with_defaults: dict[str, Any]) -> dict | None:
+    """
+    Parse stringified JSON fields in a row dictionary and apply defaults for None values.
+
+    Args:
+        row: Dictionary from database row (or None)
+        fields_with_defaults: Mapping of field names to default values
+
+    Returns:
+        Normalized dictionary with parsed JSON fields, or None if input row is None
+    """
+    if row is None:
+        return None
+
+    d = dict(row)
+    for field, default_val in fields_with_defaults.items():
+        val = d.get(field)
+        if isinstance(val, str):
+            try:
+                d[field] = json.loads(val)
+            except (json.JSONDecodeError, TypeError):
+                d[field] = default_val
+        elif val is None:
+            d[field] = default_val
+    return d
+
+
 async def create_user_interaction(interaction: UserInteraction) -> None:
     """Persist a user interaction in the database."""
     pool = await get_pool()
@@ -89,22 +116,24 @@ async def get_interactions(
             OFFSET ${offset_placeholder}
         """
         rows = await connection.fetch(data_query, *params)
-    return [dict(row) for row in rows], int(total)
+
+    # Normalize JSON fields that may be returned as strings by asyncpg
+    items = [_normalize_row(dict(row), {"chat_history": []}) for row in rows]
+    return items, int(total)
 
 
 async def create_analysis_job(params: dict[str, Any]) -> uuid.UUID:
     """Insert a new analysis job and return its identifier."""
     pool = await get_pool()
     async with pool.acquire() as connection:
-        job_id = await connection.fetchval(
+        return await connection.fetchval(
             """
             INSERT INTO query_analyses (status, analysis_parameters)
             VALUES ('pending', $1)
             RETURNING id
             """,
             json.dumps(params),
         )
-    return job_id
 
 
 async def update_analysis_job(
@@ -144,7 +173,7 @@ async def get_analysis_jobs(limit: int = 100) -> list[dict[str, Any]]:
             """,
             limit,
         )
-    return [dict(row) for row in rows]
+    return [_normalize_row(dict(row), {"analysis_parameters": {}}) for row in rows]
 
 
 async def get_analysis_job_by_id(job_id: uuid.UUID) -> dict[str, Any] | None:
@@ -164,4 +193,7 @@ async def get_analysis_job_by_id(job_id: uuid.UUID) -> dict[str, Any] | None:
             """,
             job_id,
         )
-    return dict(row) if row is not None else None
+    return _normalize_row(
+        dict(row) if row else None,
+        {"analysis_parameters": {}, "analysis_result": None}
+    )

python/src/cairo_coder/db/repository.py

@@ -4,19 +4,29 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import asyncpg
 import structlog
 
 from cairo_coder.config.manager import ConfigManager
 
 logger = structlog.get_logger(__name__)
 
-pool: asyncpg.Pool | None = None
+# Maintain one pool per running event loop to avoid cross-loop usage issues
+pools: dict[int, asyncpg.Pool] = {}
 
 
 async def get_pool() -> asyncpg.Pool:
-    """Return the global asyncpg connection pool, lazily creating it."""
-    global pool
+    """Return an asyncpg connection pool bound to the current event loop.
+
+    FastAPI's TestClient and AnyIO can run application code across different
+    event loops. Using a single cached pool may lead to cross-loop errors.
+    To prevent this, we maintain a pool per loop.
+    """
+    loop = asyncio.get_running_loop()
+    key = id(loop)
+    pool = pools.get(key)
     if pool is None:
         config = ConfigManager.load_config()
         try:
@@ -25,6 +35,7 @@ async def get_pool() -> asyncpg.Pool:
                 min_size=2,
                 max_size=10,
             )
+            pools[key] = pool
             logger.info("Database connection pool created successfully.")
         except Exception as exc:  # pragma: no cover - defensive logging
             logger.error("Failed to create database connection pool", error=str(exc))
@@ -34,11 +45,15 @@ async def get_pool() -> asyncpg.Pool:
 
 async def close_pool() -> None:
     """Close the asyncpg connection pool if it is active."""
-    global pool
-    if pool is not None:
-        await pool.close()
-        pool = None
-        logger.info("Database connection pool closed.")
+    import contextlib
+
+    # Close and clear all pools
+    global pools
+    for p in list(pools.values()):
+        with contextlib.suppress(Exception):
+            await p.close()
+    pools.clear()
+    logger.info("Database connection pool(s) closed.")
 
 
 async def execute_schema_scripts() -> None:

python/src/cairo_coder/db/session.py

@@ -28,11 +28,11 @@
 from cairo_coder.core.config import VectorStoreConfig
 from cairo_coder.core.rag_pipeline import RagPipeline
 from cairo_coder.core.types import Message, Role, StreamEventType
-from cairo_coder.dspy.document_retriever import SourceFilteredPgVectorRM
-from cairo_coder.dspy.suggestion_program import SuggestionGeneration
 from cairo_coder.db import session as db_session
 from cairo_coder.db.models import UserInteraction
 from cairo_coder.db.repository import create_user_interaction
+from cairo_coder.dspy.document_retriever import SourceFilteredPgVectorRM
+from cairo_coder.dspy.suggestion_program import SuggestionGeneration
 from cairo_coder.server.insights_api import router as insights_router
 from cairo_coder.utils.logging import setup_logging