test_integration.py

import urllib.request
import json
import random
import sys

BASE_URL = "http://127.0.0.1:8000"

def make_request(path, method="GET", data=None):
    url = f"{BASE_URL}{path}"
    headers = {"Content-Type": "application/json"}
    
    req_data = None
    if data is not None:
        req_data = json.dumps(data).encode("utf-8")
        
    req = urllib.request.Request(url, data=req_data, headers=headers, method=method)
    
    try:
        with urllib.request.urlopen(req) as response:
            status_code = response.getcode()
            res_body = response.read().decode("utf-8")
            return status_code, json.loads(res_body) if res_body else {}
    except urllib.error.HTTPError as e:
        res_body = e.read().decode("utf-8")
        print(f"HTTP Error {e.code}: {res_body}")
        raise e

def run_integration_tests():
    print("=== Starting live API Integration Tests ===")
    
    # 1. Create a unique user
    username = f"user_{random.randint(1000, 9999)}"
    email = f"{username}@example.com"
    user_payload = {
        "username": username,
        "email": email,
        "password": "secure_password",
        "full_name": "Integration Test User",
        "bio": "Testing live API endpoints"
    }
    print(f"\n1. POST /users/ (creating user: {username})")
    status, user = make_request("/users/", method="POST", data=user_payload)
    assert status == 200, f"Expected 200, got {status}"
    user_id = user["id"]
    print(f"Success! User ID: {user_id}")

    # 1b. Test POST /login
    print(f"\n1b. POST /login (authenticating user: {username})")
    login_payload = {
        "username": username,
        "password": "secure_password"
    }
    status, login_res = make_request("/login", method="POST", data=login_payload)
    assert status == 200
    assert "access_token" in login_res
    assert login_res["token_type"] == "bearer"
    print("Success! Login token retrieved.")
    # 2. Create tasks for the user
    task1_payload = {
        "title": "Integration Task 1",
        "description": "Verify task creation and association",
        "completed": False
    }
    task2_payload = {
        "title": "Integration Task 2",
        "description": "Verify another task for the same user",
        "completed": True
    }
    
    print(f"\n2. POST /users/{user_id}/tasks (creating Task 1)")
    status, task1 = make_request(f"/users/{user_id}/tasks", method="POST", data=task1_payload)
    assert status == 200
    task1_id = task1["id"]
    print(f"Success! Task 1 ID: {task1_id}")
    
    print(f"\n3. POST /users/{user_id}/tasks (creating Task 2)")
    status, task2 = make_request(f"/users/{user_id}/tasks", method="POST", data=task2_payload)
    assert status == 200
    task2_id = task2["id"]
    print(f"Success! Task 2 ID: {task2_id}")
    
    # 3. GET user detail and verify associated tasks
    print(f"\n4. GET /users/{user_id} (fetching user detail & tasks)")
    status, retrieved_user = make_request(f"/users/{user_id}")
    assert status == 200
    print("Retrieved user and tasks count:", len(retrieved_user.get("tasks", [])))
    assert len(retrieved_user["tasks"]) >= 2
    
    # Verify titles
    task_titles = [t["title"] for t in retrieved_user["tasks"]]
    print("Associated tasks in response:", task_titles)
    assert "Integration Task 1" in task_titles
    assert "Integration Task 2" in task_titles
    
    # 4. GET /tasks/{task_id}
    print(f"\n5. GET /tasks/{task1_id} (fetching single task)")
    status, single_task = make_request(f"/tasks/{task1_id}")
    assert status == 200
    assert single_task["title"] == "Integration Task 1"
    print(f"Success! Retrieved task title: '{single_task['title']}'")
    
    # 5. PUT /tasks/{task_id}
    print(f"\n6. PUT /tasks/{task1_id} (updating task)")
    update_payload = {
        "title": "Updated Integration Task 1",
        "description": "Updated description",
        "completed": True
    }
    status, updated_task = make_request(f"/tasks/{task1_id}", method="PUT", data=update_payload)
    assert status == 200
    assert updated_task["title"] == "Updated Integration Task 1"
    assert updated_task["completed"] is True
    print(f"Success! Updated task title: '{updated_task['title']}', completed: {updated_task['completed']}")
    
    # 5b. Advanced Search and Filters Tests
    print(f"\n5b. Testing Advanced Search and Filters for user {user_id}")
    
    # Let's create extra tasks first to have a diverse set
    task_a_payload = {
        "title": "Write documentation",
        "description": "Important work for the project",
        "completed": False
    }
    task_b_payload = {
        "title": "Write unit tests",
        "description": "Ensure high coverage",
        "completed": True
    }
    task_c_payload = {
        "title": "Refactor database",
        "description": "Optimize schema and indices",
        "completed": False
    }
    
    make_request(f"/users/{user_id}/tasks", method="POST", data=task_a_payload)
    make_request(f"/users/{user_id}/tasks", method="POST", data=task_b_payload)
    make_request(f"/users/{user_id}/tasks", method="POST", data=task_c_payload)
    
    # Test 5b.1: Search by text query
    print("Testing GET /tasks?q=write...")
    status, search_res = make_request(f"/tasks?user_id={user_id}&q=write")
    assert status == 200
    titles = [t["title"] for t in search_res]
    assert len(titles) == 2
    assert "Write documentation" in titles
    assert "Write unit tests" in titles
    
    # Test 5b.2: Filter by completion
    print("Testing GET /tasks?completed=false...")
    status, filter_res = make_request(f"/tasks?user_id={user_id}&completed=false")
    assert status == 200
    completed_statuses = [t["completed"] for t in filter_res]
    assert all(s is False for s in completed_statuses)
    
    # Test 5b.3: Sort tasks
    print("Testing GET /tasks?sort_by=title&sort_order=asc...")
    status, sort_res = make_request(f"/tasks?user_id={user_id}&sort_by=title&sort_order=asc")
    assert status == 200
    sorted_titles = [t["title"] for t in sort_res]
    assert sorted_titles == sorted(sorted_titles)
    
    # Test 5b.4: Pagination
    print("Testing GET /tasks?limit=2&offset=1...")
    status, paginated_res = make_request(f"/tasks?user_id={user_id}&limit=2&offset=1")
    assert status == 200
    assert len(paginated_res) == 2
    
    # Test 5b.5: Create Saved Filter Settings
    print("Testing POST /users/{user_id}/filter-settings...")
    filter_setting_payload = {
        "name": "Important Uncompleted Tasks",
        "filters": {
            "conditions": [
                {"field": "completed", "operator": "eq", "value": False},
                {"field": "description", "operator": "contains", "value": "Important"}
            ],
            "sort_by": "title",
            "sort_order": "desc",
            "limit": 10,
            "offset": 0
        }
    }
    status, setting_res = make_request(f"/users/{user_id}/filter-settings", method="POST", data=filter_setting_payload)
    assert status == 200
    setting_id = setting_res["id"]
    print(f"Filter Setting saved with ID: {setting_id}")
    
    # Test 5b.6: List Saved Filter Settings
    print("Testing GET /users/{user_id}/filter-settings...")
    status, settings_list = make_request(f"/users/{user_id}/filter-settings")
    assert status == 200
    assert len(settings_list) > 0
    assert settings_list[0]["name"] == "Important Uncompleted Tasks"
    
    # Test 5b.7: Apply Saved Filter Settings
    print("Testing GET /users/{user_id}/filter-settings/{setting_id}/apply...")
    status, applied_res = make_request(f"/users/{user_id}/filter-settings/{setting_id}/apply")
    assert status == 200
    assert len(applied_res) == 1
    assert applied_res[0]["title"] == "Write documentation"
    print("Success! Applied filter setting returned the correct matching task.")

    # 6. DELETE /tasks/{task_id}
    print(f"\n7. DELETE /tasks/{task1_id} (deleting task)")
    status, delete_res = make_request(f"/tasks/{task1_id}", method="DELETE")
    assert status == 200
    print(f"Success! Delete message: {delete_res.get('message')}")
    
    # 7. GET /tasks/{task_id} after deletion should fail
    print(f"\n8. GET /tasks/{task1_id} (should fail with 404)")
    try:
        make_request(f"/tasks/{task1_id}")
        assert False, "Should have failed with 404"
    except urllib.error.HTTPError as e:
        assert e.code == 404
        print(f"Success! Request failed as expected with 404")
        
    print("\n=== Live API Integration Tests PASSED successfully! ===")

if __name__ == "__main__":
    try:
        run_integration_tests()
    except Exception as e:
        print(f"\nTest Execution Failed: {e}")
        sys.exit(1)