test_tasks_api.py

"""
Test script for Tasks API endpoints.
Tests both GET /tasks/today and PATCH /tasks/:id/status endpoints.
"""
import asyncio
import asyncpg
import os
import requests
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configuration
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8003")
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_PORT = int(os.getenv("DB_PORT", "5432"))
DB_NAME = os.getenv("DB_NAME", "cuatrolabs")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")

# Test data
TEST_MERCHANT_ID = "550e8400-e29b-41d4-a716-446655440000"
TEST_EMPLOYEE_ID = "660e8400-e29b-41d4-a716-446655440001"


def get_auth_token():
    """Read JWT token from test_token.txt"""
    try:
        with open("test_token.txt", "r") as f:
            return f.read().strip()
    except FileNotFoundError:
        print("❌ test_token.txt not found. Run generate_test_token.py first.")
        exit(1)


async def create_test_task():
    """Create a test task in the database"""
    conn = await asyncpg.connect(
        host=DB_HOST,
        port=DB_PORT,
        database=DB_NAME,
        user=DB_USER,
        password=DB_PASSWORD
    )
    
    try:
        # Delete existing test tasks
        await conn.execute("""
            DELETE FROM trans.scm_tasks 
            WHERE merchant_id = $1::uuid 
            AND assigned_to = $2::uuid
        """, TEST_MERCHANT_ID, TEST_EMPLOYEE_ID)
        
        # Create new test task
        scheduled_time = datetime.now() + timedelta(hours=2)
        
        task_id = await conn.fetchval("""
            INSERT INTO trans.scm_tasks (
                merchant_id, assigned_to, title, description,
                status, latitude, longitude, address, scheduled_at
            ) VALUES (
                $1::uuid, $2::uuid, $3, $4, $5, $6, $7, $8, $9
            ) RETURNING id
        """, 
            TEST_MERCHANT_ID,
            TEST_EMPLOYEE_ID,
            "Test Task - Visit Customer",
            "Deliver products and collect payment",
            "not_started",
            19.0760,
            72.8777,
            "123 Test Street, Mumbai",
            scheduled_time
        )
        
        print(f"✅ Test task created: {task_id}")
        return str(task_id)
        
    finally:
        await conn.close()


def test_get_today_tasks(token):
    """Test GET /tracker/tasks/today endpoint"""
    print("\n" + "="*80)
    print("TEST 1: GET /tracker/tasks/today")
    print("="*80)
    
    url = f"{API_BASE_URL}/tracker/tasks/today"
    headers = {"Authorization": f"Bearer {token}"}
    
    print(f"Request: GET {url}")
    response = requests.get(url, headers=headers)
    
    print(f"Status Code: {response.status_code}")
    print(f"Response: {response.json()}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("success") and data.get("count", 0) > 0:
            print("✅ Test passed: Tasks retrieved successfully")
            return data["tasks"][0]["id"]  # Return first task ID for next tests
        else:
            print("⚠️  No tasks found for today")
            return None
    else:
        print("❌ Test failed")
        return None


def test_update_task_status(token, task_id, new_status, should_succeed=True):
    """Test PATCH /tracker/tasks/:id/status endpoint"""
    print("\n" + "="*80)
    print(f"TEST: Update task status to '{new_status}'")
    print("="*80)
    
    url = f"{API_BASE_URL}/tracker/tasks/{task_id}/status"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "status": new_status,
        "timestamp": int(time.time() * 1000),  # Current time in milliseconds
        "latitude": 19.0760 + (0.001 * hash(new_status) % 10),  # Slightly different location
        "longitude": 72.8777 + (0.001 * hash(new_status) % 10)
    }
    
    print(f"Request: PATCH {url}")
    print(f"Payload: {payload}")
    
    response = requests.patch(url, headers=headers, json=payload)
    
    print(f"Status Code: {response.status_code}")
    print(f"Response: {response.json()}")
    
    if should_succeed:
        if response.status_code == 200:
            print(f"✅ Test passed: Status updated to '{new_status}'")
            return True
        else:
            print(f"❌ Test failed: Expected success but got {response.status_code}")
            return False
    else:
        if response.status_code in [400, 404]:
            print(f"✅ Test passed: Invalid transition correctly rejected")
            return True
        else:
            print(f"❌ Test failed: Expected error but got {response.status_code}")
            return False


def test_invalid_task_id(token):
    """Test with invalid task ID"""
    print("\n" + "="*80)
    print("TEST: Invalid task ID (should return 404)")
    print("="*80)
    
    fake_task_id = "00000000-0000-0000-0000-000000000000"
    url = f"{API_BASE_URL}/tracker/tasks/{fake_task_id}/status"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "status": "in_progress",
        "timestamp": int(time.time() * 1000),
        "latitude": 19.0760,
        "longitude": 72.8777
    }
    
    print(f"Request: PATCH {url}")
    response = requests.patch(url, headers=headers, json=payload)
    
    print(f"Status Code: {response.status_code}")
    print(f"Response: {response.json()}")
    
    if response.status_code == 404:
        print("✅ Test passed: Invalid task ID correctly rejected")
        return True
    else:
        print(f"❌ Test failed: Expected 404 but got {response.status_code}")
        return False


async def main():
    """Run all tests"""
    print("="*80)
    print("TASKS API TEST SUITE")
    print("="*80)
    print(f"API Base URL: {API_BASE_URL}")
    print(f"Database: {DB_HOST}:{DB_PORT}/{DB_NAME}")
    print("="*80)
    
    # Get auth token
    print("\n[1/9] Getting authentication token...")
    token = get_auth_token()
    print("✅ Token loaded")
    
    # Create test task
    print("\n[2/9] Creating test task...")
    task_id = await create_test_task()
    
    # Test 1: Get today's tasks
    print("\n[3/9] Testing GET /tracker/tasks/today...")
    retrieved_task_id = test_get_today_tasks(token)
    
    if not retrieved_task_id:
        print("\n❌ Cannot continue tests without a task ID")
        return
    
    # Use the retrieved task ID for remaining tests
    task_id = retrieved_task_id
    
    # Test 2: Update to in_progress (valid transition)
    print("\n[4/9] Testing status update: not_started → in_progress...")
    test_update_task_status(token, task_id, "in_progress", should_succeed=True)
    
    # Test 3: Update to completed (valid transition)
    print("\n[5/9] Testing status update: in_progress → completed...")
    test_update_task_status(token, task_id, "completed", should_succeed=True)
    
    # Test 4: Reopen task (valid transition)
    print("\n[6/9] Testing status update: completed → in_progress...")
    test_update_task_status(token, task_id, "in_progress", should_succeed=True)
    
    # Test 5: Complete again (valid transition)
    print("\n[7/9] Testing status update: in_progress → completed...")
    test_update_task_status(token, task_id, "completed", should_succeed=True)
    
    # Test 6: Invalid transition (should fail)
    print("\n[8/9] Testing invalid transition: completed → not_started...")
    test_update_task_status(token, task_id, "not_started", should_succeed=False)
    
    # Test 7: Invalid task ID
    print("\n[9/9] Testing with invalid task ID...")
    test_invalid_task_id(token)
    
    # Summary
    print("\n" + "="*80)
    print("TEST SUITE COMPLETED")
    print("="*80)
    print("\nAll tests executed. Review results above.")
    print("\nTo verify in database:")
    print(f"  SELECT * FROM trans.scm_tasks WHERE id = '{task_id}'::uuid;")
    print("="*80)


if __name__ == "__main__":
    asyncio.run(main())