b54c2def30
- Add database migrations (006-028) for face recognition, identity, file_uuid - Add test scripts for ASR, face, search, processing - Add portal frontend (Tauri) - Add config, benchmark, and monitoring utilities - Add model checkpoints and pretrained model references
305 lines
9.3 KiB
Python
305 lines
9.3 KiB
Python
#!/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())
|