IRIS Legacy

""" Test MCP Integration - Verify agent can call IRIS and Infocert MCP servers This test validates: 1. GenericMCPClient can discover tools from IRIS MCP 2. GenericMCPClient can call IRIS MCP tools (email_search, calendar_list_events) 3. TrustySignAgent initializes with MCP client 4. Agent has correct tools (email, calendar, teams, document, signature) """ import pytest import asyncio import os from unittest.mock import patch, AsyncMock, MagicMock from src.mcp_client import GenericMCPClient, MCPServer, MCPError from src.agents.trusty_sign_agent import TrustySignAgent from src.llm_core.client import IRISClient from src.services.document_service import DocumentService from src.services.credentials_service import CredentialsService class TestGenericMCPClient: """Test GenericMCPClient JSON-RPC 2.0 protocol""" @pytest.mark.asyncio async def test_add_server(self): """Test adding MCP servers""" client = GenericMCPClient() # Add IRIS MCP client.add_server(MCPServer( name="iris", url="http://localhost:8001/mcp/", description="IRIS MCP Server" )) assert "iris" in client.servers assert client.servers["iris"].url == "http://localhost:8001/mcp/" # Add Infocert MCP client.add_server(MCPServer( name="infocert", url="https://mcp.uat.brainaihub.tech/digital-signature/sse", api_key="test-key", description="Infocert MCP" )) assert "infocert" in client.servers assert client.servers["infocert"].api_key == "test-key" @pytest.mark.asyncio async def test_get_server_info(self): """Test getting server information""" client = GenericMCPClient() client.add_server(MCPServer( name="iris", url="http://localhost:8001/mcp/", api_key="test-key" )) info = client.get_server_info() assert "iris" in info assert info["iris"]["url"] == "http://localhost:8001/mcp/" assert info["iris"]["has_api_key"] is True @pytest.mark.asyncio async def test_list_tools_mock(self): """Test listing tools from MCP server (mocked)""" client = GenericMCPClient() client.add_server(MCPServer(name="iris", url="http://localhost:8001/mcp/")) # Mock the _jsonrpc_call method mock_response = { "tools": [ {"name": "email_search", "description": "Search emails"}, {"name": "calendar_list_events", "description": "List calendar events"} ] } with patch.object(client, '_jsonrpc_call', return_value=mock_response): tools = await client.list_tools("iris") assert len(tools) == 2 assert tools[0]["name"] == "email_search" assert tools[1]["name"] == "calendar_list_events" @pytest.mark.asyncio async def test_call_tool_mock(self): """Test calling a tool on MCP server (mocked)""" client = GenericMCPClient() client.add_server(MCPServer(name="iris", url="http://localhost:8001/mcp/")) # Mock successful tool call mock_response = { "content": [ { "type": "text", "text": '{"messages": [{"id": "123", "subject": "Test"}]}' } ] } with patch.object(client, '_jsonrpc_call', return_value=mock_response): result = await client.call_tool( "iris", "email_search", {"user_email": "test@example.com", "query": "test"} ) assert isinstance(result, dict) assert "messages" in result @pytest.mark.asyncio async def test_mcp_error_handling(self): """Test MCP error handling""" client = GenericMCPClient() # Test calling non-existent server with pytest.raises(MCPError) as exc_info: await client.call_tool("nonexistent", "test", {}) assert "not found" in str(exc_info.value).lower() class TestTrustySignAgent: """Test TrustySignAgent MCP integration""" def test_agent_initialization(self): """Test agent initializes with MCP client""" # Mock dependencies iris_client = MagicMock(spec=IRISClient) document_service = MagicMock(spec=DocumentService) credentials_service = MagicMock(spec=CredentialsService) # Create agent agent = TrustySignAgent( iris_client=iris_client, document_service=document_service, credentials_service=credentials_service, user_email="test@example.com", anthropic_api_key="test-key", iris_mcp_url="http://localhost:8001/mcp/" ) # Verify MCP client configured assert agent.mcp_client is not None assert "iris" in agent.mcp_client.servers assert agent.mcp_client.servers["iris"].url == "http://localhost:8001/mcp/" def test_agent_tools_without_infocert(self): """Test agent has correct tools without Infocert MCP""" iris_client = MagicMock(spec=IRISClient) document_service = MagicMock(spec=DocumentService) credentials_service = MagicMock(spec=CredentialsService) agent = TrustySignAgent( iris_client=iris_client, document_service=document_service, credentials_service=credentials_service, user_email="test@example.com", anthropic_api_key="test-key", iris_mcp_url="http://localhost:8001/mcp/" ) tool_names = [tool.name for tool in agent.tools] # Should have email, document, calendar, teams tools assert "email_search" in tool_names assert "email_read_attachment_text" in tool_names assert "email_send_message" in tool_names assert "analyze_document" in tool_names assert "extract_critical_clauses" in tool_names assert "calendar_list_events" in tool_names assert "teams_send_message" in tool_names # Should NOT have signature tools assert "firma_auth_token" not in tool_names assert "firma_sign_document" not in tool_names def test_agent_tools_with_infocert(self): """Test agent has signature tools when Infocert MCP configured""" iris_client = MagicMock(spec=IRISClient) document_service = MagicMock(spec=DocumentService) credentials_service = MagicMock(spec=CredentialsService) agent = TrustySignAgent( iris_client=iris_client, document_service=document_service, credentials_service=credentials_service, user_email="test@example.com", anthropic_api_key="test-key", iris_mcp_url="http://localhost:8001/mcp/", infocert_mcp_url="https://mcp.uat.brainaihub.tech/digital-signature/sse", infocert_mcp_api_key="test-key" ) tool_names = [tool.name for tool in agent.tools] # Should have ALL tools including signature assert "email_search" in tool_names assert "analyze_document" in tool_names assert "firma_auth_token" in tool_names assert "firma_get_certificates" in tool_names assert "firma_request_otp" in tool_names assert "firma_authorize_otp" in tool_names assert "firma_sign_document" in tool_names # Verify Infocert server registered assert "infocert" in agent.mcp_client.servers def test_mcp_call_wrapper(self): """Test _mcp_call sync wrapper""" iris_client = MagicMock(spec=IRISClient) document_service = MagicMock(spec=DocumentService) credentials_service = MagicMock(spec=CredentialsService) agent = TrustySignAgent( iris_client=iris_client, document_service=document_service, credentials_service=credentials_service, user_email="test@example.com", anthropic_api_key="test-key" ) # Mock MCP client call mock_result = {"messages": [{"id": "123"}]} with patch.object(agent.mcp_client, 'call_tool', new_callable=AsyncMock, return_value=mock_result): result = agent._mcp_call( "iris", "email_search", {"user_email": "test@example.com", "query": "test"} ) # Should convert to string assert isinstance(result, str) assert "messages" in result @pytest.mark.skipif( os.getenv("IRIS_MCP_URL") is None, reason="IRIS_MCP_URL not configured - skip integration test" ) class TestRealMCPIntegration: """ Integration tests with real IRIS MCP server Run only if IRIS_MCP_URL is configured """ @pytest.mark.asyncio async def test_list_iris_tools(self): """Test listing tools from real IRIS MCP server""" client = GenericMCPClient() client.add_server(MCPServer( name="iris", url=os.getenv("IRIS_MCP_URL", "http://localhost:8001/mcp/"), api_key=os.getenv("IRIS_MCP_API_KEY") )) tools = await client.list_tools("iris") # Verify expected tools exist tool_names = [tool["name"] for tool in tools] assert "email_search" in tool_names assert "email_list_messages" in tool_names assert "calendar_list_events" in tool_names assert "teams_send_message" in tool_names print(f"\n✅ Found {len(tools)} tools from IRIS MCP") for tool in tools[:5]: print(f" - {tool['name']}: {tool.get('description', '')[:60]}") if __name__ == "__main__": # Run quick test print("🧪 Testing MCP Integration\n") # Test 1: GenericMCPClient initialization client = GenericMCPClient() client.add_server(MCPServer(name="iris", url="http://localhost:8001/mcp/")) print("✅ GenericMCPClient created") print(f" Servers: {list(client.servers.keys())}") # Test 2: Agent initialization from unittest.mock import MagicMock agent = TrustySignAgent( iris_client=MagicMock(), document_service=MagicMock(), credentials_service=MagicMock(), user_email="test@example.com", anthropic_api_key="test-key" ) print(f"\n✅ TrustySignAgent created") print(f" Tools: {len(agent.tools)}") print(f" MCP Servers: {list(agent.mcp_client.servers.keys())}") # Test 3: Agent with Infocert agent_full = TrustySignAgent( iris_client=MagicMock(), document_service=MagicMock(), credentials_service=MagicMock(), user_email="test@example.com", anthropic_api_key="test-key", infocert_mcp_url="https://mcp.uat.brainaihub.tech/digital-signature/sse", infocert_mcp_api_key="test-key" ) print(f"\n✅ TrustySignAgent with Infocert created") print(f" Tools: {len(agent_full.tools)} (should be {len(agent.tools) + 5})") print(f" MCP Servers: {list(agent_full.mcp_client.servers.keys())}") tool_names = [t.name for t in agent_full.tools] print(f"\n📋 All tools:") for name in sorted(tool_names): emoji = "🔐" if name.startswith("firma_") else "📧" if "email" in name else "📊" print(f" {emoji} {name}") print("\n✅ All MCP integration tests passed!")