import sys
import json
from src.main import get_apps, get_job_summary, get_stage_metrics, fetch_sql_plan, analyze_spark_job
def test_tools():
print("=== Testing MCP Tools ===")
# 1. Get Apps
print("\n[1] Calling get_apps()...")
apps = get_apps()
print(f"Found {len(apps)} applications.")
if not apps:
print("No apps found. Please run a Spark job first.")
return
# Use the first app found
app_id = apps[0]['id']
print(f"Using App ID: {app_id}")
# 2. Get Job Summary
print(f"\n[2] Calling get_job_summary({app_id})...")
jobs = get_job_summary(app_id)
print(f"Found {len(jobs)} jobs.")
if jobs:
print(f"Sample Job: {jobs[0].get('name', 'Unknown')}")
# 3. Get Stage Metrics
print(f"\n[3] Calling get_stage_metrics({app_id})...")
stages = get_stage_metrics(app_id)
print(f"Found {len(stages)} stages.")
# 4. Fetch SQL Plan
print(f"\n[4] Calling fetch_sql_plan({app_id})...")
sql = fetch_sql_plan(app_id)
print(f"Found {len(sql)} SQL executions.")
# 5. Analyze Spark Job
print(f"\n[5] Calling analyze_spark_job({app_id})...")
# We pass None for code_content here just to test the fetching part
report = analyze_spark_job(app_id, code_content=None)
# Print a brief summary of the report
recs = report.get('recommendations', [])
print(f"Analysis complete. Generated {len(recs)} recommendations.")
if recs:
print("Top Recommendation:", recs[0]['issue'], "-", recs[0]['suggestion'])
if __name__ == "__main__":
test_tools()
