API Playground

import requests
import time

class HITLIntegrationTest:
    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
        self.base_url = "https://api.hitl.sh/v1"
        
    def run_complete_test(self):
        """Run complete integration test"""
        print("🧪 Starting HITL.sh Integration Test...")
        
        # Step 1: Create a test loop
        loop = self.create_test_loop()
        if not loop:
            return False
        
        loop_id = loop['data']['loop']['id']
        print(f"✅ Loop created: {loop_id}")
        
        # Step 2: Create a test request
        request = self.create_test_request(loop_id)
        if not request:
            return False
        
        request_id = request['data']['request']['id']
        print(f"✅ Request created: {request_id}")
        
        # Step 3: Monitor request status
        self.monitor_request(request_id)
        
        # Step 4: Cleanup (optional)
        # self.cleanup_test_data(loop_id)
        
        print("🎉 Integration test completed successfully!")
        return True
    
    def create_test_loop(self):
        """Create a test loop"""
        response = requests.post(f"{self.base_url}/loops", headers=self.headers, json={
            "name": f"Integration Test {int(time.time())}",
            "description": "Automated integration test loop",
            "icon": "🧪"
        })
        
        if response.status_code == 201:
            return response.json()
        else:
            print(f"❌ Failed to create loop: {response.text}")
            return None
    
    def create_test_request(self, loop_id):
        """Create a test request"""
        response = requests.post(f"{self.base_url}/requests", headers=self.headers, json={
            "loop_id": loop_id,
            "request_text": "Integration Test: Please confirm this request is working correctly.",
            "response_type": "single_select",
            "response_config": {
                "options": ["Working correctly", "Has issues", "Needs investigation"]
            },
            "priority": "high",
            "processing_type": "time-sensitive"
        })
        
        if response.status_code == 201:
            return response.json()
        else:
            print(f"❌ Failed to create request: {response.text}")
            return None
    
    def monitor_request(self, request_id):
        """Monitor request until completion or timeout"""
        max_checks = 20
        check_interval = 30  # seconds
        
        for i in range(max_checks):
            response = requests.get(
                f"{self.base_url}/requests/{request_id}", 
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            
            if response.status_code == 200:
                request_data = response.json()['data']['request']
                status = request_data['status']
                
                print(f"📊 Request status: {status}")
                
                if status == 'completed':
                    response_data = request_data.get('response_data')
                    print(f"✅ Request completed with response: {response_data}")
                    return True
                elif status in ['cancelled', 'timeout']:
                    print(f"⚠️ Request ended with status: {status}")
                    return False
            
            time.sleep(check_interval)
        
        print("⏰ Monitoring timeout - request may still be pending")
        return False

# Usage
# tester = HITLIntegrationTest('your_api_key_here')
# tester.run_complete_test()

def test_rate_limits(api_key):
    """Test rate limit behavior"""
    headers = {"Authorization": f"Bearer {api_key}"}
    
    # Make rapid requests to trigger rate limiting
    for i in range(105):  # Exceed 100/hour limit
        response = requests.get("https://api.hitl.sh/v1/api/loops", headers=headers)
        
        if response.status_code == 429:
            print(f"🚫 Rate limit hit after {i+1} requests")
            print(f"Response: {response.json()}")
            break
        elif response.status_code != 200:
            print(f"❌ Unexpected error: {response.status_code}")
            break
        else:
            print(f"✅ Request {i+1} successful")
    
    return i

# test_rate_limits('your_api_key_here')

import threading
from flask import Flask, request
import requests

class WebhookTester:
    def __init__(self, webhook_url, webhook_secret):
        self.webhook_url = webhook_url
        self.webhook_secret = webhook_secret
        self.received_events = []
        
    def start_webhook_server(self, port=5000):
        """Start local webhook server for testing"""
        app = Flask(__name__)
        
        @app.route('/webhook', methods=['POST'])
        def handle_webhook():
            # Store received webhook for testing
            self.received_events.append({
                'headers': dict(request.headers),
                'body': request.json,
                'timestamp': time.time()
            })
            return {'status': 'received'}
        
        app.run(port=port, debug=False)
    
    def trigger_webhook_event(self, api_key):
        """Create a request to trigger webhook events"""
        # This would create a request that triggers webhooks
        # when it gets claimed/completed
        pass
    
    def verify_webhook_received(self, event_type, timeout=60):
        """Wait for and verify specific webhook event"""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            for event in self.received_events:
                if event['body']['event'] == event_type:
                    return event
            time.sleep(1)
        
        return None

# Usage:
# tester = WebhookTester('http://localhost:5000/webhook', 'secret')
# threading.Thread(target=tester.start_webhook_server).start()
# event = tester.verify_webhook_received('request.completed', timeout=120)

import concurrent.futures
import time

def load_test_api(api_key, num_requests=50, max_workers=10):
    """Simple load test for API endpoints"""
    
    def make_request(request_num):
        headers = {"Authorization": f"Bearer {api_key}"}
        start_time = time.time()
        
        try:
            response = requests.get("https://api.hitl.sh/v1/api/loops", headers=headers)
            end_time = time.time()
            
            return {
                'request_num': request_num,
                'status_code': response.status_code,
                'response_time': end_time - start_time,
                'success': response.status_code == 200
            }
        except Exception as e:
            return {
                'request_num': request_num,
                'error': str(e),
                'success': False
            }
    
    # Execute load test
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(make_request, i) for i in range(num_requests)]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]
    
    # Analyze results
    successful_requests = [r for r in results if r['success']]
    failed_requests = [r for r in results if not r['success']]
    
    if successful_requests:
        avg_response_time = sum(r['response_time'] for r in successful_requests) / len(successful_requests)
    else:
        avg_response_time = 0
    
    print(f"Load Test Results:")
    print(f"  Total requests: {num_requests}")
    print(f"  Successful: {len(successful_requests)}")
    print(f"  Failed: {len(failed_requests)}")
    print(f"  Success rate: {len(successful_requests)/num_requests*100:.1f}%")
    print(f"  Average response time: {avg_response_time:.3f}s")
    
    return results

# load_test_api('your_api_key_here', num_requests=25, max_workers=5)