#!/usr/bin/env python3 """ Tests for Story 7.02: AIVAMemoryAPI — GET /conversations/recent AIVA RLM Nexus PRD v2 — Track A, Module 7 Black box tests (BB1-BB4): verify public API behaviour from the outside. White box tests (WB1-WB4): verify internal cap logic, Pydantic shape, default parameter value, and empty-list type. All tests use httpx.ASGITransport — zero real server or network I/O. Database connections are mocked via FastAPI's dependency_overrides so no Elestio Postgres is required. """ from __future__ import annotations import sys from datetime import datetime, timezone from typing import Any from unittest.mock import MagicMock import httpx import pytest sys.path.insert(0, "/mnt/e/genesis-system") from api.aiva_memory_api import ( ConversationSummary, RecentConversationsResponse, app, get_pg_connection, get_recent_conversations, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _SAMPLE_ROWS = [ ( "conv-001", datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc), "Discussed strategy for TradiVoice launch", ["Send pricing deck", "Book demo"], "excited", ), ( "conv-002", datetime(2026, 2, 24, 9, 0, 0, tzinfo=timezone.utc), "Voice bridge deployment review", ["Fix TTS config"], "focused", ), ( "conv-003", datetime(2026, 2, 23, 8, 0, 0, tzinfo=timezone.utc), "AIVA memory architecture planning", [], "neutral", ), ( "conv-004", datetime(2026, 2, 22, 7, 0, 0, tzinfo=timezone.utc), "Older conversation", ["Old action"], "calm", ), ] def _make_mock_conn(rows: list) -> MagicMock: """Return a mock psycopg2 connection whose cursor returns *rows*.""" cursor = MagicMock() cursor.fetchall.return_value = rows conn = MagicMock() conn.cursor.return_value = cursor return conn def _client_with_conn(conn: Any) -> httpx.AsyncClient: """ Return an AsyncClient whose /conversations/recent endpoint uses *conn* as its injected Postgres connection (via FastAPI dependency_overrides). """ app.dependency_overrides[get_pg_connection] = lambda: conn transport = httpx.ASGITransport(app=app) return httpx.AsyncClient(transport=transport, base_url="http://test") def _client_no_db() -> httpx.AsyncClient: """Return an AsyncClient with no database connection (conn=None).""" app.dependency_overrides[get_pg_connection] = lambda: None transport = httpx.ASGITransport(app=app) return httpx.AsyncClient(transport=transport, base_url="http://test") # --------------------------------------------------------------------------- # Fixture: ensure dependency_overrides is cleaned up after every test # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def cleanup_overrides(): """Remove all dependency overrides after every test.""" yield app.dependency_overrides.clear() # --------------------------------------------------------------------------- # BB1: GET /conversations/recent → 200, conversations array present # --------------------------------------------------------------------------- class TestBB1_ReturnsConversationsArray: """BB1: GET /conversations/recent always returns HTTP 200 with a 'conversations' key containing a list.""" @pytest.mark.anyio async def test_status_code_200(self): conn = _make_mock_conn(_SAMPLE_ROWS[:3]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent") assert response.status_code == 200 @pytest.mark.anyio async def test_conversations_key_present(self): conn = _make_mock_conn(_SAMPLE_ROWS[:3]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent") assert "conversations" in response.json() @pytest.mark.anyio async def test_conversations_value_is_list(self): conn = _make_mock_conn(_SAMPLE_ROWS[:3]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent") assert isinstance(response.json()["conversations"], list) @pytest.mark.anyio async def test_response_content_type_is_json(self): conn = _make_mock_conn(_SAMPLE_ROWS[:3]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent") assert "application/json" in response.headers["content-type"] # --------------------------------------------------------------------------- # BB2: n=1 → exactly 1 conversation when DB has multiple # --------------------------------------------------------------------------- class TestBB2_NParamRespected: """BB2: The ?n query parameter controls how many rows are requested from the database and how many conversations appear in the response.""" @pytest.mark.anyio async def test_n_equals_1_returns_one_row(self): # Mock returns exactly what the real DB would — only 1 row conn = _make_mock_conn(_SAMPLE_ROWS[:1]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conversations = response.json()["conversations"] assert len(conversations) == 1 @pytest.mark.anyio async def test_n_equals_2_returns_two_rows(self): conn = _make_mock_conn(_SAMPLE_ROWS[:2]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=2") conversations = response.json()["conversations"] assert len(conversations) == 2 @pytest.mark.anyio async def test_n_param_is_forwarded_to_db_execute(self): """The cursor.execute call must receive the (possibly capped) n value.""" conn = _make_mock_conn(_SAMPLE_ROWS[:2]) async with _client_with_conn(conn) as client: await client.get("/conversations/recent?n=2") # Verify cursor.execute was called with n=2 as the LIMIT cursor = conn.cursor.return_value call_args = cursor.execute.call_args # Second arg to execute is a tuple containing the LIMIT value limit_value = call_args[0][1][0] assert limit_value == 2 # --------------------------------------------------------------------------- # BB3: No data → {"conversations": []} # --------------------------------------------------------------------------- class TestBB3_EmptyWhenNoData: """BB3: When the database returns no rows (or there is no connection) the response is HTTP 200 with an empty conversations list — never a 404.""" @pytest.mark.anyio async def test_empty_db_returns_empty_list(self): conn = _make_mock_conn([]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent") assert response.status_code == 200 assert response.json() == {"conversations": []} @pytest.mark.anyio async def test_no_connection_returns_empty_list(self): async with _client_no_db() as client: response = await client.get("/conversations/recent") assert response.status_code == 200 assert response.json() == {"conversations": []} @pytest.mark.anyio async def test_empty_response_is_not_404(self): async with _client_no_db() as client: response = await client.get("/conversations/recent") assert response.status_code != 404 # --------------------------------------------------------------------------- # BB4: n=100 → capped at 20 # --------------------------------------------------------------------------- class TestBB4_NCappedAt20: """BB4: Any n > 20 must be capped to 20 before the DB query is issued.""" @pytest.mark.anyio async def test_n_100_capped_to_20_in_db_call(self): conn = _make_mock_conn(_SAMPLE_ROWS[:4]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=100") # The cursor must have received LIMIT=20, not LIMIT=100 cursor = conn.cursor.return_value call_args = cursor.execute.call_args limit_value = call_args[0][1][0] assert limit_value == 20 @pytest.mark.anyio async def test_n_50_capped_to_20(self): conn = _make_mock_conn([]) async with _client_with_conn(conn) as client: await client.get("/conversations/recent?n=50") cursor = conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 20 @pytest.mark.anyio async def test_n_exactly_20_not_capped(self): """n=20 should remain 20 (cap is inclusive).""" conn = _make_mock_conn([]) async with _client_with_conn(conn) as client: await client.get("/conversations/recent?n=20") cursor = conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 20 # --------------------------------------------------------------------------- # WB1: n cap logic — unit test without HTTP layer # --------------------------------------------------------------------------- class TestWB1_CapLogic: """WB1: The cap logic n = min(n, 20) is applied correctly in the endpoint function itself, independent of the HTTP layer.""" @pytest.mark.anyio async def test_cap_applied_to_large_n(self): mock_conn = _make_mock_conn([]) result = await get_recent_conversations(n=999, pg_conn=mock_conn) cursor = mock_conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 20 @pytest.mark.anyio async def test_small_n_not_affected_by_cap(self): mock_conn = _make_mock_conn([]) await get_recent_conversations(n=5, pg_conn=mock_conn) cursor = mock_conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 5 @pytest.mark.anyio async def test_n_equals_1_not_affected(self): mock_conn = _make_mock_conn([]) await get_recent_conversations(n=1, pg_conn=mock_conn) cursor = mock_conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 1 # --------------------------------------------------------------------------- # WB2: Pydantic response model — all ConversationSummary fields present # --------------------------------------------------------------------------- class TestWB2_ResponseModelFields: """WB2: Every item in the conversations list has exactly the five fields required by ConversationSummary.""" _REQUIRED_FIELDS = {"id", "started_at", "summary", "action_items", "emotional_signal"} @pytest.mark.anyio async def test_conversation_has_all_required_fields(self): conn = _make_mock_conn(_SAMPLE_ROWS[:1]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conv = response.json()["conversations"][0] assert set(conv.keys()) == self._REQUIRED_FIELDS @pytest.mark.anyio async def test_action_items_is_a_list(self): conn = _make_mock_conn(_SAMPLE_ROWS[:1]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conv = response.json()["conversations"][0] assert isinstance(conv["action_items"], list) @pytest.mark.anyio async def test_id_is_a_string(self): conn = _make_mock_conn(_SAMPLE_ROWS[:1]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conv = response.json()["conversations"][0] assert isinstance(conv["id"], str) @pytest.mark.anyio async def test_started_at_is_a_string(self): conn = _make_mock_conn(_SAMPLE_ROWS[:1]) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conv = response.json()["conversations"][0] assert isinstance(conv["started_at"], str) @pytest.mark.anyio async def test_pydantic_model_direct_validation(self): """ConversationSummary Pydantic model accepts valid data without error.""" cs = ConversationSummary( id="x", started_at="2026-02-25T00:00:00+00:00", summary="Test", action_items=["do thing"], emotional_signal="happy", ) assert cs.id == "x" assert cs.action_items == ["do thing"] @pytest.mark.anyio async def test_none_action_items_row_returned_as_empty_list(self): """Rows where action_items IS NULL in Postgres must return [] not None.""" rows_with_null_actions = [ ( "conv-null", datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc), "Null action items test", None, # Simulates NULL Postgres array "calm", ) ] conn = _make_mock_conn(rows_with_null_actions) async with _client_with_conn(conn) as client: response = await client.get("/conversations/recent?n=1") conv = response.json()["conversations"][0] assert conv["action_items"] == [] # --------------------------------------------------------------------------- # WB3: Default n=3 when param not provided # --------------------------------------------------------------------------- class TestWB3_DefaultN: """WB3: When ?n is not specified the endpoint defaults to n=3.""" @pytest.mark.anyio async def test_default_n_is_3_in_db_call(self): conn = _make_mock_conn(_SAMPLE_ROWS[:3]) async with _client_with_conn(conn) as client: await client.get("/conversations/recent") cursor = conn.cursor.return_value limit_value = cursor.execute.call_args[0][1][0] assert limit_value == 3 @pytest.mark.anyio async def test_default_n_endpoint_signature(self): """The function signature must declare n=3 as the default.""" import inspect sig = inspect.signature(get_recent_conversations) n_param = sig.parameters["n"] assert n_param.default == 3, ( f"Expected n default=3, got {n_param.default}" ) # --------------------------------------------------------------------------- # WB4: Empty conversations is an empty list, not None # --------------------------------------------------------------------------- class TestWB4_EmptyConversationsIsNotNone: """WB4: The response conversations field is always a list — never None, never missing, never a different type.""" @pytest.mark.anyio async def test_empty_response_conversations_is_list(self): async with _client_no_db() as client: response = await client.get("/conversations/recent") conversations = response.json()["conversations"] assert conversations is not None assert isinstance(conversations, list) @pytest.mark.anyio async def test_empty_response_conversations_length_is_0(self): async with _client_no_db() as client: response = await client.get("/conversations/recent") conversations = response.json()["conversations"] assert len(conversations) == 0 @pytest.mark.anyio async def test_pydantic_response_model_empty_list(self): """RecentConversationsResponse with empty list serialises correctly.""" resp = RecentConversationsResponse(conversations=[]) assert resp.conversations == [] assert resp.model_dump() == {"conversations": []} # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": result = pytest.main([__file__, "-v", "--tb=short"]) sys.exit(result)