test_google_ads_mcp.py•5.92 kB
import asyncio
import json
import os
import sys
from pathlib import Path
# Add the parent directory to Python path for imports
sys.path.insert(0, str(Path(__file__).parent))
# Import your MCP server module
import google_ads_server
def test_format_customer_id():
"""Test the format_customer_id function with various input formats."""
test_cases = [
# Regular ID
("9873186703", "9873186703"),
# ID with dashes
("987-318-6703", "9873186703"),
# ID with quotes
('"9873186703"', "9873186703"),
# ID with escaped quotes
('\"9873186703\"', "9873186703"),
# ID with leading zeros
("0009873186703", "9873186703"),
# Short ID that needs padding
("12345", "0000012345"),
# ID with other non-digit characters
("{9873186703}", "9873186703"),
]
print("\n=== Testing format_customer_id with various formats ===")
for input_id, expected in test_cases:
result = google_ads_server.format_customer_id(input_id)
print(f"Input: {input_id}")
print(f"Result: {result}")
print(f"Expected: {expected}")
print(f"Test {'PASSED' if result == expected else 'FAILED'}")
print("-" * 50)
async def test_mcp_tools():
"""Test Google Ads MCP tools directly."""
# Get a list of available customer IDs first
print("=== Testing list_accounts ===")
accounts_result = await google_ads_server.list_accounts()
print(accounts_result)
# Parse the accounts to extract a customer ID for further tests
customer_id = None
for line in accounts_result.split('\n'):
if line.startswith("Account ID:"):
customer_id = line.replace("Account ID:", "").strip()
break
if not customer_id:
print("No customer IDs found. Cannot continue testing.")
return
print(f"\nUsing customer ID: {customer_id} for testing\n")
# Test campaign performance
print("\n=== Testing get_campaign_performance ===")
campaign_result = await google_ads_server.get_campaign_performance(customer_id, days=90)
print(campaign_result)
# Test ad performance
print("\n=== Testing get_ad_performance ===")
ad_result = await google_ads_server.get_ad_performance(customer_id, days=90)
print(ad_result)
# Test ad creatives
print("\n=== Testing get_ad_creatives ===")
creatives_result = await google_ads_server.get_ad_creatives(customer_id)
print(creatives_result)
# Test custom GAQL query
print("\n=== Testing run_gaql ===")
query = """
SELECT
campaign.id,
campaign.name,
campaign.status
FROM campaign
LIMIT 5
"""
gaql_result = await google_ads_server.run_gaql(customer_id, query, format="json")
print(gaql_result)
async def test_asset_methods():
"""Test Asset-related MCP tools directly."""
# Get a list of available customer IDs first
print("=== Testing Asset Methods ===")
accounts_result = await google_ads_server.list_accounts()
# Parse the accounts to extract a customer ID for further tests
customer_id = None
for line in accounts_result.split('\n'):
if line.startswith("Account ID:"):
customer_id = line.replace("Account ID:", "").strip()
break
if not customer_id:
print("No customer IDs found. Cannot continue testing.")
return
print(f"\nUsing customer ID: {customer_id} for testing asset methods\n")
# Test get_image_assets
print("\n=== Testing get_image_assets ===")
image_assets_result = await google_ads_server.get_image_assets(customer_id, limit=10)
print(image_assets_result)
# Extract an asset ID for further testing if available
asset_id = None
for line in image_assets_result.split('\n'):
if line.startswith("1. Asset ID:"):
asset_id = line.replace("1. Asset ID:", "").strip()
break
# Use a smaller number of days for testing to avoid the INVALID_VALUE_WITH_DURING_OPERATOR error
days_to_test = 30 # Use 30 instead of 90
# Test get_asset_usage if we found an asset ID
if asset_id:
print(f"\n=== Testing get_asset_usage with asset ID: {asset_id} ===")
try:
asset_usage_result = await google_ads_server.get_asset_usage(customer_id, asset_id=asset_id, asset_type="IMAGE")
print(asset_usage_result)
except Exception as e:
print(f"Error in get_asset_usage: {str(e)}")
else:
print("\nNo asset ID found to test get_asset_usage")
# Test analyze_image_assets with a valid date range
print(f"\n=== Testing analyze_image_assets with {days_to_test} days ===")
try:
analyze_result = await google_ads_server.analyze_image_assets(customer_id, days=days_to_test)
print(analyze_result)
except Exception as e:
print(f"Error in analyze_image_assets: {str(e)}")
if __name__ == "__main__":
# Run format_customer_id tests first
# test_format_customer_id()
# Setup environment variables if they're not already set
if not os.environ.get("GOOGLE_ADS_CREDENTIALS_PATH"):
# Set environment variables for testing (comment out if already set in your environment)
os.environ["GOOGLE_ADS_CREDENTIALS_PATH"] = "google_ads_token.json"
os.environ["GOOGLE_ADS_DEVELOPER_TOKEN"] = "YOUR_DEVELOPER_TOKEN" # Replace with placeholder
os.environ["GOOGLE_ADS_CLIENT_ID"] = "YOUR_CLIENT_ID" # Replace with placeholder
os.environ["GOOGLE_ADS_CLIENT_SECRET"] = "YOUR_CLIENT_SECRET" # Replace with placeholder
# Run the MCP tools test (uncomment to run full tests)
# asyncio.run(test_mcp_tools())
# Run the asset methods test (uncomment to run full tests)
asyncio.run(test_asset_methods())
