PYTHON test_connectors2.py

#!/usr/bin/env python3
"""Clean up old test connectors and re-test ServiceTitan."""
import requests as http_req
import json

BASE = 'http://127.0.0.1:5003'

def login():
    s = http_req.Session()
    s.post(f'{BASE}/api/auth/login',
           json={'email': 'grepples15@gmail.com', 'password': 'temp_pw_2026'})
    return s

s = login()

# Delete existing test connectors
print('=== Deleting old connectors ===')
r = s.get(f'{BASE}/api/connectors')
data = r.json()
items = data.get('connectors', []) if isinstance(data, dict) else data
for c in items:
    cid = c.get('id') or c.get('connector_id')
    if cid:
        r = s.delete(f'{BASE}/api/connectors/{cid}')
        print(f'  DELETE {cid[:16]} ({c.get("service")}): {r.status_code} {r.text[:100]}')
print()

# Create ServiceTitan connector
print('=== Creating ServiceTitan connector ===')
payload = {
    "service": "servicetitan",
    "config": {
        "client_id": "test_client_id",
        "client_secret": "test_client_secret",
        "access_token": "test_access_token",
    },
    "name": "Test ServiceTitan"
}
r = s.post(f'{BASE}/api/connectors', json=payload)
print(f'  Status: {r.status_code}')
result = r.json()
print(f'  Response: {json.dumps(result, indent=2)[:400]}')
connector_id = result.get('connector', {}).get('id')
print()

if connector_id:
    # Test connect
    print(f'=== Testing ServiceTitan connect ===')
    r = s.post(f'{BASE}/api/connectors/{connector_id}/connect')
    print(f'  Connect: [{r.status_code}] {r.text[:300]}')
    print()

    # Test status
    print(f'=== Testing ServiceTitan status ===')
    r = s.get(f'{BASE}/api/connectors/{connector_id}/status')
    print(f'  Status: [{r.status_code}] {json.dumps(r.json(), indent=2)[:400]}')
    print()

    # Test sync
    print(f'=== Testing ServiceTitan sync ===')
    r = s.post(f'{BASE}/api/connectors/{connector_id}/sync')
    print(f'  Sync: [{r.status_code}] {json.dumps(r.json(), indent=2)[:400]}')
    print()

    # Test disconnect
    print(f'=== Testing ServiceTitan disconnect ===')
    r = s.post(f'{BASE}/api/connectors/{connector_id}/disconnect')
    print(f'  Disconnect: [{r.status_code}] {r.text[:300]}')
    print()

    # Test delete
    print(f'=== Testing ServiceTitan delete ===')
    r = s.delete(f'{BASE}/api/connectors/{connector_id}')
    print(f'  Delete: [{r.status_code}] {r.text[:200]}')

# Test Facebook Ads creation
print()
print('=== Creating Facebook Ads connector ===')
payload = {
    "service": "facebook_ads",
    "config": {
        "access_token": "EAABtest_fake_token",
        "page_id": "12345",
    },
    "name": "Test Facebook Ads"
}
r = s.post(f'{BASE}/api/connectors', json=payload)
print(f'  Status: {r.status_code}')
result = r.json()
print(f'  Response: {json.dumps(result, indent=2)[:400]}')
fb_id = result.get('connector', {}).get('id')
print()

if fb_id:
    r = s.post(f'{BASE}/api/connectors/{fb_id}/sync')
    print(f'  Sync: [{r.status_code}] {r.text[:300]}')
    r = s.delete(f'{BASE}/api/connectors/{fb_id}')
    print(f'  Delete: [{r.status_code}] {r.text[:100]}')

# Test JobNimbus creation
print()
print('=== Creating JobNimbus connector ===')
payload = {
    "service": "jobnimbus",
    "config": {
        "api_key": "test_api_key",
        "base_url": "https://test.jobnimbus.com/api/v2",
    },
    "name": "Test JobNimbus"
}
r = s.post(f'{BASE}/api/connectors', json=payload)
print(f'  Status: {r.status_code}')
result = r.json()
print(f'  Response: {json.dumps(result, indent=2)[:400]}')
jn_id = result.get('connector', {}).get('id')
if jn_id:
    r = s.delete(f'{BASE}/api/connectors/{jn_id}')
    print(f'  Delete: [{r.status_code}] {r.text[:100]}')

# Test Angi creation
print()
print('=== Creating Angi connector ===')
payload = {
    "service": "angi",
    "config": {
        "api_key": "test_angi_key",
        "webhook_secret": "test_secret",
    },
    "name": "Test Angi"
}
r = s.post(f'{BASE}/api/connectors', json=payload)
print(f'  Status: {r.status_code}')
result = r.json()
print(f'  Response: {json.dumps(result, indent=2)[:400]}')
angi_id = result.get('connector', {}).get('id')
if angi_id:
    # Now test the Angi webhook
    print('  Testing Angi webhook...')
    r = s.post(f'{BASE}/api/connectors/webhooks/angi',
               json={'lead_id': 'test_lead_123', 'data': {'name': 'Test Lead'}},
               headers={'Content-Type': 'application/json'})
    print(f'  Angi webhook: [{r.status_code}] {r.text[:200]}')
    r = s.delete(f'{BASE}/api/connectors/{angi_id}')
    print(f'  Delete: [{r.status_code}] {r.text[:100]}')

# Test Zapier creation
print()
print('=== Creating Zapier connector ===')
payload = {
    "service": "zapier",
    "config": {
        "api_key": "test_zapier_key",
        "webhook_url": "https://hooks.zapier.com/test",
    },
    "name": "Test Zapier"
}
r = s.post(f'{BASE}/api/connectors', json=payload)
print(f'  Status: {r.status_code}')
result = r.json()
print(f'  Response: {json.dumps(result, indent=2)[:400]}')
zap_id = result.get('connector', {}).get('id')
if zap_id:
    print('  Testing Zapier webhook...')
    r = s.post(f'{BASE}/api/connectors/webhooks/zapier',
               json={'zap_id': 'test_zap', 'data': {'field': 'value'}},
               headers={'Content-Type': 'application/json'})
    print(f'  Zapier webhook: [{r.status_code}] {r.text[:200]}')
    r = s.delete(f'{BASE}/api/connectors/{zap_id}')
    print(f'  Delete: [{r.status_code}] {r.text[:100]}')

print()
print('ALL TESTS COMPLETE')