test_tools.py

#!/usr/bin/env python3
"""
Test script for Unipile LinkedIn MCP tools.
Tests each tool individually to verify API connectivity and response formats.
"""

import asyncio
import json
import sys
from dotenv import load_dotenv

load_dotenv()

# Import all tools from the MCP server
from unipile_linkedin import (
    list_accounts,
    get_my_profile,
    search_people,
    search_people_sales_nav,
    search_companies,
    search_posts,
    get_search_params,
    get_profile,
    get_company_profile,
    send_invitation,
    list_invitations_sent,
    list_invitations_received,
    list_relations,
    list_chats,
    get_chat_messages,
    start_chat,
    send_message,
    send_inmail,
    get_inmail_credits,
    cancel_invitation,
    accept_invitation,
    decline_invitation,
)


def print_result(name: str, result: dict, truncate: bool = True):
    """Pretty print a test result"""
    print(f"\n{'='*60}")
    print(f"TEST: {name}")
    print(f"{'='*60}")

    if "error" in result:
        print(f"❌ ERROR: {result.get('error', 'Unknown error')}")
        return False
    else:
        output = json.dumps(result, indent=2, default=str)
        if truncate and len(output) > 2000:
            output = output[:2000] + "\n... (truncated)"
        print(f"✅ SUCCESS")
        print(output)
        return True


async def test_accounts():
    """Test account-related tools"""
    print("\n" + "="*60)
    print("TESTING: ACCOUNT TOOLS")
    print("="*60)

    # Test list_accounts
    result = await list_accounts()
    success = print_result("list_accounts", result)

    # Test get_my_profile
    result = await get_my_profile()
    success = print_result("get_my_profile", result) and success

    return success


async def test_search():
    """Test search-related tools"""
    print("\n" + "="*60)
    print("TESTING: SEARCH TOOLS")
    print("="*60)

    success = True

    # Test get_search_params for locations (note: param_type must be uppercase)
    result = await get_search_params(param_type="LOCATION", query="San Francisco")
    success = print_result("get_search_params (LOCATION)", result) and success

    # Test get_search_params for industries
    result = await get_search_params(param_type="INDUSTRY", query="Software")
    success = print_result("get_search_params (INDUSTRY)", result) and success

    # Test search_people (Classic)
    result = await search_people(keywords="software engineer", limit=5)
    success = print_result("search_people (Classic)", result) and success

    # Save a provider_id for later tests if we got results
    # The provider_id from search has a prefix like "search_people_profile:"
    # We need to extract the actual ID for profile lookup
    provider_id = None
    public_identifier = None
    if "items" in result and len(result["items"]) > 0:
        item = result["items"][0]
        provider_id = item.get("provider_id")
        public_identifier = item.get("public_identifier")
        # Extract clean provider_id if it has a prefix
        if provider_id and ":" in provider_id:
            provider_id = provider_id.split(":")[-1]
        print(f"\n📝 Got provider_id: {provider_id}")
        print(f"📝 Got public_identifier: {public_identifier}")

    # Test search_people_sales_nav
    result = await search_people_sales_nav(keywords="CTO", limit=5)
    success = print_result("search_people_sales_nav", result) and success

    # Test search_companies
    result = await search_companies(keywords="AI startup", limit=5)
    success = print_result("search_companies", result) and success

    # Test search_posts
    result = await search_posts(keywords="artificial intelligence", limit=5)
    success = print_result("search_posts", result) and success

    return success, provider_id, public_identifier


async def test_profiles(provider_id: str = None, public_identifier: str = None):
    """Test profile-related tools"""
    print("\n" + "="*60)
    print("TESTING: PROFILE TOOLS")
    print("="*60)

    success = True

    # Test get_profile if we have a provider_id or public_identifier
    if provider_id:
        result = await get_profile(provider_id=provider_id)
        success = print_result("get_profile (provider_id)", result) and success
    elif public_identifier:
        result = await get_profile(provider_id=public_identifier)
        success = print_result("get_profile (public_identifier)", result) and success
    else:
        print("⚠️  Skipping get_profile - no provider_id or public_identifier available")

    # Test get_company_profile - try with numeric ID first since name might not work
    # LinkedIn's company ID is 1337
    result = await get_company_profile(company_id="1337")
    success = print_result("get_company_profile (LinkedIn ID: 1337)", result) and success

    return success


async def test_connections():
    """Test connection-related tools (read-only)"""
    print("\n" + "="*60)
    print("TESTING: CONNECTION TOOLS (READ-ONLY)")
    print("="*60)

    success = True

    # Test list_invitations_sent
    result = await list_invitations_sent(limit=5)
    success = print_result("list_invitations_sent", result) and success

    # Test list_invitations_received
    result = await list_invitations_received(limit=5)
    success = print_result("list_invitations_received", result) and success

    # Test list_relations
    result = await list_relations(limit=5)
    success = print_result("list_relations", result) and success

    # Note: Not testing send_invitation, accept/decline/cancel as they modify state
    print("\n⚠️  Skipping send_invitation, accept/decline/cancel (state-modifying)")

    return success


async def test_messaging():
    """Test messaging-related tools (read-only)"""
    print("\n" + "="*60)
    print("TESTING: MESSAGING TOOLS (READ-ONLY)")
    print("="*60)

    success = True

    # Test list_chats
    result = await list_chats(limit=5)
    success = print_result("list_chats", result) and success

    # Get a chat_id for message testing
    chat_id = None
    if "items" in result and len(result["items"]) > 0:
        chat_id = result["items"][0].get("id")

    # Test get_chat_messages if we have a chat
    if chat_id:
        result = await get_chat_messages(chat_id=chat_id, limit=5)
        success = print_result("get_chat_messages", result) and success
    else:
        print("⚠️  Skipping get_chat_messages - no chat_id available")

    # Note: Not testing send_message, start_chat as they modify state
    print("\n⚠️  Skipping send_message, start_chat (state-modifying)")

    return success


async def test_inmail():
    """Test InMail-related tools (read-only)"""
    print("\n" + "="*60)
    print("TESTING: INMAIL TOOLS")
    print("="*60)

    success = True

    # Test get_inmail_credits
    result = await get_inmail_credits()
    success = print_result("get_inmail_credits", result) and success

    # Note: Not testing send_inmail as it uses credits and modifies state
    print("\n⚠️  Skipping send_inmail (uses credits, state-modifying)")

    return success


async def main():
    """Run all tests"""
    print("\n" + "#"*60)
    print("# UNIPILE LINKEDIN MCP - TOOL TESTS")
    print("#"*60)

    all_success = True

    # Test accounts
    success = await test_accounts()
    all_success = all_success and success

    # Test search (and get a provider_id for profile tests)
    success, provider_id, public_identifier = await test_search()
    all_success = all_success and success

    # Test profiles
    success = await test_profiles(provider_id, public_identifier)
    all_success = all_success and success

    # Test connections (read-only)
    success = await test_connections()
    all_success = all_success and success

    # Test messaging (read-only)
    success = await test_messaging()
    all_success = all_success and success

    # Test InMail
    success = await test_inmail()
    all_success = all_success and success

    # Summary
    print("\n" + "#"*60)
    print("# TEST SUMMARY")
    print("#"*60)

    if all_success:
        print("\n✅ ALL TESTS PASSED!")
        return 0
    else:
        print("\n❌ SOME TESTS FAILED - check output above")
        return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))