Spaces:
Sleeping
Sleeping
| import unittest | |
| from unittest.mock import MagicMock, patch | |
| import sys | |
| import os | |
| import threading | |
| # Add src to path | |
| sys.path.append(os.getcwd()) | |
| from src.drone.connection_manager import DroneConnectionManager, PathType, PathStatus | |
| from src.model_manager import ModelManager, ModelType | |
| from morphguard_api import MorphGuardAPI | |
| class TestSystemIntegration(unittest.TestCase): | |
| def setUp(self): | |
| self.cm = DroneConnectionManager() | |
| def test_drone_connection_passive_mode(self): | |
| print("\n[Test] Verifying Drone Passive Connection Logic...") | |
| # 1. Register external connection (Simulating src/routes/drone.py) | |
| mock_ws = MagicMock() | |
| remote_ip = "192.168.1.105" | |
| # Ensure path exists before registering (standard init does this, but being safe) | |
| if PathType.PI_WIFI not in self.cm.paths: | |
| self.cm.register_path(PathType.PI_WIFI, "ws://placeholder") | |
| result = self.cm.register_external_connection(PathType.PI_WIFI, mock_ws, remote_ip) | |
| # Verify Registration | |
| self.assertTrue(result, "Should yield successful registration") | |
| path = self.cm.paths[PathType.PI_WIFI] | |
| self.assertEqual(path.health.status, PathStatus.CONNECTED) | |
| self.assertEqual(path.connection, mock_ws) | |
| self.assertEqual(path.uri, f"ws://{remote_ip}") | |
| self.assertIn(PathType.PI_WIFI, self.cm.active_paths) | |
| print("✓ Passive connection registered successfully") | |
| print(f" Status: {path.health.status}") | |
| print(f" URI Updated: {path.uri}") | |
| def test_model_auto_detection(self, mock_exists): | |
| print("\n[Test] Verifying Model Path Auto-Detection...") | |
| # Simulating that models exist in the root models/ dir | |
| def side_effect(path): | |
| if path == "models/morph_detector.pth": return True | |
| if path == "models/demorpher.pth": return True | |
| return False | |
| mock_exists.side_effect = side_effect | |
| # Initialize API without explicit paths | |
| # Note: We mock get_model_manager to avoid real singleton state issues | |
| with patch('src.model_manager.get_model_manager') as mock_mm_getter: | |
| mock_mm = MagicMock() | |
| mock_mm.device = MagicMock() | |
| mock_mm_getter.return_value = mock_mm | |
| api = MorphGuardAPI() | |
| # Check if it auto-detected the paths | |
| self.assertEqual(api.config["detector_path"], "models/morph_detector.pth") | |
| self.assertEqual(api.config["demorph_path"], "models/demorpher.pth") | |
| print(f"✓ Detector path detected: {api.config['detector_path']}") | |
| print(f"✓ Demorpher path detected: {api.config['demorph_path']}") | |
| def test_model_manager_lazy_loading(self): | |
| print("\n[Test] Verifying Smart Model Manager Lazy Loading...") | |
| mm = ModelManager() | |
| # Reset for test | |
| mm.unload_all() | |
| # Verify no models loaded initially | |
| self.assertEqual(len(mm.models), 0, "ModelManager should start empty") | |
| print("✓ Manager starts empty") | |
| # Mock loader | |
| mock_loader = MagicMock(return_value="FAKE_MODEL_INSTANCE") | |
| # Request model | |
| model = mm.get_model(ModelType.DETECTOR, mock_loader) | |
| # Verify loaded | |
| self.assertEqual(model, "FAKE_MODEL_INSTANCE") | |
| self.assertIn(ModelType.DETECTOR, mm.models) | |
| mock_loader.assert_called_once() | |
| print("✓ Model loaded on demand (Lazy Loading)") | |
| # Request again (should rely on cache) | |
| model2 = mm.get_model(ModelType.DETECTOR, mock_loader) | |
| mock_loader.assert_called_once() # Call count shouldn't increase | |
| print("✓ Cached model used on second request") | |
| # Verify Unloading | |
| mm.unload_model(ModelType.DETECTOR) | |
| self.assertNotIn(ModelType.DETECTOR, mm.models) | |
| print("✓ Model unloaded successfully") | |
| if __name__ == '__main__': | |
| unittest.main() | |