138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import requests
|
|
import json
|
|
|
|
# Configuration
|
|
BASE_URL = "http://localhost:8000"
|
|
LOGIN_URL = f"{BASE_URL}/auth/login"
|
|
ASSETS_URL = f"{BASE_URL}/assets/"
|
|
TASKS_URL = f"{BASE_URL}/tasks/"
|
|
|
|
def test_task_status_update():
|
|
print("Testing Task Status Update Functionality")
|
|
print("=" * 50)
|
|
|
|
# Login first
|
|
print("1. Logging in...")
|
|
login_data = {
|
|
"email": "admin@vfx.com",
|
|
"password": "admin123"
|
|
}
|
|
|
|
response = requests.post(LOGIN_URL, json=login_data)
|
|
if response.status_code != 200:
|
|
print(f"❌ Login failed: {response.status_code}")
|
|
print(response.text)
|
|
return
|
|
|
|
token_data = response.json()
|
|
token = token_data["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
print("✅ Login successful")
|
|
|
|
# Get projects first
|
|
print("\n2. Getting projects...")
|
|
projects_response = requests.get(f"{BASE_URL}/projects/", headers=headers)
|
|
if projects_response.status_code != 200:
|
|
print(f"❌ Failed to get projects: {projects_response.status_code}")
|
|
return
|
|
|
|
projects = projects_response.json()
|
|
if not projects:
|
|
print("❌ No projects found")
|
|
return
|
|
|
|
project_id = projects[0]["id"]
|
|
print(f"✅ Using project ID: {project_id}")
|
|
|
|
# Get assets
|
|
print("\n3. Getting assets...")
|
|
params = {"project_id": project_id}
|
|
response = requests.get(ASSETS_URL, params=params, headers=headers)
|
|
|
|
if response.status_code != 200:
|
|
print(f"❌ Failed to get assets: {response.status_code}")
|
|
return
|
|
|
|
assets = response.json()
|
|
if not assets:
|
|
print("❌ No assets found")
|
|
return
|
|
|
|
asset = assets[0]
|
|
print(f"✅ Using asset: {asset['name']} (ID: {asset['id']})")
|
|
print(f" Current task status: {asset.get('task_status', {})}")
|
|
|
|
# Test creating a new task if none exists
|
|
print("\n4. Testing task creation...")
|
|
task_type = "modeling"
|
|
|
|
# Check if task already exists
|
|
existing_task_id = None
|
|
if asset.get('task_details'):
|
|
for task_detail in asset['task_details']:
|
|
if task_detail['task_type'] == task_type:
|
|
existing_task_id = task_detail.get('task_id')
|
|
break
|
|
|
|
if not existing_task_id:
|
|
print(f" Creating new {task_type} task...")
|
|
create_response = requests.post(
|
|
f"{ASSETS_URL}{asset['id']}/tasks?task_type={task_type}",
|
|
headers=headers
|
|
)
|
|
|
|
if create_response.status_code == 201:
|
|
task_info = create_response.json()
|
|
task_id = task_info['task_id']
|
|
print(f"✅ Created task with ID: {task_id}")
|
|
else:
|
|
print(f"❌ Failed to create task: {create_response.status_code}")
|
|
print(create_response.text)
|
|
return
|
|
else:
|
|
task_id = existing_task_id
|
|
print(f"✅ Using existing task ID: {task_id}")
|
|
|
|
# Test updating task status
|
|
print("\n5. Testing task status update...")
|
|
new_status = "in_progress"
|
|
|
|
update_response = requests.put(
|
|
f"{TASKS_URL}{task_id}/status",
|
|
json={"status": new_status},
|
|
headers=headers
|
|
)
|
|
|
|
if update_response.status_code == 200:
|
|
updated_task = update_response.json()
|
|
print(f"✅ Task status updated to: {updated_task['status']}")
|
|
else:
|
|
print(f"❌ Failed to update task status: {update_response.status_code}")
|
|
print(update_response.text)
|
|
return
|
|
|
|
# Verify the update by fetching assets again
|
|
print("\n6. Verifying update...")
|
|
response = requests.get(ASSETS_URL, params=params, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
updated_assets = response.json()
|
|
updated_asset = next((a for a in updated_assets if a['id'] == asset['id']), None)
|
|
|
|
if updated_asset:
|
|
updated_status = updated_asset.get('task_status', {}).get(task_type)
|
|
if updated_status == new_status:
|
|
print(f"✅ Status update verified: {task_type} = {updated_status}")
|
|
else:
|
|
print(f"❌ Status update not reflected: expected {new_status}, got {updated_status}")
|
|
else:
|
|
print("❌ Asset not found in updated list")
|
|
else:
|
|
print(f"❌ Failed to verify update: {response.status_code}")
|
|
|
|
print("\n✅ Task status update test completed!")
|
|
|
|
if __name__ == "__main__":
|
|
test_task_status_update() |