calejocontrol
/
CalejoControl
1
0
Fork 0
Code Issues Pull Requests 4 Packages Projects Releases Wiki Activity

Clean up test structure and improve test runner 

- Renamed run_tests_with_better_output.py to run_tests_by_system.py (more descriptive)
- Removed legacy test_phase1.py file (no tests collected)
- Updated test sections to reflect current test structure
- Test runner now organizes tests by system/component with timing
- All 197 tests passing

Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
openhands 2025-10-28 10:25:00 +00:00
parent 58ba34b230
commit 84edcb14ff
2 changed files with 1 additions and 262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
run_tests_with_better_output.py → run_tests_by_system.py
View File

@ -55,7 +55,7 @@ def main():
("UNIT TESTS", "tests/unit/"), ("UNIT TESTS", "tests/unit/"),
("INTEGRATION TESTS", "tests/integration/"), ("INTEGRATION TESTS", "tests/integration/"),
("SAFETY FRAMEWORK TESTS", "tests/test_safety.py"), ("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 all_passed = True

261
tests/test_phase1.py
View File

@ -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())