363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""
|
|
Test script to verify task endpoints support custom statuses.
|
|
|
|
Tests:
|
|
1. Task creation uses default status if not specified
|
|
2. Task status validation checks both system and custom statuses
|
|
3. Status resolution works across project boundaries
|
|
4. Bulk status update validates custom statuses
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from database import Base
|
|
from models.user import User, UserRole
|
|
from models.project import Project, ProjectType, ProjectStatus, ProjectMember
|
|
from models.task import Task
|
|
from models.asset import Asset, AssetCategory, AssetStatus
|
|
from models.episode import Episode, EpisodeStatus
|
|
from models.shot import Shot, ShotStatus
|
|
import json
|
|
|
|
# Create test database
|
|
TEST_DB = "test_task_custom_status.db"
|
|
if os.path.exists(TEST_DB):
|
|
os.remove(TEST_DB)
|
|
|
|
engine = create_engine(f"sqlite:///{TEST_DB}")
|
|
Base.metadata.create_all(engine)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
|
|
def test_task_creation_with_default_status():
|
|
"""Test that task creation uses project default status"""
|
|
print("\n=== Test 1: Task Creation with Default Status ===")
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Create test user
|
|
user = User(
|
|
email="coordinator@test.com",
|
|
password_hash="test",
|
|
first_name="Test",
|
|
last_name="Coordinator",
|
|
role=UserRole.COORDINATOR,
|
|
is_approved=True
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
|
|
# Create project with custom statuses
|
|
project = Project(
|
|
name="Test Project",
|
|
code_name="TEST",
|
|
client_name="Test Client",
|
|
project_type=ProjectType.CINEMA,
|
|
status=ProjectStatus.IN_PROGRESS,
|
|
custom_task_statuses=json.dumps([
|
|
{
|
|
"id": "custom_todo",
|
|
"name": "To Do",
|
|
"color": "#FF0000",
|
|
"order": 0,
|
|
"is_default": True
|
|
},
|
|
{
|
|
"id": "custom_doing",
|
|
"name": "Doing",
|
|
"color": "#00FF00",
|
|
"order": 1,
|
|
"is_default": False
|
|
}
|
|
])
|
|
)
|
|
db.add(project)
|
|
db.commit()
|
|
|
|
# Add user as project member
|
|
member = ProjectMember(
|
|
user_id=user.id,
|
|
project_id=project.id
|
|
)
|
|
db.add(member)
|
|
db.commit()
|
|
|
|
# Create asset for task
|
|
asset = Asset(
|
|
project_id=project.id,
|
|
name="Test Asset",
|
|
category=AssetCategory.CHARACTERS,
|
|
status=AssetStatus.IN_PROGRESS
|
|
)
|
|
db.add(asset)
|
|
db.commit()
|
|
|
|
# Import helper functions
|
|
from routers.tasks import get_project_default_status, validate_task_status
|
|
|
|
# Test 1a: Get default status
|
|
default_status = get_project_default_status(db, project.id)
|
|
assert default_status == "custom_todo", f"Expected 'custom_todo', got '{default_status}'"
|
|
print(f"✓ Default status correctly identified: {default_status}")
|
|
|
|
# Test 1b: Create task without explicit status (should use default)
|
|
task1 = Task(
|
|
project_id=project.id,
|
|
asset_id=asset.id,
|
|
name="Test Task 1",
|
|
task_type="modeling",
|
|
status=get_project_default_status(db, project.id)
|
|
)
|
|
db.add(task1)
|
|
db.commit()
|
|
|
|
assert task1.status == "custom_todo", f"Expected 'custom_todo', got '{task1.status}'"
|
|
print(f"✓ Task created with default status: {task1.status}")
|
|
|
|
# Test 1c: Create task with explicit custom status
|
|
task2 = Task(
|
|
project_id=project.id,
|
|
asset_id=asset.id,
|
|
name="Test Task 2",
|
|
task_type="modeling",
|
|
status="custom_doing"
|
|
)
|
|
db.add(task2)
|
|
db.commit()
|
|
|
|
assert task2.status == "custom_doing", f"Expected 'custom_doing', got '{task2.status}'"
|
|
print(f"✓ Task created with explicit custom status: {task2.status}")
|
|
|
|
print("✓ Test 1 PASSED: Task creation with default status works correctly")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_status_validation():
|
|
"""Test that status validation checks both system and custom statuses"""
|
|
print("\n=== Test 2: Status Validation ===")
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Get existing project
|
|
project = db.query(Project).first()
|
|
|
|
from routers.tasks import validate_task_status
|
|
|
|
# Test 2a: Validate system status
|
|
assert validate_task_status(db, project.id, "not_started"), "System status 'not_started' should be valid"
|
|
assert validate_task_status(db, project.id, "in_progress"), "System status 'in_progress' should be valid"
|
|
assert validate_task_status(db, project.id, "submitted"), "System status 'submitted' should be valid"
|
|
assert validate_task_status(db, project.id, "approved"), "System status 'approved' should be valid"
|
|
assert validate_task_status(db, project.id, "retake"), "System status 'retake' should be valid"
|
|
print("✓ All system statuses validated successfully")
|
|
|
|
# Test 2b: Validate custom status
|
|
assert validate_task_status(db, project.id, "custom_todo"), "Custom status 'custom_todo' should be valid"
|
|
assert validate_task_status(db, project.id, "custom_doing"), "Custom status 'custom_doing' should be valid"
|
|
print("✓ All custom statuses validated successfully")
|
|
|
|
# Test 2c: Validate invalid status
|
|
assert not validate_task_status(db, project.id, "invalid_status"), "Invalid status should not be valid"
|
|
assert not validate_task_status(db, project.id, "custom_nonexistent"), "Non-existent custom status should not be valid"
|
|
print("✓ Invalid statuses correctly rejected")
|
|
|
|
print("✓ Test 2 PASSED: Status validation works correctly")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_status_resolution_across_projects():
|
|
"""Test that status resolution works correctly across project boundaries"""
|
|
print("\n=== Test 3: Status Resolution Across Projects ===")
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Create second project with different custom statuses
|
|
project2 = Project(
|
|
name="Test Project 2",
|
|
code_name="TEST2",
|
|
client_name="Test Client 2",
|
|
project_type=ProjectType.TV,
|
|
status=ProjectStatus.IN_PROGRESS,
|
|
custom_task_statuses=json.dumps([
|
|
{
|
|
"id": "project2_status",
|
|
"name": "Project 2 Status",
|
|
"color": "#0000FF",
|
|
"order": 0,
|
|
"is_default": True
|
|
}
|
|
])
|
|
)
|
|
db.add(project2)
|
|
db.commit()
|
|
|
|
# Create asset for project 2
|
|
asset2 = Asset(
|
|
project_id=project2.id,
|
|
name="Test Asset 2",
|
|
category=AssetCategory.PROPS,
|
|
status=AssetStatus.IN_PROGRESS
|
|
)
|
|
db.add(asset2)
|
|
db.commit()
|
|
|
|
from routers.tasks import validate_task_status, get_project_default_status
|
|
|
|
project1 = db.query(Project).filter(Project.code_name == "TEST").first()
|
|
|
|
# Test 3a: Project 1 custom status should not be valid for Project 2
|
|
assert not validate_task_status(db, project2.id, "custom_todo"), \
|
|
"Project 1 custom status should not be valid for Project 2"
|
|
print("✓ Project 1 custom status correctly rejected for Project 2")
|
|
|
|
# Test 3b: Project 2 custom status should not be valid for Project 1
|
|
assert not validate_task_status(db, project1.id, "project2_status"), \
|
|
"Project 2 custom status should not be valid for Project 1"
|
|
print("✓ Project 2 custom status correctly rejected for Project 1")
|
|
|
|
# Test 3c: System statuses should be valid for both projects
|
|
assert validate_task_status(db, project1.id, "not_started"), \
|
|
"System status should be valid for Project 1"
|
|
assert validate_task_status(db, project2.id, "not_started"), \
|
|
"System status should be valid for Project 2"
|
|
print("✓ System statuses valid for both projects")
|
|
|
|
# Test 3d: Each project has its own default status
|
|
default1 = get_project_default_status(db, project1.id)
|
|
default2 = get_project_default_status(db, project2.id)
|
|
assert default1 == "custom_todo", f"Project 1 default should be 'custom_todo', got '{default1}'"
|
|
assert default2 == "project2_status", f"Project 2 default should be 'project2_status', got '{default2}'"
|
|
print(f"✓ Project 1 default: {default1}, Project 2 default: {default2}")
|
|
|
|
print("✓ Test 3 PASSED: Status resolution across projects works correctly")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_bulk_status_update_validation():
|
|
"""Test that bulk status update validates custom statuses"""
|
|
print("\n=== Test 4: Bulk Status Update Validation ===")
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
project1 = db.query(Project).filter(Project.code_name == "TEST").first()
|
|
project2 = db.query(Project).filter(Project.code_name == "TEST2").first()
|
|
|
|
# Get tasks from both projects
|
|
tasks_p1 = db.query(Task).filter(Task.project_id == project1.id).all()
|
|
tasks_p2 = db.query(Task).filter(Task.project_id == project2.id).all()
|
|
|
|
from routers.tasks import validate_task_status
|
|
|
|
# Test 4a: Validate status for each task's project
|
|
for task in tasks_p1:
|
|
# Should be valid for project 1
|
|
assert validate_task_status(db, task.project_id, "custom_todo"), \
|
|
f"custom_todo should be valid for task {task.id} in project 1"
|
|
# Should not be valid for project 2 status
|
|
assert not validate_task_status(db, task.project_id, "project2_status"), \
|
|
f"project2_status should not be valid for task {task.id} in project 1"
|
|
print(f"✓ Validated {len(tasks_p1)} tasks from Project 1")
|
|
|
|
for task in tasks_p2:
|
|
# Should be valid for project 2
|
|
assert validate_task_status(db, task.project_id, "project2_status"), \
|
|
f"project2_status should be valid for task {task.id} in project 2"
|
|
# Should not be valid for project 1 status
|
|
assert not validate_task_status(db, task.project_id, "custom_todo"), \
|
|
f"custom_todo should not be valid for task {task.id} in project 2"
|
|
print(f"✓ Validated {len(tasks_p2)} tasks from Project 2")
|
|
|
|
# Test 4b: System statuses should be valid for all tasks
|
|
for task in tasks_p1 + tasks_p2:
|
|
assert validate_task_status(db, task.project_id, "in_progress"), \
|
|
f"System status should be valid for task {task.id}"
|
|
print(f"✓ System statuses valid for all {len(tasks_p1) + len(tasks_p2)} tasks")
|
|
|
|
print("✓ Test 4 PASSED: Bulk status update validation works correctly")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_project_without_custom_statuses():
|
|
"""Test that projects without custom statuses use system defaults"""
|
|
print("\n=== Test 5: Project Without Custom Statuses ===")
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Create project without custom statuses
|
|
project3 = Project(
|
|
name="Test Project 3",
|
|
code_name="TEST3",
|
|
client_name="Test Client 3",
|
|
project_type=ProjectType.TV,
|
|
status=ProjectStatus.IN_PROGRESS
|
|
)
|
|
db.add(project3)
|
|
db.commit()
|
|
|
|
from routers.tasks import get_project_default_status, validate_task_status
|
|
|
|
# Test 5a: Default status should be system default
|
|
default = get_project_default_status(db, project3.id)
|
|
assert default == "not_started", f"Expected 'not_started', got '{default}'"
|
|
print(f"✓ Project without custom statuses uses system default: {default}")
|
|
|
|
# Test 5b: System statuses should be valid
|
|
assert validate_task_status(db, project3.id, "not_started"), "System status should be valid"
|
|
assert validate_task_status(db, project3.id, "in_progress"), "System status should be valid"
|
|
print("✓ System statuses valid for project without custom statuses")
|
|
|
|
# Test 5c: Custom statuses from other projects should not be valid
|
|
assert not validate_task_status(db, project3.id, "custom_todo"), \
|
|
"Custom status from other project should not be valid"
|
|
print("✓ Custom statuses from other projects correctly rejected")
|
|
|
|
print("✓ Test 5 PASSED: Projects without custom statuses work correctly")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=" * 60)
|
|
print("Testing Task Endpoints Custom Status Support")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
test_task_creation_with_default_status()
|
|
test_status_validation()
|
|
test_status_resolution_across_projects()
|
|
test_bulk_status_update_validation()
|
|
test_project_without_custom_statuses()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✓ ALL TESTS PASSED")
|
|
print("=" * 60)
|
|
|
|
except AssertionError as e:
|
|
print(f"\n✗ TEST FAILED: {e}")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"\n✗ ERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
finally:
|
|
# Cleanup
|
|
try:
|
|
if os.path.exists(TEST_DB):
|
|
os.remove(TEST_DB)
|
|
except PermissionError:
|
|
# On Windows, the file might still be locked
|
|
pass
|