test_notifications.py

"""
Test script for POST /notifications/register endpoint.
Tests push token registration with JWT authentication.

Usage:
    python test_notifications.py <JWT_TOKEN>
"""
import sys
import asyncio
import httpx


BASE_URL = "http://localhost:8003"
ENDPOINT = "/tracker/notifications/register"


async def test_register_ios_token(token: str):
    """Test registering iOS push token"""
    print("\n" + "="*60)
    print("Test 1: Register iOS Push Token")
    print("="*60)
    
    payload = {
        "token": "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
        "platform": "ios"
    }
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}{ENDPOINT}",
                json=payload,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json"
                },
                timeout=10.0
            )
            
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.json()}")
            
            if response.status_code == 200:
                print("✓ Test 1 PASSED")
                return True
            else:
                print("✗ Test 1 FAILED")
                return False
                
    except Exception as e:
        print(f"✗ Test 1 FAILED: {str(e)}")
        return False


async def test_register_android_token(token: str):
    """Test registering Android push token"""
    print("\n" + "="*60)
    print("Test 2: Register Android Push Token")
    print("="*60)
    
    payload = {
        "token": "fcm_token_xxxxxxxxxxxxxxxxxxxxxxxxxx",
        "platform": "android"
    }
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}{ENDPOINT}",
                json=payload,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json"
                },
                timeout=10.0
            )
            
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.json()}")
            
            if response.status_code == 200:
                print("✓ Test 2 PASSED")
                return True
            else:
                print("✗ Test 2 FAILED")
                return False
                
    except Exception as e:
        print(f"✗ Test 2 FAILED: {str(e)}")
        return False


async def test_update_token(token: str):
    """Test updating existing token"""
    print("\n" + "="*60)
    print("Test 3: Update Existing Token")
    print("="*60)
    
    payload = {
        "token": "ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
        "platform": "ios"
    }
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}{ENDPOINT}",
                json=payload,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json"
                },
                timeout=10.0
            )
            
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.json()}")
            
            if response.status_code == 200:
                print("✓ Test 3 PASSED")
                return True
            else:
                print("✗ Test 3 FAILED")
                return False
                
    except Exception as e:
        print(f"✗ Test 3 FAILED: {str(e)}")
        return False


async def test_invalid_platform(token: str):
    """Test with invalid platform"""
    print("\n" + "="*60)
    print("Test 4: Invalid Platform (Should fail)")
    print("="*60)
    
    payload = {
        "token": "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
        "platform": "windows"
    }
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}{ENDPOINT}",
                json=payload,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json"
                },
                timeout=10.0
            )
            
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.text}")
            
            if response.status_code == 422:
                print("✓ Test 4 PASSED")
                return True
            else:
                print("✗ Test 4 FAILED")
                return False
                
    except Exception as e:
        print(f"✗ Test 4 FAILED: {str(e)}")
        return False


async def test_missing_token_auth(token: str):
    """Test without JWT token"""
    print("\n" + "="*60)
    print("Test 5: Missing JWT Token (Should fail)")
    print("="*60)
    
    payload = {
        "token": "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
        "platform": "ios"
    }
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{BASE_URL}{ENDPOINT}",
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=10.0
            )
            
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.text}")
            
            if response.status_code in [401, 403]:
                print("✓ Test 5 PASSED")
                return True
            else:
                print("✗ Test 5 FAILED")
                return False
                
    except Exception as e:
        print(f"✗ Test 5 FAILED: {str(e)}")
        return False


async def run_all_tests(token: str):
    """Run all test cases"""
    print("="*60)
    print("POST /notifications/register - Test Suite")
    print("="*60)
    
    results = []
    results.append(await test_register_ios_token(token))
    results.append(await test_register_android_token(token))
    results.append(await test_update_token(token))
    results.append(await test_invalid_platform(token))
    results.append(await test_missing_token_auth(token))
    
    print("\n" + "="*60)
    print("Test Summary")
    print("="*60)
    passed = sum(results)
    total = len(results)
    print(f"Tests Passed: {passed}/{total}")
    
    if passed == total:
        print("✅ All tests passed!")
    else:
        print("❌ Some tests failed")


def main():
    if len(sys.argv) < 2:
        print("Usage: python test_notifications.py <JWT_TOKEN>")
        print("\nGenerate token with: python generate_test_token.py")
        sys.exit(1)
    
    token = sys.argv[1]
    asyncio.run(run_all_tests(token))


if __name__ == "__main__":
    main()