"""Unit & Integration tests for Developer Knowledge MCP Client and Tools.""" import pytest from app.tools.mcp_developer_knowledge import ( DeveloperKnowledgeMCPClient, developerknowledge_answer_query, developerknowledge_get_documents, developerknowledge_search_documents, get_mcp_client, ) from app.adk.tools import mcp_search_tool, mcp_get_tool, mcp_answer_tool def test_mcp_client_search_documents(): """Verify DeveloperKnowledgeMCPClient search_documents method.""" client = get_mcp_client() res = client.search_documents(query="Cloud Run", category="compute") assert res["source"] in ("live_mcp", "offline_fallback") assert "documents" in res or "data" in res def test_mcp_client_get_documents(): """Verify DeveloperKnowledgeMCPClient get_documents method.""" client = get_mcp_client() res = client.get_documents(document_uri="https://cloud.google.com/run/docs/overview") assert res["source"] in ("live_mcp", "offline_fallback") assert "document" in res or "data" in res def test_mcp_client_answer_query(): """Verify DeveloperKnowledgeMCPClient answer_query method.""" client = get_mcp_client() res = client.answer_query(query="Check Cloud Run release status") assert res["source"] in ("live_mcp", "offline_fallback") def test_langchain_mcp_tools(): """Verify LangChain @tool wrappers for Developer Knowledge MCP.""" search_res = developerknowledge_search_documents.invoke({"query": "PubSub", "category": "messaging"}) assert search_res is not None get_res = developerknowledge_get_documents.invoke({"document_uri": "https://cloud.google.com/pubsub/docs"}) assert get_res is not None ans_res = developerknowledge_answer_query.invoke({"query": "What is the release status of Cloud Storage?"}) assert ans_res is not None def test_adk_mcp_function_tools(): """Verify ADK FunctionTools for Developer Knowledge MCP.""" assert mcp_search_tool.name == "developerknowledge_search_documents" assert mcp_get_tool.name == "developerknowledge_get_documents" assert mcp_answer_tool.name == "developerknowledge_answer_query"