114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test processing state transitions
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
import time
|
|
import threading
|
|
|
|
# Add project root to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from backend.utils.state_manager import state_manager
|
|
from backend.models.data_models import UploadedFile, MeshResult
|
|
from datetime import datetime
|
|
|
|
def test_processing_states():
|
|
"""Test processing state transitions"""
|
|
print("Testing processing state transitions...")
|
|
|
|
# Reset state
|
|
state_manager.clear_current_file()
|
|
|
|
# 1. Create a mock uploaded file
|
|
print("\n1. Creating mock uploaded file...")
|
|
mock_file = UploadedFile(
|
|
id="test-123",
|
|
filename="test_blade.step",
|
|
file_path="/path/to/test.step",
|
|
upload_time=datetime.now(),
|
|
status="UPLOADED"
|
|
)
|
|
state_manager.set_current_file(mock_file)
|
|
|
|
# Check initial state
|
|
system_state = state_manager.get_system_state()
|
|
print(f"✓ File uploaded, ready for processing: {system_state['is_ready_for_processing']}")
|
|
|
|
# 2. Start processing
|
|
print("\n2. Starting processing...")
|
|
state_manager.start_processing("Starting mesh generation...")
|
|
|
|
system_state = state_manager.get_system_state()
|
|
print(f"✓ Processing started: {system_state['is_processing']}")
|
|
print(f"✓ Processing status: {system_state['processing_status']['status']}")
|
|
print(f"✓ Start time: {system_state['processing_status']['start_time']}")
|
|
|
|
# 3. Simulate processing time
|
|
print("\n3. Simulating processing...")
|
|
time.sleep(1) # Simulate 1 second of processing
|
|
|
|
# 4. Complete processing with result
|
|
print("\n4. Completing processing...")
|
|
mock_result = MeshResult(
|
|
mesh_image_path="/path/to/mesh.png",
|
|
element_count=12345,
|
|
node_count=67890,
|
|
min_element_quality=0.85,
|
|
processing_time=1.0
|
|
)
|
|
|
|
state_manager.set_mesh_result(mock_result)
|
|
state_manager.complete_processing("Mesh generation completed successfully")
|
|
|
|
system_state = state_manager.get_system_state()
|
|
print(f"✓ Processing completed: {system_state['processing_status']['status']}")
|
|
print(f"✓ Has mesh result: {system_state['mesh_result'] is not None}")
|
|
print(f"✓ Processing time: {system_state['processing_time']} seconds")
|
|
|
|
# 5. Test error scenario
|
|
print("\n5. Testing error scenario...")
|
|
state_manager.start_processing("Starting another process...")
|
|
time.sleep(0.5)
|
|
state_manager.set_processing_error("Simulated processing error")
|
|
|
|
system_state = state_manager.get_system_state()
|
|
print(f"✓ Error status: {system_state['processing_status']['status']}")
|
|
print(f"✓ Error message: {system_state['processing_status']['error_message']}")
|
|
|
|
# 6. Test session data
|
|
print("\n6. Testing session data...")
|
|
state_manager.set_session_data("test_key", "test_value")
|
|
state_manager.set_session_data("processing_count", 5)
|
|
|
|
system_state = state_manager.get_system_state()
|
|
print(f"✓ Session data: {system_state['session_data']}")
|
|
|
|
# 7. Test thread safety
|
|
print("\n7. Testing thread safety...")
|
|
|
|
def worker_thread(thread_id):
|
|
for i in range(10):
|
|
state_manager.set_session_data(f"thread_{thread_id}_count", i)
|
|
time.sleep(0.01)
|
|
|
|
threads = []
|
|
for i in range(3):
|
|
t = threading.Thread(target=worker_thread, args=(i,))
|
|
threads.append(t)
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
system_state = state_manager.get_system_state()
|
|
thread_data = {k: v for k, v in system_state['session_data'].items() if k.startswith('thread_')}
|
|
print(f"✓ Thread safety test completed, thread data: {thread_data}")
|
|
|
|
print("\n✓ All processing state tests completed successfully!")
|
|
|
|
if __name__ == '__main__':
|
|
print("CAE Mesh Generator Processing States Test")
|
|
print("=" * 50)
|
|
test_processing_states() |