#!/usr/bin/env python3 """ Test Unified Configuration This script tests that all processor configurations are properly loaded from environment variables and have sensible defaults. """ import os import sys import json import subprocess from typing import Dict, Any, List def test_environment_variable( var_name: str, expected_default: Any, test_value: Any = None ) -> bool: """Test that an environment variable has the expected default value""" # First, check if it's set current_value = os.environ.get(var_name) if current_value is None: print(f"✓ {var_name}: Not set, using default '{expected_default}'") return True # Try to convert to appropriate type try: if isinstance(expected_default, bool): actual = current_value.lower() == "true" expected = expected_default elif isinstance(expected_default, int): actual = int(current_value) expected = int(expected_default) elif isinstance(expected_default, float): actual = float(current_value) expected = float(expected_default) elif isinstance(expected_default, list): actual = [s.strip() for s in current_value.split(",") if s.strip()] expected = expected_default else: actual = current_value expected = expected_default if actual == expected: print(f"✓ {var_name}: Set to '{current_value}' (matches default)") else: print( f"✓ {var_name}: Set to '{current_value}' (different from default '{expected_default}')" ) return True except (ValueError, TypeError) as e: print(f"✗ {var_name}: Invalid value '{current_value}' - {e}") return False def test_processor_health(processor_script: str) -> bool: """Test processor health check""" try: result = subprocess.run( [ sys.executable, processor_script, "dummy.mp4", "dummy.json", "--check-health", ], capture_output=True, text=True, timeout=10, ) if result.returncode == 0: health_data = json.loads(result.stdout) if health_data.get("status") == "healthy": print(f"✓ {processor_script}: Health check passed") return True else: print( f"✗ {processor_script}: Health check failed - {health_data.get('status')}" ) return False else: print( f"✗ {processor_script}: Health check failed with exit code {result.returncode}" ) print(f" Error: {result.stderr}") return False except subprocess.TimeoutExpired: print(f"✗ {processor_script}: Health check timed out") return False except Exception as e: print(f"✗ {processor_script}: Health check error - {e}") return False def test_configuration_groups(): """Test all configuration groups""" print("\n" + "=" * 80) print("Testing Unified Configuration") print("=" * 80) # ASR Configuration print("\n1. ASR Configuration:") print("-" * 40) asr_tests = [ ("MOMENTRY_ASR_TIMEOUT", "3600"), ("MOMENTRY_ASR_PROCESS_TIMEOUT", "1800"), ("MOMENTRY_ASR_CHUNK_TIMEOUT", "300"), ("MOMENTRY_ASR_MODEL_SIZE", "medium"), ("MOMENTRY_ASR_DEVICE", "cpu"), ("MOMENTRY_ASR_LANGUAGE", "auto"), ("MOMENTRY_ASR_TASK", "transcribe"), ("MOMENTRY_ASR_BEAM_SIZE", "5"), ("MOMENTRY_ASR_BEST_OF", "5"), ] asr_passed = 0 for var_name, default in asr_tests: if test_environment_variable(var_name, default): asr_passed += 1 # OCR Configuration print("\n2. OCR Configuration:") print("-" * 40) ocr_tests = [ ("MOMENTRY_OCR_TIMEOUT", "1800"), ("MOMENTRY_OCR_LANGUAGES", "en"), ("MOMENTRY_OCR_CONFIDENCE", "0.7"), ("MOMENTRY_OCR_GPU", "false"), ("MOMENTRY_OCR_MODEL_PATH", "~/.EasyOCR/model"), ] ocr_passed = 0 for var_name, default in ocr_tests: if test_environment_variable(var_name, default): ocr_passed += 1 # YOLO Configuration print("\n3. YOLO Configuration:") print("-" * 40) yolo_tests = [ ("MOMENTRY_YOLO_TIMEOUT", "7200"), ("MOMENTRY_YOLO_MODEL_SIZE", "yolov8n.pt"), ("MOMENTRY_YOLO_CONFIDENCE", "0.25"), ("MOMENTRY_YOLO_IOU", "0.45"), ("MOMENTRY_YOLO_GPU", "false"), ("MOMENTRY_YOLO_AUTO_SAVE_INTERVAL", "30"), ("MOMENTRY_YOLO_AUTO_SAVE_FRAMES", "300"), ("MOMENTRY_YOLO_CLASSES", ""), ] yolo_passed = 0 for var_name, default in yolo_tests: if test_environment_variable(var_name, default): yolo_passed += 1 # Face Configuration print("\n4. Face Configuration:") print("-" * 40) face_tests = [ ("MOMENTRY_FACE_TIMEOUT", "3600"), ("MOMENTRY_FACE_METHOD", "haar"), ("MOMENTRY_FACE_CONFIDENCE", "0.5"), ("MOMENTRY_FACE_MIN_SIZE", "30"), ("MOMENTRY_FACE_MAX_SIZE", "300"), ("MOMENTRY_FACE_SCALE_FACTOR", "1.1"), ("MOMENTRY_FACE_MIN_NEIGHBORS", "3"), ("MOMENTRY_FACE_GPU", "false"), ] face_passed = 0 for var_name, default in face_tests: if test_environment_variable(var_name, default): face_passed += 1 # Pose Configuration print("\n5. Pose Configuration:") print("-" * 40) pose_tests = [ ("MOMENTRY_POSE_TIMEOUT", "7200"), ("MOMENTRY_POSE_MODEL_SIZE", "yolov8n-pose.pt"), ("MOMENTRY_POSE_CONFIDENCE", "0.25"), ("MOMENTRY_POSE_IOU", "0.45"), ("MOMENTRY_POSE_GPU", "false"), ("MOMENTRY_POSE_KEYPOINT_CONFIDENCE", "0.5"), ("MOMENTRY_POSE_MAX_PERSONS", "10"), ] pose_passed = 0 for var_name, default in pose_tests: if test_environment_variable(var_name, default): pose_passed += 1 # Processor Health Checks print("\n6. Processor Health Checks:") print("-" * 40) processors = [ "scripts/asr_processor_contract_v2.py", "scripts/ocr_processor_contract_v1.py", "scripts/yolo_processor_contract_v1.py", "scripts/face_processor_contract_v1.py", "scripts/pose_processor_contract_v1.py", ] health_passed = 0 for processor in processors: if os.path.exists(processor): if test_processor_health(processor): health_passed += 1 else: print(f"⚠ {processor}: Script not found") # Summary print("\n" + "=" * 80) print("SUMMARY") print("=" * 80) total_tests = ( len(asr_tests) + len(ocr_tests) + len(yolo_tests) + len(face_tests) + len(pose_tests) ) total_passed = asr_passed + ocr_passed + yolo_passed + face_passed + pose_passed print(f"ASR Configuration: {asr_passed}/{len(asr_tests)} passed") print(f"OCR Configuration: {ocr_passed}/{len(ocr_tests)} passed") print(f"YOLO Configuration: {yolo_passed}/{len(yolo_tests)} passed") print(f"Face Configuration: {face_passed}/{len(face_tests)} passed") print(f"Pose Configuration: {pose_passed}/{len(pose_tests)} passed") print(f"Processor Health: {health_passed}/{len(processors)} passed") print("-" * 40) print( f"TOTAL: {total_passed}/{total_tests} configuration tests passed" ) print( f" {health_passed}/{len(processors)} processor health checks passed" ) if total_passed == total_tests and health_passed == len(processors): print("\n✅ All tests passed!") return True else: print("\n❌ Some tests failed") return False def test_rust_config_compilation(): """Test that Rust configuration compiles correctly""" print("\n7. Rust Configuration Compilation:") print("-" * 40) try: result = subprocess.run( ["cargo", "check", "--lib"], capture_output=True, text=True, timeout=30, ) if result.returncode == 0: print("✓ Rust configuration compiles successfully") return True else: print(f"✗ Rust configuration compilation failed") print(f" Error: {result.stderr[:500]}...") return False except subprocess.TimeoutExpired: print("✗ Rust configuration check timed out") return False except Exception as e: print(f"✗ Rust configuration check error - {e}") return False def main(): """Main test function""" print("Momentry Core - Unified Configuration Test Suite") print("=" * 80) # Run configuration tests config_ok = test_configuration_groups() # Run Rust compilation test rust_ok = test_rust_config_compilation() # Final result print("\n" + "=" * 80) print("FINAL RESULT") print("=" * 80) if config_ok and rust_ok: print("✅ All tests passed! Unified configuration is working correctly.") return 0 else: print("❌ Some tests failed. Please check the errors above.") return 1 if __name__ == "__main__": sys.exit(main())