LinkedIn MCP Server

#!/usr/bin/env python3 """ LinkedIn MCP Client Test Suite Comprehensive tests for LinkedIn MCP integration """ import requests import json import time import os from typing import Dict, Any, Optional from dotenv import load_dotenv # Load environment variables load_dotenv() # Configuration API_BASE_URL = os.getenv("API_BASE", "http://localhost:8002") MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8888/mcp") LINKEDIN_ACCESS_TOKEN = os.getenv("LINKEDIN_ACCESS_TOKEN") class LinkedInMCPTester: def __init__(self, api_base: str, access_token: Optional[str] = None): self.api_base = api_base.rstrip('/') self.access_token = access_token self.session = requests.Session() self.test_results = [] def log_test(self, test_name: str, success: bool, details: str = "", response_data: Any = None): """Log test results""" result = { "test": test_name, "success": success, "details": details, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } if response_data and isinstance(response_data, dict): result["response_preview"] = str(response_data)[:200] + "..." if len(str(response_data)) > 200 else str(response_data) self.test_results.append(result) status = "✅ PASS" if success else "❌ FAIL" print(f"{status} {test_name}") if details: print(f" {details}") if not success and response_data: print(f" Response: {response_data}") print() def make_request(self, method: str, endpoint: str, data: Dict[str, Any] = None, headers: Dict[str, str] = None) -> tuple: """Make HTTP request and return success status and response""" try: url = f"{self.api_base}{endpoint}" # Add access token to headers if available request_headers = headers or {} if self.access_token and "access_token" not in request_headers: request_headers["access_token"] = self.access_token if method.upper() == "GET": response = self.session.get(url, headers=request_headers, timeout=30) elif method.upper() == "POST": response = self.session.post(url, json=data, headers=request_headers, timeout=30) else: raise ValueError(f"Unsupported method: {method}") response.raise_for_status() return True, response.json() except requests.exceptions.RequestException as e: return False, {"error": str(e)} except json.JSONDecodeError as e: return False, {"error": f"JSON decode error: {str(e)}", "response_text": response.text[:500]} except Exception as e: return False, {"error": f"Unexpected error: {str(e)}"} def test_health_check(self): """Test service health""" success, response = self.make_request("GET", "/health") if success: if response.get("status") == "healthy": self.log_test("Health Check", True, "Service is healthy and MCP server is connected", response) else: self.log_test("Health Check", False, "Service reports unhealthy status", response) else: self.log_test("Health Check", False, "Failed to connect to service", response) def test_server_info(self): """Test MCP server information""" success, response = self.make_request("GET", "/server/info") if success and "server" in response: server_name = response["server"].get("name", "unknown") self.log_test("Server Info", True, f"Connected to {server_name}", response) else: self.log_test("Server Info", False, "Failed to get server information", response) def test_list_tools(self): """Test listing available tools""" success, response = self.make_request("GET", "/tools") if success and "tools" in response: tool_count = len(response["tools"]) tool_names = [tool["name"] for tool in response["tools"]] self.log_test("List Tools", True, f"Found {tool_count} tools: {', '.join(tool_names)}") else: self.log_test("List Tools", False, "Failed to list tools", response) def test_list_resources(self): """Test listing available resources""" success, response = self.make_request("GET", "/resources") if success and "resources" in response: resource_count = len(response["resources"]) resource_names = [resource["name"] for resource in response["resources"]] self.log_test("List Resources", True, f"Found {resource_count} resources: {', '.join(resource_names)}") else: self.log_test("List Resources", False, "Failed to list resources", response) def test_get_resource(self): """Test getting a specific resource""" success, response = self.make_request("GET", "/resources/linkedin-api-guide") if success and "contents" in response: content_length = len(response["contents"][0]["text"]) if response["contents"] else 0 self.log_test("Get Resource", True, f"Retrieved resource with {content_length} characters") else: self.log_test("Get Resource", False, "Failed to get resource", response) def test_profile_no_auth(self): """Test profile endpoint without authentication (should fail gracefully)""" # Temporarily remove access token old_token = self.access_token self.access_token = None success, response = self.make_request("GET", "/profile") # Restore token self.access_token = old_token if not success and "access token" in str(response).lower(): self.log_test("Profile No Auth", True, "Correctly rejected request without access token") else: self.log_test("Profile No Auth", False, "Should have rejected request without access token", response) def test_profile_with_mock_auth(self): """Test profile endpoint with mock authentication""" if not self.access_token: # Use a mock token for testing the API structure test_data = { "access_token": "mock_token_for_testing" } success, response = self.make_request("POST", "/profile", test_data) # We expect this to fail with LinkedIn API error, but it should reach the MCP server if not success and ("linkedin" in str(response).lower() or "401" in str(response) or "token" in str(response).lower()): self.log_test("Profile Mock Auth", True, "Request properly formatted and sent to LinkedIn API") else: self.log_test("Profile Mock Auth", False, "Unexpected response format", response) else: # Real token provided, try actual request success, response = self.make_request("GET", "/profile") if success: self.log_test("Profile Real Auth", True, "Successfully retrieved profile", response) else: self.log_test("Profile Real Auth", False, "Failed with real token", response) def test_post_creation_mock(self): """Test post creation with mock data""" test_data = { "content": "Test post from LinkedIn MCP Client - Hello LinkedIn! 🚀", "visibility": "PUBLIC", "access_token": "mock_token_for_testing" } success, response = self.make_request("POST", "/posts", test_data) # Expect this to fail with auth error, but shows the endpoint works if not success and ("linkedin" in str(response).lower() or "token" in str(response).lower()): self.log_test("Post Creation Mock", True, "Post creation endpoint properly structured") else: self.log_test("Post Creation Mock", False, "Unexpected response", response) def test_company_search_mock(self): """Test company search with mock data""" test_data = { "keywords": "Microsoft", "count": 5, "access_token": "mock_token_for_testing" } success, response = self.make_request("POST", "/companies/search", test_data) if not success and ("linkedin" in str(response).lower() or "token" in str(response).lower()): self.log_test("Company Search Mock", True, "Company search endpoint properly structured") else: self.log_test("Company Search Mock", False, "Unexpected response", response) def test_network_analysis_mock(self): """Test network analysis with mock data""" test_data = { "analysis_type": "connections", "access_token": "mock_token_for_testing" } success, response = self.make_request("POST", "/analytics/network", test_data) if not success and ("linkedin" in str(response).lower() or "token" in str(response).lower()): self.log_test("Network Analysis Mock", True, "Network analysis endpoint properly structured") else: self.log_test("Network Analysis Mock", False, "Unexpected response", response) def test_endpoint_coverage(self): """Test that all major endpoints are accessible""" endpoints_to_test = [ ("GET", "/"), ("GET", "/docs", True), # Third parameter indicates we don't expect JSON ("GET", "/redoc", True) ] for method, endpoint, *optional_flags in endpoints_to_test: expect_json = not (optional_flags and optional_flags[0]) try: url = f"{self.api_base}{endpoint}" response = self.session.get(url, timeout=10) if response.status_code == 200: if expect_json: try: response.json() self.log_test(f"Endpoint {endpoint}", True, "Accessible and returns valid JSON") except: self.log_test(f"Endpoint {endpoint}", False, "Accessible but invalid JSON") else: self.log_test(f"Endpoint {endpoint}", True, "Accessible") else: self.log_test(f"Endpoint {endpoint}", False, f"HTTP {response.status_code}") except Exception as e: self.log_test(f"Endpoint {endpoint}", False, f"Connection error: {str(e)}") def run_all_tests(self): """Run all tests""" print("🚀 Starting LinkedIn MCP Client Test Suite") print("=" * 50) print(f"API Base URL: {self.api_base}") print(f"MCP Server URL: {MCP_SERVER_URL}") print(f"Access Token: {'Provided' if self.access_token else 'Not provided (using mock)'}") print("=" * 50) print() # Core functionality tests self.test_health_check() self.test_server_info() self.test_list_tools() self.test_list_resources() self.test_get_resource() # Authentication tests self.test_profile_no_auth() self.test_profile_with_mock_auth() # LinkedIn API endpoint tests self.test_post_creation_mock() self.test_company_search_mock() self.test_network_analysis_mock() # Endpoint coverage tests self.test_endpoint_coverage() # Summary print("\n" + "=" * 50) print("📊 TEST SUMMARY") print("=" * 50) total_tests = len(self.test_results) passed_tests = len([r for r in self.test_results if r["success"]]) failed_tests = total_tests - passed_tests print(f"Total Tests: {total_tests}") print(f"Passed: {passed_tests} ✅") print(f"Failed: {failed_tests} ❌") print(f"Success Rate: {(passed_tests/total_tests)*100:.1f}%") if failed_tests > 0: print("\n🔍 Failed Tests:") for result in self.test_results: if not result["success"]: print(f" - {result['test']}: {result['details']}") print("\n💡 Next Steps:") if not self.access_token: print(" - Set LINKEDIN_ACCESS_TOKEN environment variable for full testing") print(" - Check /docs endpoint for full API documentation") print(" - Use /health endpoint to monitor service status") return passed_tests == total_tests def main(): """Main test function""" tester = LinkedInMCPTester(API_BASE_URL, LINKEDIN_ACCESS_TOKEN) try: success = tester.run_all_tests() exit_code = 0 if success else 1 print(f"\n🏁 Test suite completed with exit code: {exit_code}") return exit_code except KeyboardInterrupt: print("\n⚠️ Tests interrupted by user") return 130 except Exception as e: print(f"\n💥 Test suite crashed: {str(e)}") return 1 if __name__ == "__main__": exit(main())