test_serverless_job.py•1.8 kB
#!/usr/bin/env python3
"""
Test script to verify serverless job creation works without cluster permissions.
"""
import os
import sys
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def test_serverless_job_creation():
"""Test creating and running a job with serverless compute"""
# Import the MCP function
from main import mcp_create_and_run_job_for_notebook
# Test parameters
notebook_path = "/Users/stephen.hsu@databricks.com/claude_generated_folder/sample_claude_upload.py"
job_name = "Test Serverless Job"
username = "stephen.hsu@databricks.com"
print("Testing serverless job creation...")
print(f"Notebook path: {notebook_path}")
print(f"Job name: {job_name}")
print(f"Username: {username}")
print("-" * 50)
try:
# Call the MCP function
result = mcp_create_and_run_job_for_notebook(
notebook_path=notebook_path,
job_name=job_name,
username=username
)
print("Result:")
print(f"Status: {result.get('status')}")
if result.get('status') == 'success':
print(f"✅ Job created successfully!")
print(f"Job ID: {result.get('job_id')}")
print(f"Run ID: {result.get('run_id')}")
print(f"Job URL: {result.get('job_url')}")
print(f"Run URL: {result.get('run_url')}")
print(f"Message: {result.get('message')}")
else:
print(f"❌ Job creation failed:")
print(f"Error: {result.get('detail')}")
except Exception as e:
print(f"❌ Exception occurred: {str(e)}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
test_serverless_job_creation()