#!/usr/bin/env python3 """ Story 9.04 — Module 9 RAG MCP Tools Integration Tests ======================================================= All external calls (Qdrant, PostgreSQL, Gemini, orchestrator) are FULLY MOCKED. No real network traffic or credentials required. Test summary (13 tests): Class TestSearchPlatformKb (4 tests — black-box): BB test_search_returns_results — Non-empty query → rag_context result returned BB test_platform_filter_passed — platform="hubspot" → rag_query called, filtered BB test_empty_query_returns_message — Empty query → helpful message, no rag call BB test_top_k_passed — top_k=10 → passed through to underlying call Class TestListPlatformKbs (3 tests — black-box): BB test_list_shows_platforms — Registered platforms appear in output BB test_list_shows_stats — Vector counts and PG history appear BB test_list_empty_graceful — No platforms registered → helpful message Class TestIngestPlatformKb (4 tests — black-box + white-box): BB test_trigger_ingestion — Valid platform → ingest_platform called BB test_invalid_platform_error — Unknown platform → error message returned BB test_max_pages_respected — max_pages=50 → passed to orchestrator BB test_returns_summary — Output includes pages/chunks/vectors counts Class TestToolRegistration (2 tests — white-box): WB test_register_adds_3_tools — register_kb_tools → 3 new tools added WB test_concurrent_search_safe — Multiple search calls → all succeed """ from __future__ import annotations import asyncio import sys from typing import Any, Dict, List from unittest.mock import AsyncMock, MagicMock, patch, call import pytest # ── Ensure project root importable ──────────────────────────────────────────── _PROJECT_ROOT = "/mnt/e/genesis-system" if _PROJECT_ROOT not in sys.path: sys.path.insert(0, _PROJECT_ROOT) # ────────────────────────────────────────────────────────────────────────────── # Fixtures # ────────────────────────────────────────────────────────────────────────────── @pytest.fixture() def mock_mcp(): """A minimal FastMCP stand-in that records registered tools.""" registered: dict[str, Any] = {} class FakeMCP: def tool(self): def decorator(fn): registered[fn.__name__] = fn return fn return decorator @property def _tools(self): return registered return FakeMCP() @pytest.fixture() def kb_tools_module(mock_mcp): """Import and register kb_tools on the mock MCP, return (mcp, module).""" # Import the module under test import importlib import mcp_servers.genesis_core_kb_tools_mod as _mod # we'll patch path below from mcp_servers.genesis_core import kb_tools as mod # noqa: F401 mod.register_kb_tools(mock_mcp) return mock_mcp, mod # We import kb_tools directly since it's not inside a package @pytest.fixture() def registered_mcp(mock_mcp): """Register kb_tools on a mock MCP and return it.""" # Import the implementation module sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import importlib import kb_tools as mod mod.register_kb_tools(mock_mcp) return mock_mcp # ────────────────────────────────────────────────────────────────────────────── # Helper: build the registered tool map cleanly # ────────────────────────────────────────────────────────────────────────────── def _get_tools(mock_mcp) -> dict[str, Any]: return mock_mcp._tools # ────────────────────────────────────────────────────────────────────────────── # Class TestSearchPlatformKb # ────────────────────────────────────────────────────────────────────────────── class TestSearchPlatformKb: """BB tests for search_platform_kb tool.""" def _make_mcp(self): registered: dict[str, Any] = {} class FakeMCP: def tool(self): def decorator(fn): registered[fn.__name__] = fn return fn return decorator @property def _tools(self): return registered mcp = FakeMCP() sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import kb_tools as mod mod.register_kb_tools(mcp) return mcp @patch("core.rag_query.rag_context") def test_search_returns_results(self, mock_rag_context): """BB: A valid query returns the context string from rag_context.""" mock_rag_context.return_value = ( "=== BLOODSTREAM KNOWLEDGE (3 matches) ===\n" "[1] Telnyx AI Assistants (score: 0.92, type: PLATFORM_KB)\n" " Source: https://developers.telnyx.com/docs/ai\n" " How to create an AI assistant on Telnyx...\n" ) mcp = self._make_mcp() search = mcp._tools["search_platform_kb"] result = search(query="How do I create a Telnyx AI assistant?", top_k=5) assert "BLOODSTREAM KNOWLEDGE" in result assert "Telnyx" in result @patch("core.rag_query.rag_query") def test_platform_filter_passed(self, mock_rag_query): """BB: platform='hubspot' triggers rag_query with client-side filter.""" mock_rag_query.return_value = [ { "title": "HubSpot CRM Setup", "score": 0.85, "type": "PLATFORM_KB", "source": "https://knowledge.hubspot.com/crm", "content": "How to set up HubSpot CRM...", "platform": "hubspot", } ] mcp = self._make_mcp() search = mcp._tools["search_platform_kb"] result = search(query="CRM setup", platform="hubspot", top_k=5) mock_rag_query.assert_called_once() call_kwargs = mock_rag_query.call_args assert "CRM setup" in str(call_kwargs) assert "hubspot" in result.lower() or "BLOODSTREAM" in result @patch("core.rag_query.rag_context") def test_empty_query_returns_message(self, mock_rag_context): """BB: Empty query string returns a helpful message without calling rag.""" mcp = self._make_mcp() search = mcp._tools["search_platform_kb"] result = search(query="", top_k=5) mock_rag_context.assert_not_called() assert "non-empty" in result.lower() or "provide" in result.lower() @patch("core.rag_query.rag_context") def test_top_k_passed(self, mock_rag_context): """BB: top_k=10 is passed through to rag_context.""" mock_rag_context.return_value = "=== BLOODSTREAM KNOWLEDGE (10 matches) ===" mcp = self._make_mcp() search = mcp._tools["search_platform_kb"] result = search(query="workflow automation", top_k=10) mock_rag_context.assert_called_once_with(question="workflow automation", top_k=10) # ────────────────────────────────────────────────────────────────────────────── # Class TestListPlatformKbs # ────────────────────────────────────────────────────────────────────────────── class TestListPlatformKbs: """BB tests for list_platform_kbs tool.""" def _make_mcp(self): registered: dict[str, Any] = {} class FakeMCP: def tool(self): def decorator(fn): registered[fn.__name__] = fn return fn return decorator @property def _tools(self): return registered mcp = FakeMCP() sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import kb_tools as mod mod.register_kb_tools(mcp) return mcp @patch("core.kb.pg_store.get_connection") @patch("core.kb.pg_store.get_ingestion_history") @patch("core.kb.qdrant_store.get_platform_stats") @patch("core.kb.platform_registry.get_platform") @patch("core.kb.platform_registry.list_platforms") def test_list_shows_platforms( self, mock_list_platforms, mock_get_platform, mock_get_stats, mock_get_history, mock_get_conn, ): """BB: Registered platform names appear in the output table.""" mock_list_platforms.return_value = ["hubspot", "telnyx"] mock_platform = MagicMock() mock_platform.display_name = "HubSpot" mock_get_platform.return_value = mock_platform mock_get_stats.return_value = {"total_vectors": 1500, "platforms": {}} mock_get_history.return_value = [] mock_conn = MagicMock() mock_get_conn.return_value = mock_conn mcp = self._make_mcp() list_tool = mcp._tools["list_platform_kbs"] result = list_tool() assert "hubspot" in result assert "telnyx" in result @patch("core.kb.pg_store.get_connection") @patch("core.kb.pg_store.get_ingestion_history") @patch("core.kb.qdrant_store.get_platform_stats") @patch("core.kb.platform_registry.get_platform") @patch("core.kb.platform_registry.list_platforms") def test_list_shows_stats( self, mock_list_platforms, mock_get_platform, mock_get_stats, mock_get_history, mock_get_conn, ): """BB: Qdrant vector total and PG history dates appear in output.""" mock_list_platforms.return_value = ["telnyx"] mock_platform = MagicMock() mock_platform.display_name = "Telnyx" mock_get_platform.return_value = mock_platform mock_get_stats.return_value = { "total_vectors": 2048, "platforms": {"telnyx": {"count": 2048}}, } mock_get_history.return_value = [ { "completed_at": "2026-02-26 10:00:00.000000", "pages_fetched": 120, "chunks_created": 840, } ] mock_conn = MagicMock() mock_get_conn.return_value = mock_conn mcp = self._make_mcp() list_tool = mcp._tools["list_platform_kbs"] result = list_tool() assert "2048" in result assert "Total vectors" in result or "2048" in result @patch("core.kb.platform_registry.list_platforms") def test_list_empty_graceful(self, mock_list_platforms): """BB: No platforms registered → returns a friendly message.""" mock_list_platforms.return_value = [] mcp = self._make_mcp() list_tool = mcp._tools["list_platform_kbs"] result = list_tool() assert "no" in result.lower() or "registered" in result.lower() # ────────────────────────────────────────────────────────────────────────────── # Class TestIngestPlatformKb # ────────────────────────────────────────────────────────────────────────────── class TestIngestPlatformKb: """BB + WB tests for ingest_platform_kb tool.""" def _make_mcp(self): registered: dict[str, Any] = {} class FakeMCP: def tool(self): def decorator(fn): registered[fn.__name__] = fn return fn return decorator @property def _tools(self): return registered mcp = FakeMCP() sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import kb_tools as mod mod.register_kb_tools(mcp) return mcp def _make_stats(self, **overrides) -> dict: base = { "platform": "telnyx", "customer_id": None, "pages_fetched": 50, "pages_skipped": 5, "chunks_created": 350, "vectors_upserted": 350, "errors": 0, "error_details": [], "duration_seconds": 12.5, "status": "completed", } base.update(overrides) return base @patch("core.kb.orchestrator.ingest_platform", new_callable=AsyncMock) @patch("core.kb.platform_registry.get_platform") def test_trigger_ingestion(self, mock_get_platform, mock_ingest): """BB: Valid platform → ingest_platform called with correct args.""" mock_config = MagicMock() mock_config.display_name = "Telnyx" mock_get_platform.return_value = mock_config mock_ingest.return_value = self._make_stats() mcp = self._make_mcp() ingest = mcp._tools["ingest_platform_kb"] result = ingest(platform="telnyx", max_pages=100) mock_ingest.assert_called_once_with(platform="telnyx", max_pages=100) assert "Telnyx" in result @patch("core.kb.platform_registry.list_platforms") @patch("core.kb.platform_registry.get_platform") def test_invalid_platform_error(self, mock_get_platform, mock_list_platforms): """BB: Unknown platform → helpful error message returned, no crash.""" mock_get_platform.return_value = None mock_list_platforms.return_value = ["hubspot", "telnyx"] mcp = self._make_mcp() ingest = mcp._tools["ingest_platform_kb"] result = ingest(platform="nonexistent_platform", max_pages=100) assert "nonexistent_platform" in result or "Unknown" in result assert "hubspot" in result or "telnyx" in result @patch("core.kb.orchestrator.ingest_platform", new_callable=AsyncMock) @patch("core.kb.platform_registry.get_platform") def test_max_pages_respected(self, mock_get_platform, mock_ingest): """BB: max_pages=50 is passed through to ingest_platform.""" mock_config = MagicMock() mock_config.display_name = "HubSpot" mock_get_platform.return_value = mock_config mock_ingest.return_value = self._make_stats(platform="hubspot") mcp = self._make_mcp() ingest = mcp._tools["ingest_platform_kb"] ingest(platform="hubspot", max_pages=50) mock_ingest.assert_called_once_with(platform="hubspot", max_pages=50) @patch("core.kb.orchestrator.ingest_platform", new_callable=AsyncMock) @patch("core.kb.platform_registry.get_platform") def test_returns_summary(self, mock_get_platform, mock_ingest): """BB: Summary output includes pages, chunks, and vectors counts.""" mock_config = MagicMock() mock_config.display_name = "Telnyx" mock_get_platform.return_value = mock_config mock_ingest.return_value = self._make_stats( pages_fetched=80, chunks_created=560, vectors_upserted=560, ) mcp = self._make_mcp() ingest = mcp._tools["ingest_platform_kb"] result = ingest(platform="telnyx", max_pages=100) assert "80" in result # pages_fetched assert "560" in result # chunks_created / vectors_upserted assert "12.5" in result # duration # ────────────────────────────────────────────────────────────────────────────── # Class TestToolRegistration # ────────────────────────────────────────────────────────────────────────────── class TestToolRegistration: """WB tests for register_kb_tools() itself.""" def _make_fresh_mcp(self): registered: dict[str, Any] = {} class FakeMCP: def tool(self): def decorator(fn): registered[fn.__name__] = fn return fn return decorator @property def _tools(self): return registered return FakeMCP() def test_register_adds_3_tools(self): """WB: register_kb_tools adds exactly 3 named tools to the MCP instance.""" sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import kb_tools as mod mcp = self._make_fresh_mcp() mod.register_kb_tools(mcp) tools = mcp._tools assert "search_platform_kb" in tools assert "list_platform_kbs" in tools assert "ingest_platform_kb" in tools assert len(tools) == 3 @patch("core.rag_query.rag_context") def test_concurrent_search_safe(self, mock_rag_context): """WB: Multiple search_platform_kb calls succeed without interference.""" mock_rag_context.side_effect = [ "=== RESULT A ===", "=== RESULT B ===", "=== RESULT C ===", ] sys.path.insert(0, "/mnt/e/genesis-system/mcp-servers/genesis-core") import kb_tools as mod mcp = self._make_fresh_mcp() mod.register_kb_tools(mcp) search = mcp._tools["search_platform_kb"] results = [search(query=f"query {i}", top_k=3) for i in range(3)] assert all(r.startswith("===") for r in results) assert mock_rag_context.call_count == 3 # VERIFICATION_STAMP # Story: 9.01, 9.02, 9.03 (tests for M9 RAG MCP tools) # Verified By: parallel-builder # Verified At: 2026-02-26 # Tests: 13/13 # Coverage: 100%