leocrm/scripts/test_all_endpoints.py

#!/usr/bin/env python3
"""Test all 556 API endpoints against production."""
import subprocess, json, sys, os, re, time

BASE = 'https://crm.media-on.de'
ORIGIN = 'https://crm.media-on.de'
COOKIE_FILE = '/tmp/admin_test_cookies.txt'
ROUTES_FILE = '/tmp/all_api_routes.json'

# Global state
CSRF = ''
PASS = 0
FAIL = 0
SKIP = 0
BUGS = []
TEST_UUID = '00000000-0000-0000-0000-000000000000'

def login():
    global CSRF
    cmd = [
        'curl', '-s', '-X', 'POST', f'{BASE}/api/v1/auth/login',
        '-H', 'Content-Type: application/json',
        '-H', f'Origin: {ORIGIN}',
        '-d', json.dumps({"email": "admin@media-on.de", "password": "Admin123!"}),
        '-c', COOKIE_FILE
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
    data = json.loads(result.stdout)
    CSRF = data.get('csrf_token', '')
    return CSRF

def replace_path_params(path):
    return re.sub(r'\{[^}]+\}', TEST_UUID, path)

def test_route(method, path, retry=0):
    global PASS, FAIL, SKIP, CSRF
    
    if not path.startswith('/api/'):
        SKIP += 1
        return None
    
    if 'ws' in path.lower() or 'websocket' in path.lower():
        SKIP += 1
        return None
    
    test_path = replace_path_params(path)
    url = f'{BASE}{test_path}'
    
    cmd = ['curl', '-s', '-o', '/dev/null', '-w', '%{http_code}',
           '-X', method, url,
           '-H', f'Origin: {ORIGIN}',
           '-H', f'X-CSRF-Token: {CSRF}',
           '-b', COOKIE_FILE]
    
    if method in ('POST', 'PUT', 'PATCH'):
        cmd.extend(['-H', 'Content-Type: application/json', '-d', '{}'])
    
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        status = result.stdout.strip()
        
        # 429 = Rate limited — wait and retry
        if status == '429' and retry < 3:
            time.sleep(5)
            return test_route(method, path, retry + 1)
        
        # 401 = Session expired — re-login and retry
        if status == '401' and retry < 2:
            time.sleep(1)
            login()
            return test_route(method, path, retry + 1)
        
        # 500 = always a bug
        if status == '500':
            FAIL += 1
            bug = f'{method} {path} → 500'
            BUGS.append(bug)
            print(f'❌ {bug}')
            return status
        
        # 405 = wrong method
        if status == '405':
            FAIL += 1
            bug = f'{method} {path} → 405'
            BUGS.append(bug)
            print(f'❌ {bug}')
            return status
        
        # 403 for admin = bug
        if status == '403':
            FAIL += 1
            bug = f'{method} {path} → 403 (admin denied)'
            BUGS.append(bug)
            print(f'❌ {bug}')
            return status
        
        # Acceptable statuses
        if status in ('200', '201', '204', '400', '404', '422', '401'):
            PASS += 1
            return status
        else:
            FAIL += 1
            bug = f'{method} {path} → {status}'
            BUGS.append(bug)
            print(f'❌ {bug}')
            return status
    except subprocess.TimeoutExpired:
        SKIP += 1
        return 'TIMEOUT'
    except Exception as e:
        SKIP += 1
        return f'ERROR'

def main():
    global PASS, FAIL, SKIP
    
    # Load routes
    with open(ROUTES_FILE) as f:
        routes = json.load(f)
    
    # Login
    csrf = login()
    print(f'Login: CSRF={csrf[:20]}...')
    
    # Verify login
    verify = subprocess.run(
        ['curl', '-s', '-o', '/dev/null', '-w', '%{http_code}',
         f'{BASE}/api/v1/auth/me',
         '-H', f'Origin: {ORIGIN}',
         '-H', f'X-CSRF-Token: {csrf}',
         '-b', COOKIE_FILE],
        capture_output=True, text=True, timeout=10
    )
    print(f'Verify login: {verify.stdout}')
    
    if verify.stdout != '200':
        print('Login failed!')
        sys.exit(1)
    
    # Test all routes
    print(f'\nTesting {len(routes)} routes...')
    for i, route in enumerate(routes):
        method = route['method']
        path = route['path']
        
        if i % 50 == 0:
            print(f'  Progress: {i}/{len(routes)}... (P:{PASS} F:{FAIL} S:{SKIP})')
        
        test_route(method, path)
        time.sleep(0.3)  # Rate limiting
    
    # Results
    print(f'\n=========================================')
    print(f'ALLE 556 API ENDPOINTS GETESTET')
    print(f'=========================================')
    print(f'  Total: {len(routes)}')
    print(f'  Passed: {PASS}')
    print(f'  Failed: {FAIL}')
    print(f'  Skipped: {SKIP}')
    print(f'  Bugs: {len(BUGS)}')
    print(f'=========================================')
    
    # Save bugs
    with open('/tmp/api_test_bugs.txt', 'w') as f:
        for bug in BUGS:
            f.write(bug + '\n')
    
    if BUGS:
        print(f'\nAll bugs:')
        for bug in BUGS:
            print(f'  {bug}')
    else:
        print(f'\n✅ No bugs found!')

if __name__ == '__main__':
    main()