test_api_examples.py

"""Test the API with real examples to verify it's working."""

from __future__ import annotations

import json
import sys
import time
from datetime import datetime

import requests

BASE_URL = "https://LogicGoInfotechSpaces-duplicate-transaction-detection.hf.space"


def print_section(title: str):
    """Print a formatted section header."""
    print("\n" + "=" * 70)
    print(f"  {title}")
    print("=" * 70)


def test_health_endpoint():
    """Test the health endpoint."""
    print_section("TEST 1: Health Check Endpoint")
    
    try:
        response = requests.get(f"{BASE_URL}/health", timeout=10)
        print(f"URL: {BASE_URL}/health")
        print(f"Status Code: {response.status_code}")
        print(f"Response: {json.dumps(response.json(), indent=2)}")
        
        if response.status_code == 200:
            print("\n[SUCCESS] Health endpoint is working!")
            return True
        else:
            print(f"\n[ERROR] Unexpected status code: {response.status_code}")
            return False
    except Exception as e:
        print(f"\n[ERROR] Health check failed: {e}")
        return False


def test_suggestions_endpoint():
    """Test the suggestions endpoint with different parameters."""
    print_section("TEST 2: Suggestions Endpoint")
    
    # Test 1: Default limit
    print("\n--- Test 2.1: Get suggestions (default limit) ---")
    try:
        response = requests.get(f"{BASE_URL}/suggestions", timeout=60)
        print(f"URL: {BASE_URL}/suggestions")
        print(f"Status Code: {response.status_code}")
        
        if response.status_code == 200:
            suggestions = response.json()
            print(f"Total suggestions returned: {len(suggestions)}")
            
            if suggestions:
                print("\nFirst suggestion example:")
                first = suggestions[0]
                print(f"  - ID: {first.get('_id')}")
                print(f"  - Candidate IDs: {first.get('candidate_ids', [])[:3]}")
                print(f"  - Message: {first.get('message')}")
                print(f"  - Amount Delta: {first.get('details', {}).get('amount_delta_pct', 0):.2f}%")
                print(f"  - Time Delta: {first.get('details', {}).get('time_delta_minutes', 0):.2f} minutes")
                print(f"  - Merchant Match: {first.get('details', {}).get('merchant_match_rule')}")
                print(f"  - Generated At: {first.get('audit', {}).get('generated_at')}")
                print(f"  - Status: {first.get('status')}")
            else:
                print("  No suggestions found (scheduler may not have run yet)")
            
            print("\n[SUCCESS] Suggestions endpoint is working!")
            return True
        else:
            print(f"\n[ERROR] Unexpected status code: {response.status_code}")
            print(f"Response: {response.text[:200]}")
            return False
    except Exception as e:
        print(f"\n[ERROR] Suggestions request failed: {e}")
        return False


def test_suggestions_with_limit():
    """Test suggestions endpoint with custom limit."""
    print_section("TEST 3: Suggestions with Custom Limit")
    
    print("\n--- Test 3.1: Get 5 suggestions ---")
    try:
        response = requests.get(f"{BASE_URL}/suggestions", params={"limit": 5}, timeout=60)
        print(f"URL: {BASE_URL}/suggestions?limit=5")
        print(f"Status Code: {response.status_code}")
        
        if response.status_code == 200:
            suggestions = response.json()
            print(f"Suggestions returned: {len(suggestions)} (requested: 5)")
            
            if suggestions:
                print("\nSample suggestions summary:")
                for i, sug in enumerate(suggestions[:3], 1):
                    candidates = sug.get('candidate_ids', [])
                    print(f"  {i}. {len(candidates)} candidates, "
                          f"Amount delta: {sug.get('details', {}).get('amount_delta_pct', 0):.2f}%, "
                          f"Time delta: {sug.get('details', {}).get('time_delta_minutes', 0):.2f} min")
            
            print("\n[SUCCESS] Custom limit parameter working!")
            return True
        else:
            print(f"\n[ERROR] Unexpected status code: {response.status_code}")
            return False
    except Exception as e:
        print(f"\n[ERROR] Request failed: {e}")
        return False


