tests/unit/test_contacts_api.py

#!/usr/bin/env python3
"""
Test script for customer contacts API endpoints
"""
import requests
import json
import sys

# Configuration
BASE_URL = "http://localhost:8000/api/v1/customers"

def test_list_contacts_by_customer(customer_id):
    """Test listing contacts by customer using the new API"""
    print(f"🔍 Testing: List contacts by customer {customer_id}")
    try:
        response = requests.get(f"http://localhost:8000/api/v1/contacts/?customer_id={customer_id}")
        print(f"   Status: {response.status_code}")
        if response.status_code == 200:
            data = response.json()
            print(f"   Found {len(data.get('items', []))} contacts")
            print(f"   Total: {data.get('total', 0)}")
            if data.get('items'):
                first_contact = data['items'][0]
                print(f"   First contact: {first_contact.get('first_name', '')} {first_contact.get('last_name', '')} (ID: {first_contact.get('contact_id')})")
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")

def test_get_contact_by_id(contact_id):
    """Test getting a specific contact by ID"""
    print(f"🔍 Testing: Get contact by ID {contact_id}")
    try:
        response = requests.get(f"{BASE_URL}/contacts/{contact_id}")
        print(f"   Status: {response.status_code}")
        if response.status_code == 200:
            contact = response.json()
            print(f"   Contact: {contact.get('first_name', '')} {contact.get('last_name', '')}")
            print(f"   Customer ID: {contact.get('customer_id')}")
            print(f"   Email: {contact.get('email_address', 'N/A')}")
            print(f"   Phone: {contact.get('work_phone', 'N/A')}")
            return contact
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")
    return None

def test_list_customer_contacts_legacy(customer_id):
    """Test listing contacts for a specific customer using the legacy customer-based endpoint"""
    print(f"🔍 Testing: List contacts for customer {customer_id} (legacy endpoint)")
    try:
        response = requests.get(f"{BASE_URL}/{customer_id}/contacts")
        print(f"   Status: {response.status_code}")
        if response.status_code == 200:
            data = response.json()
            print(f"   Found {len(data.get('items', []))} contacts for customer {customer_id}")
            print(f"   Total: {data.get('total', 0)}")
            if data.get('items'):
                first_contact = data['items'][0]
                print(f"   First contact: {first_contact.get('first_name', '')} {first_contact.get('last_name', '')} (ID: {first_contact.get('contact_id')})")
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")

def test_create_contact(customer_id):
    """Test creating a new contact"""
    print(f"🔍 Testing: Create contact for customer {customer_id}")
    try:
        contact_data = {
            "first_name": "Test",
            "last_name": "Contact",
            "title": "Test Manager",
            "email_address": "test.contact@example.com",
            "work_phone": "555-0123",
            "enabled": True
        }
        
        response = requests.post(
            f"{BASE_URL}/{customer_id}/contacts",
            json=contact_data
        )
        print(f"   Status: {response.status_code}")
        if response.status_code == 201:
            contact = response.json()
            print(f"   Created contact: {contact.get('first_name')} {contact.get('last_name')} (ID: {contact.get('contact_id')})")
            return contact.get('contact_id')
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")
    return None

def test_update_contact(customer_id, contact_id):
    """Test updating a contact"""
    print(f"🔍 Testing: Update contact {contact_id} for customer {customer_id}")
    try:
        contact_data = {
            "first_name": "Updated Test",
            "last_name": "Contact",
            "title": "Senior Test Manager",
            "email_address": "updated.test.contact@example.com",
            "work_phone": "555-0456",
            "enabled": True
        }
        
        response = requests.put(
            f"{BASE_URL}/{customer_id}/contacts/{contact_id}",
            json=contact_data
        )
        print(f"   Status: {response.status_code}")
        if response.status_code == 200:
            contact = response.json()
            print(f"   Updated contact: {contact.get('first_name')} {contact.get('last_name')}")
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")

def test_get_customer_contact(customer_id, contact_id):
    """Test getting a specific contact for a customer"""
    print(f"🔍 Testing: Get contact {contact_id} for customer {customer_id}")
    try:
        response = requests.get(f"{BASE_URL}/{customer_id}/contacts/{contact_id}")
        print(f"   Status: {response.status_code}")
        if response.status_code == 200:
            contact = response.json()
            print(f"   Contact: {contact.get('first_name', '')} {contact.get('last_name', '')}")
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")

def test_delete_contact(customer_id, contact_id):
    """Test deleting a contact"""
    print(f"🔍 Testing: Delete contact {contact_id} for customer {customer_id}")
    try:
        response = requests.delete(f"{BASE_URL}/{customer_id}/contacts/{contact_id}")
        print(f"   Status: {response.status_code}")
        if response.status_code == 204:
            print("   ✅ Contact deleted successfully")
        else:
            print(f"   Error: {response.text}")
    except Exception as e:
        print(f"   Exception: {e}")

def main():
    print("🚀 Testing Customer Contacts API")
    print("=" * 50)
    
    # Test the new contacts API with customer_id parameter
    test_customer_id = 198
    test_list_contacts_by_customer(test_customer_id)
    print()
    
    # Get a sample contact to work with using the legacy endpoint
    print("🔍 Testing: Get sample contact from legacy endpoint")
    sample_contact = None
    try:
        response = requests.get(f"{BASE_URL}/{test_customer_id}/contacts?page=1&page_size=1")
        if response.status_code == 200:
            data = response.json()
            if data.get('items'):
                sample_contact = data['items'][0]
                contact_id = sample_contact.get('contact_id')
                customer_id = sample_contact.get('customer_id')
                print(f"   Using sample contact {contact_id} for customer {customer_id}")
            else:
                print("   No contacts found in database")
                # Let's try creating one for testing
                sample_contact = {"customer_id": test_customer_id}
                contact_id = None
                customer_id = test_customer_id
    except Exception as e:
        print(f"   Exception: {e}")
        # Set fallback values
        sample_contact = {"customer_id": test_customer_id}
        contact_id = None
        customer_id = test_customer_id
    
    print()
    
    # Test individual endpoints if we have a contact
    if contact_id:
        test_get_contact_by_id(contact_id)
        print()
    
    test_list_customer_contacts_legacy(customer_id)
    print()
    
    if contact_id:
        test_get_customer_contact(customer_id, contact_id)
        print()
    
    # Test CRUD operations with a test customer (198 from earlier)
    print(f"🔧 Testing CRUD operations with customer {test_customer_id}")
    print()
    
    # Create a test contact
    new_contact_id = test_create_contact(test_customer_id)
    print()
    
    if new_contact_id:
        # Update the test contact
        test_update_contact(test_customer_id, new_contact_id)
        print()
        
        # Get the updated contact
        test_get_customer_contact(test_customer_id, new_contact_id)
        print()
        
        # Delete the test contact
        test_delete_contact(test_customer_id, new_contact_id)
        print()
    
    print("✅ Testing completed!")

if __name__ == "__main__":
    main()