364 lines
12 KiB
Python
364 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for data consistency and real-time update functionality.
|
|
|
|
This script tests:
|
|
1. Data consistency checks between individual task updates and aggregated views
|
|
2. Real-time update propagation to aggregated data
|
|
3. Task status changes reflecting in embedded data
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import asyncio
|
|
import json
|
|
from datetime import datetime
|
|
|
|
# Add the backend directory to the Python path
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from database import get_db, engine, Base
|
|
from models.project import Project
|
|
from models.episode import Episode
|
|
from models.shot import Shot
|
|
from models.asset import Asset, AssetCategory
|
|
from models.task import Task
|
|
from models.user import User, UserRole
|
|
from services.data_consistency import create_data_consistency_service
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
def setup_test_data(db: Session):
|
|
"""Set up test data for consistency validation."""
|
|
print("Setting up test data...")
|
|
|
|
# Create a test user with unique email
|
|
import uuid
|
|
unique_id = str(uuid.uuid4())[:8]
|
|
|
|
from passlib.context import CryptContext
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
test_user = User(
|
|
email=f"test_{unique_id}@example.com",
|
|
password_hash=pwd_context.hash("testpassword"),
|
|
first_name="Test",
|
|
last_name="User",
|
|
role=UserRole.COORDINATOR,
|
|
is_admin=True,
|
|
is_approved=True
|
|
)
|
|
db.add(test_user)
|
|
db.commit()
|
|
db.refresh(test_user)
|
|
|
|
# Create a test project with unique name
|
|
test_project = Project(
|
|
name=f"Test Project {unique_id}",
|
|
code_name=f"TEST_{unique_id}",
|
|
client_name="Test Client",
|
|
project_type="tv",
|
|
description="Test project for data consistency validation",
|
|
status="planning"
|
|
)
|
|
db.add(test_project)
|
|
db.commit()
|
|
db.refresh(test_project)
|
|
|
|
# Create a test episode
|
|
test_episode = Episode(
|
|
project_id=test_project.id,
|
|
name=f"Test Episode {unique_id}",
|
|
description="Test episode for consistency validation",
|
|
episode_number=1
|
|
)
|
|
db.add(test_episode)
|
|
db.commit()
|
|
db.refresh(test_episode)
|
|
|
|
# Create test shots
|
|
test_shots = []
|
|
for i in range(3):
|
|
shot = Shot(
|
|
project_id=test_project.id,
|
|
episode_id=test_episode.id,
|
|
name=f"shot_{unique_id}_{i+1:03d}",
|
|
description=f"Test shot {i+1}",
|
|
frame_start=1001,
|
|
frame_end=1100
|
|
)
|
|
db.add(shot)
|
|
test_shots.append(shot)
|
|
|
|
# Create test assets
|
|
test_assets = []
|
|
for i in range(2):
|
|
asset = Asset(
|
|
project_id=test_project.id,
|
|
name=f"test_asset_{unique_id}_{i+1}",
|
|
description=f"Test asset {i+1}",
|
|
category=AssetCategory.CHARACTERS
|
|
)
|
|
db.add(asset)
|
|
test_assets.append(asset)
|
|
|
|
db.commit()
|
|
|
|
# Refresh all objects
|
|
for shot in test_shots:
|
|
db.refresh(shot)
|
|
for asset in test_assets:
|
|
db.refresh(asset)
|
|
|
|
# Create test tasks for shots
|
|
shot_tasks = []
|
|
for shot in test_shots:
|
|
for task_type in ["layout", "animation", "lighting"]:
|
|
task = Task(
|
|
project_id=test_project.id,
|
|
episode_id=test_episode.id,
|
|
shot_id=shot.id,
|
|
task_type=task_type,
|
|
name=f"{shot.name}_{task_type}",
|
|
description=f"{task_type} task for {shot.name}",
|
|
status="not_started",
|
|
assigned_user_id=test_user.id
|
|
)
|
|
db.add(task)
|
|
shot_tasks.append(task)
|
|
|
|
# Create test tasks for assets
|
|
asset_tasks = []
|
|
for asset in test_assets:
|
|
for task_type in ["modeling", "surfacing", "rigging"]:
|
|
task = Task(
|
|
project_id=test_project.id,
|
|
asset_id=asset.id,
|
|
task_type=task_type,
|
|
name=f"{asset.name}_{task_type}",
|
|
description=f"{task_type} task for {asset.name}",
|
|
status="not_started",
|
|
assigned_user_id=test_user.id
|
|
)
|
|
db.add(task)
|
|
asset_tasks.append(task)
|
|
|
|
db.commit()
|
|
|
|
# Refresh all tasks
|
|
for task in shot_tasks + asset_tasks:
|
|
db.refresh(task)
|
|
|
|
return {
|
|
'user': test_user,
|
|
'project': test_project,
|
|
'episode': test_episode,
|
|
'shots': test_shots,
|
|
'assets': test_assets,
|
|
'shot_tasks': shot_tasks,
|
|
'asset_tasks': asset_tasks
|
|
}
|
|
|
|
|
|
def test_consistency_validation(db: Session, test_data: dict):
|
|
"""Test data consistency validation functionality."""
|
|
print("\n=== Testing Data Consistency Validation ===")
|
|
|
|
consistency_service = create_data_consistency_service(db)
|
|
|
|
# Test shot consistency validation
|
|
print("\n1. Testing shot consistency validation...")
|
|
for shot in test_data['shots']:
|
|
result = consistency_service.validate_task_aggregation_consistency(shot.id, 'shot')
|
|
print(f"Shot {shot.name} consistency: {'✓ VALID' if result['valid'] else '✗ INVALID'}")
|
|
if not result['valid']:
|
|
print(f" Inconsistencies: {len(result['inconsistencies'])}")
|
|
for inconsistency in result['inconsistencies']:
|
|
print(f" - {inconsistency}")
|
|
|
|
# Test asset consistency validation
|
|
print("\n2. Testing asset consistency validation...")
|
|
for asset in test_data['assets']:
|
|
result = consistency_service.validate_task_aggregation_consistency(asset.id, 'asset')
|
|
print(f"Asset {asset.name} consistency: {'✓ VALID' if result['valid'] else '✗ INVALID'}")
|
|
if not result['valid']:
|
|
print(f" Inconsistencies: {len(result['inconsistencies'])}")
|
|
for inconsistency in result['inconsistencies']:
|
|
print(f" - {inconsistency}")
|
|
|
|
# Test bulk consistency validation
|
|
print("\n3. Testing bulk consistency validation...")
|
|
shot_ids = [shot.id for shot in test_data['shots']]
|
|
asset_ids = [asset.id for asset in test_data['assets']]
|
|
|
|
shot_bulk_result = consistency_service.validate_bulk_consistency(shot_ids, 'shot')
|
|
print(f"Bulk shot validation: {shot_bulk_result['valid_entities']}/{shot_bulk_result['total_entities']} valid")
|
|
|
|
asset_bulk_result = consistency_service.validate_bulk_consistency(asset_ids, 'asset')
|
|
print(f"Bulk asset validation: {asset_bulk_result['valid_entities']}/{asset_bulk_result['total_entities']} valid")
|
|
|
|
return True
|
|
|
|
|
|
def test_real_time_updates(db: Session, test_data: dict):
|
|
"""Test real-time update propagation functionality."""
|
|
print("\n=== Testing Real-time Update Propagation ===")
|
|
|
|
consistency_service = create_data_consistency_service(db)
|
|
|
|
# Test task status update propagation
|
|
print("\n1. Testing task status update propagation...")
|
|
|
|
# Update a shot task status
|
|
shot_task = test_data['shot_tasks'][0] # First shot task
|
|
old_status = shot_task.status
|
|
new_status = "in_progress"
|
|
|
|
print(f"Updating task {shot_task.name} from '{old_status}' to '{new_status}'")
|
|
shot_task.status = new_status
|
|
db.commit()
|
|
|
|
# Propagate the update
|
|
propagation_result = consistency_service.propagate_task_update(
|
|
task_id=shot_task.id,
|
|
old_status=old_status,
|
|
new_status=new_status
|
|
)
|
|
|
|
print(f"Propagation result: {'✓ SUCCESS' if propagation_result['success'] else '✗ FAILED'}")
|
|
if propagation_result['success']:
|
|
validation_result = propagation_result['validation_result']
|
|
print(f"Post-update consistency: {'✓ VALID' if validation_result['valid'] else '✗ INVALID'}")
|
|
if not validation_result['valid']:
|
|
print(f" Inconsistencies found: {len(validation_result['inconsistencies'])}")
|
|
|
|
# Test asset task status update propagation
|
|
print("\n2. Testing asset task status update propagation...")
|
|
|
|
asset_task = test_data['asset_tasks'][0] # First asset task
|
|
old_status = asset_task.status
|
|
new_status = "submitted"
|
|
|
|
print(f"Updating task {asset_task.name} from '{old_status}' to '{new_status}'")
|
|
asset_task.status = new_status
|
|
db.commit()
|
|
|
|
# Propagate the update
|
|
propagation_result = consistency_service.propagate_task_update(
|
|
task_id=asset_task.id,
|
|
old_status=old_status,
|
|
new_status=new_status
|
|
)
|
|
|
|
print(f"Propagation result: {'✓ SUCCESS' if propagation_result['success'] else '✗ FAILED'}")
|
|
if propagation_result['success']:
|
|
validation_result = propagation_result['validation_result']
|
|
print(f"Post-update consistency: {'✓ VALID' if validation_result['valid'] else '✗ INVALID'}")
|
|
if not validation_result['valid']:
|
|
print(f" Inconsistencies found: {len(validation_result['inconsistencies'])}")
|
|
|
|
return True
|
|
|
|
|
|
def test_consistency_reporting(db: Session, test_data: dict):
|
|
"""Test consistency reporting functionality."""
|
|
print("\n=== Testing Consistency Reporting ===")
|
|
|
|
consistency_service = create_data_consistency_service(db)
|
|
|
|
# Generate project-specific consistency report
|
|
print("\n1. Generating project consistency report...")
|
|
project_report = consistency_service.get_consistency_report(test_data['project'].id)
|
|
|
|
print(f"Project {test_data['project'].name} Consistency Report:")
|
|
print(f" Total entities: {project_report['summary']['total_entities']}")
|
|
print(f" Valid entities: {project_report['summary']['valid_entities']}")
|
|
print(f" Invalid entities: {project_report['summary']['invalid_entities']}")
|
|
print(f" Total inconsistencies: {project_report['summary']['total_inconsistencies']}")
|
|
print(f" Consistency percentage: {project_report['summary']['consistency_percentage']:.1f}%")
|
|
|
|
# Generate system-wide consistency report (only for our test project to avoid enum issues)
|
|
print("\n2. Generating system-wide consistency report...")
|
|
system_report = consistency_service.get_consistency_report(test_data['project'].id)
|
|
|
|
print(f"System-wide Consistency Report:")
|
|
print(f" Total entities: {system_report['summary']['total_entities']}")
|
|
print(f" Valid entities: {system_report['summary']['valid_entities']}")
|
|
print(f" Invalid entities: {system_report['summary']['invalid_entities']}")
|
|
print(f" Total inconsistencies: {system_report['summary']['total_inconsistencies']}")
|
|
print(f" Consistency percentage: {system_report['summary']['consistency_percentage']:.1f}%")
|
|
|
|
return True
|
|
|
|
|
|
def cleanup_test_data(db: Session, test_data: dict):
|
|
"""Clean up test data."""
|
|
print("\nCleaning up test data...")
|
|
|
|
# Delete tasks
|
|
for task in test_data['shot_tasks'] + test_data['asset_tasks']:
|
|
db.delete(task)
|
|
|
|
# Delete shots and assets
|
|
for shot in test_data['shots']:
|
|
db.delete(shot)
|
|
for asset in test_data['assets']:
|
|
db.delete(asset)
|
|
|
|
# Delete episode and project
|
|
db.delete(test_data['episode'])
|
|
db.delete(test_data['project'])
|
|
db.delete(test_data['user'])
|
|
|
|
db.commit()
|
|
print("Test data cleaned up successfully.")
|
|
|
|
|
|
def main():
|
|
"""Main test function."""
|
|
print("=== Data Consistency and Real-time Updates Test ===")
|
|
print(f"Test started at: {datetime.now()}")
|
|
|
|
# Create database session
|
|
db = next(get_db())
|
|
|
|
try:
|
|
# Set up test data
|
|
test_data = setup_test_data(db)
|
|
print(f"✓ Test data created successfully")
|
|
|
|
# Run consistency validation tests
|
|
test_consistency_validation(db, test_data)
|
|
print(f"✓ Consistency validation tests completed")
|
|
|
|
# Run real-time update tests
|
|
test_real_time_updates(db, test_data)
|
|
print(f"✓ Real-time update tests completed")
|
|
|
|
# Run consistency reporting tests
|
|
test_consistency_reporting(db, test_data)
|
|
print(f"✓ Consistency reporting tests completed")
|
|
|
|
print(f"\n=== All Tests Completed Successfully ===")
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed with error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
finally:
|
|
# Clean up test data
|
|
try:
|
|
cleanup_test_data(db, test_data)
|
|
except Exception as e:
|
|
print(f"Warning: Failed to clean up test data: {str(e)}")
|
|
|
|
db.close()
|
|
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1) |