def test_scheduler_activity():
    """Check if scheduler is generating new suggestions."""
    print_section("TEST 4: Scheduler Activity Check")
    
    print("\nChecking if scheduler is actively generating suggestions...")
    print("(This checks timestamps to see if suggestions are recent)")
    
    try:
        response = requests.get(f"{BASE_URL}/suggestions", params={"limit": 10}, timeout=60)
        if response.status_code == 200:
            suggestions = response.json()
            
            if suggestions:
                # Get the most recent suggestion
                most_recent = suggestions[0]  # They should be sorted by date
                generated_at_str = most_recent.get('audit', {}).get('generated_at')
                
                if generated_at_str:
                    try:
                        # Parse the timestamp
                        generated_at = datetime.fromisoformat(generated_at_str.replace('Z', '+00:00'))
                        now = datetime.utcnow()
                        age_minutes = (now - generated_at.replace(tzinfo=None)).total_seconds() / 60
                        
                        print(f"\nMost recent suggestion:")
                        print(f"  Generated at: {generated_at_str}")
                        print(f"  Age: {age_minutes:.1f} minutes ago")
                        
                        if age_minutes < 120:  # Less than 2 hours
                            print(f"\n[SUCCESS] Scheduler appears to be active! (Recent suggestions found)")
                        else:
                            print(f"\n[INFO] Most recent suggestion is {age_minutes:.1f} minutes old")
                            print("  (Scheduler may be running but no new duplicates detected)")
                    except Exception as e:
                        print(f"\n[INFO] Could not parse timestamp: {e}")
                
                # Count suggestions by status
                statuses = {}
                for sug in suggestions:
                    status = sug.get('status', 'unknown')
                    statuses[status] = statuses.get(status, 0) + 1
                
                print(f"\nSuggestion status breakdown:")
                for status, count in statuses.items():
                    print(f"  - {status}: {count}")
                
                return True
            else:
                print("\n[INFO] No suggestions found yet")
                print("  The scheduler may need more time to detect duplicates")
                return True
        else:
            print(f"\n[ERROR] Failed to get suggestions: {response.status_code}")
            return False
    except Exception as e:
        print(f"\n[ERROR] Check failed: {e}")
        return False


def test_api_response_format():
    """Verify the API response format matches expected schema."""
    print_section("TEST 5: API Response Format Validation")
    
    try:
        response = requests.get(f"{BASE_URL}/suggestions", params={"limit": 1}, timeout=60)
        if response.status_code == 200:
            suggestions = response.json()
            
            if suggestions:
                suggestion = suggestions[0]
                print("\nValidating response structure...")
                
                required_fields = ['_id', 'candidate_ids', 'message', 'details', 'audit', 'status']
                missing_fields = [field for field in required_fields if field not in suggestion]
                
                if missing_fields:
                    print(f"[ERROR] Missing required fields: {missing_fields}")
                    return False
                else:
                    print("[OK] All required fields present")
                
                # Validate details structure
                details = suggestion.get('details', {})
                detail_fields = ['amount_delta_pct', 'time_delta_minutes', 'merchant_match_rule']
                missing_details = [f for f in detail_fields if f not in details]
                
                if missing_details:
                    print(f"[WARN] Missing detail fields: {missing_details}")
                else:
                    print("[OK] Details structure valid")
                
                # Validate audit structure
                audit = suggestion.get('audit', {})
                if 'generated_by' in audit and 'generated_at' in audit:
                    print("[OK] Audit structure valid")
                else:
                    print("[WARN] Audit structure incomplete")
                
                print("\n[SUCCESS] Response format is valid!")
                return True
            else:
                print("[INFO] No suggestions to validate (format check skipped)")
                return True
        else:
            print(f"[ERROR] Failed to get response: {response.status_code}")
            return False
    except Exception as e:
        print(f"[ERROR] Validation failed: {e}")
        return False


def main():
    """Run all API tests."""
    print("\n" + "=" * 70)
    print("  API TESTING SUITE")
    print("  Testing: " + BASE_URL)
    print("=" * 70)
    
    results = []
    
    # Run all tests
    results.append(("Health Endpoint", test_health_endpoint()))
    results.append(("Suggestions Endpoint", test_suggestions_endpoint()))
    results.append(("Custom Limit Parameter", test_suggestions_with_limit()))
    results.append(("Scheduler Activity", test_scheduler_activity()))
    results.append(("Response Format", test_api_response_format()))
    
    # Summary
    print_section("TEST SUMMARY")
    
    passed = sum(1 for _, result in results if result)
    total = len(results)
    
    print("\nTest Results:")
    for test_name, result in results:
        status = "[PASS]" if result else "[FAIL]"
        print(f"  {status} {test_name}")
    
    print(f"\n{'=' * 70}")
    print(f"  Total: {passed}/{total} tests passed")
    print("=" * 70)
    
    if passed == total:
        print("\n[SUCCESS] All API tests passed! The API is working correctly.")
        return 0
    else:
        print(f"\n[WARNING] {total - passed} test(s) failed. Check the details above.")
        return 1


if __name__ == "__main__":
    sys.exit(main())