diff --git a/run_tests_with_better_output.py b/run_tests_by_system.py similarity index 97% rename from run_tests_with_better_output.py rename to run_tests_by_system.py index e2884b4..b8bbbb4 100755 --- a/run_tests_with_better_output.py +++ b/run_tests_by_system.py @@ -55,7 +55,7 @@ def main(): ("UNIT TESTS", "tests/unit/"), ("INTEGRATION TESTS", "tests/integration/"), ("SAFETY FRAMEWORK TESTS", "tests/test_safety.py"), - ("PHASE 1 TESTS", "tests/test_phase1.py") + ("SAFETY FRAMEWORK UNIT TESTS", "tests/unit/test_safety_framework.py") ] all_passed = True diff --git a/tests/test_phase1.py b/tests/test_phase1.py deleted file mode 100755 index 2435223..0000000 --- a/tests/test_phase1.py +++ /dev/null @@ -1,261 +0,0 @@ -#!/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.flexible_client import FlexibleDatabaseClient -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 = FlexibleDatabaseClient( - database_url=settings.database_url, - pool_size=settings.db_pool_size, - max_overflow=settings.db_max_overflow - ) - 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()) \ No newline at end of file