#!/usr/bin/env python3 """ Test script for Phase 1 implementation. Tests core infrastructure components: - Database connectivity and queries - Auto-discovery functionality - Configuration management - Safety framework loading """ import asyncio import sys import os # Add src to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from src.database.client import DatabaseClient from src.core.auto_discovery import AutoDiscovery from src.core.safety import SafetyLimitEnforcer from src.core.logging import setup_logging from config.settings import settings class Phase1Tester: """Test class for Phase 1 components.""" def __init__(self): self.logger = setup_logging() self.db_client = DatabaseClient( database_url=settings.database_url, min_connections=settings.db_min_connections, max_connections=settings.db_max_connections ) self.auto_discovery = AutoDiscovery(self.db_client) self.safety_enforcer = SafetyLimitEnforcer(self.db_client) self.tests_passed = 0 self.tests_failed = 0 async def run_all_tests(self): """Run all Phase 1 tests.""" self.logger.info("starting_phase1_tests") print("\n" + "="*60) print("Calejo Control Adapter - Phase 1 Test Suite") print("="*60) try: # Test 1: Database connection await self.test_database_connection() # Test 2: Database queries await self.test_database_queries() # Test 3: Auto-discovery await self.test_auto_discovery() # Test 4: Safety framework await self.test_safety_framework() # Test 5: Configuration await self.test_configuration() # Print summary self.print_test_summary() except Exception as e: self.logger.error("test_suite_failed", error=str(e)) raise finally: await self.db_client.disconnect() async def test_database_connection(self): """Test database connectivity.""" print("\n1. Testing Database Connection...") try: await self.db_client.connect() # Test health check is_healthy = self.db_client.health_check() if is_healthy: print(" ✓ Database connection successful") self.tests_passed += 1 else: print(" ✗ Database health check failed") self.tests_failed += 1 except Exception as e: print(f" ✗ Database connection failed: {e}") self.tests_failed += 1 raise async def test_database_queries(self): """Test database queries.""" print("\n2. Testing Database Queries...") try: # Test pump stations query stations = self.db_client.get_pump_stations() print(f" ✓ Found {len(stations)} pump stations") # Test pumps query pumps = self.db_client.get_pumps() print(f" ✓ Found {len(pumps)} pumps") # Test safety limits query safety_limits = self.db_client.get_safety_limits() print(f" ✓ Found {len(safety_limits)} safety limits") # Test pump plans query pump_plans = self.db_client.get_latest_pump_plans() print(f" ✓ Found {len(pump_plans)} active pump plans") self.tests_passed += 1 except Exception as e: print(f" ✗ Database queries failed: {e}") self.tests_failed += 1 raise async def test_auto_discovery(self): """Test auto-discovery functionality.""" print("\n3. Testing Auto-Discovery...") try: await self.auto_discovery.discover() stations = self.auto_discovery.get_stations() pumps = self.auto_discovery.get_pumps() print(f" ✓ Discovered {len(stations)} stations") print(f" ✓ Discovered {len(pumps)} pumps") # Test individual station/pump retrieval if stations: station_id = list(stations.keys())[0] station = self.auto_discovery.get_station(station_id) if station: print(f" ✓ Station retrieval successful: {station['station_name']}") station_pumps = self.auto_discovery.get_pumps(station_id) if station_pumps: pump_id = station_pumps[0]['pump_id'] pump = self.auto_discovery.get_pump(station_id, pump_id) if pump: print(f" ✓ Pump retrieval successful: {pump['pump_name']}") # Test validation validation = self.auto_discovery.validate_discovery() if validation['valid']: print(" ✓ Auto-discovery validation passed") else: print(f" ⚠ Auto-discovery validation issues: {validation['issues']}") self.tests_passed += 1 except Exception as e: print(f" ✗ Auto-discovery failed: {e}") self.tests_failed += 1 raise async def test_safety_framework(self): """Test safety framework loading.""" print("\n4. Testing Safety Framework...") try: await self.safety_enforcer.load_safety_limits() limits_count = len(self.safety_enforcer.safety_limits_cache) print(f" ✓ Loaded {limits_count} safety limits") # Test setpoint enforcement if limits_count > 0: # Get first pump with safety limits pumps = self.auto_discovery.get_pumps() if pumps: pump = pumps[0] station_id = pump['station_id'] pump_id = pump['pump_id'] # Test within limits enforced, violations = self.safety_enforcer.enforce_setpoint( station_id, pump_id, 35.0 ) if enforced == 35.0 and not violations: print(" ✓ Setpoint enforcement within limits successful") # Test below minimum enforced, violations = self.safety_enforcer.enforce_setpoint( station_id, pump_id, 10.0 ) if enforced > 10.0 and violations: print(" ✓ Setpoint enforcement below minimum successful") self.tests_passed += 1 except Exception as e: print(f" ✗ Safety framework failed: {e}") self.tests_failed += 1 raise async def test_configuration(self): """Test configuration management.""" print("\n5. Testing Configuration Management...") try: # Test database URL generation db_url = settings.database_url if db_url: print(" ✓ Database URL generation successful") # Test safe settings dict safe_settings = settings.get_safe_dict() if 'db_password' in safe_settings and safe_settings['db_password'] == '***MASKED***': print(" ✓ Sensitive field masking successful") # Test configuration validation print(f" ✓ Configuration loaded: {settings.app_name} v{settings.app_version}") print(f" ✓ Environment: {settings.environment}") self.tests_passed += 1 except Exception as e: print(f" ✗ Configuration test failed: {e}") self.tests_failed += 1 raise def print_test_summary(self): """Print test summary.""" print("\n" + "="*60) print("TEST SUMMARY") print("="*60) print(f"Tests Passed: {self.tests_passed}") print(f"Tests Failed: {self.tests_failed}") total_tests = self.tests_passed + self.tests_failed if total_tests > 0: success_rate = (self.tests_passed / total_tests) * 100 print(f"Success Rate: {success_rate:.1f}%") if self.tests_failed == 0: print("\n🎉 All Phase 1 tests passed!") print("Phase 1 implementation is ready for development.") else: print(f"\n⚠ {self.tests_failed} test(s) failed.") print("Please review the failed tests before proceeding.") print("="*60) async def main(): """Run Phase 1 tests.""" tester = Phase1Tester() await tester.run_all_tests() if __name__ == "__main__": asyncio.run(main())