test_serverless_dlt.py•2.1 kB
#!/usr/bin/env python3
"""
Test script to verify serverless DLT pipeline creation works without cluster permissions.
"""
import os
import sys
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def test_serverless_dlt_pipeline():
"""Test creating and running a serverless DLT pipeline"""
# Import the MCP function
from main import mcp_create_and_run_serverless_dlt_pipeline
# Test parameters
notebook_path = "/Users/stephen.hsu@databricks.com/claude_generated_folder/sample_dlt_pipeline.py"
pipeline_name = "Test Serverless DLT Pipeline"
username = "stephen.hsu@databricks.com"
print("Testing serverless DLT pipeline creation...")
print(f"Notebook path: {notebook_path}")
print(f"Pipeline name: {pipeline_name}")
print(f"Username: {username}")
print("-" * 50)
try:
# Call the MCP function
result = mcp_create_and_run_serverless_dlt_pipeline(
notebook_path=notebook_path,
pipeline_name=pipeline_name,
username=username
)
print("Result:")
print(f"Status: {result.get('status')}")
if result.get('status') == 'success':
print(f"✅ DLT Pipeline created successfully!")
print(f"Execution ID: {result.get('execution_id')}")
print(f"Pipeline ID: {result.get('pipeline_id')}")
print(f"Update ID: {result.get('update_id')}")
print(f"Pipeline Name: {result.get('pipeline_name')}")
print(f"Pipeline URL: {result.get('pipeline_url')}")
print(f"Update URL: {result.get('update_url')}")
print(f"Message: {result.get('message')}")
print(f"Validation: {result.get('validation')}")
else:
print(f"❌ DLT Pipeline 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_dlt_pipeline()