303 lines
10 KiB
Python
303 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Calejo Control Adapter - Detailed Test Runner
|
|
|
|
This script runs all tests with detailed output and coverage reports.
|
|
It creates a temporary SQLite database for integration tests.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import sqlite3
|
|
import subprocess
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
# Colors for output
|
|
class Colors:
|
|
RED = '\033[91m'
|
|
GREEN = '\033[92m'
|
|
YELLOW = '\033[93m'
|
|
BLUE = '\033[94m'
|
|
MAGENTA = '\033[95m'
|
|
CYAN = '\033[96m'
|
|
WHITE = '\033[97m'
|
|
BOLD = '\033[1m'
|
|
END = '\033[0m'
|
|
|
|
def print_color(color, message):
|
|
print(f"{color}{message}{Colors.END}")
|
|
|
|
def print_info(message):
|
|
print_color(Colors.BLUE, f"[INFO] {message}")
|
|
|
|
def print_success(message):
|
|
print_color(Colors.GREEN, f"[SUCCESS] {message}")
|
|
|
|
def print_warning(message):
|
|
print_color(Colors.YELLOW, f"[WARNING] {message}")
|
|
|
|
def print_error(message):
|
|
print_color(Colors.RED, f"[ERROR] {message}")
|
|
|
|
def print_header(message):
|
|
print_color(Colors.CYAN + Colors.BOLD, f"\n{'='*60}")
|
|
print_color(Colors.CYAN + Colors.BOLD, f" {message}")
|
|
print_color(Colors.CYAN + Colors.BOLD, f"{'='*60}\n")
|
|
|
|
def create_sqlite_test_db():
|
|
"""Create a temporary SQLite database for integration tests."""
|
|
temp_db = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
temp_db.close()
|
|
|
|
conn = sqlite3.connect(temp_db.name)
|
|
cursor = conn.cursor()
|
|
|
|
# Create tables
|
|
cursor.execute("""
|
|
CREATE TABLE stations (
|
|
station_id TEXT PRIMARY KEY,
|
|
station_name TEXT,
|
|
location TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE pumps (
|
|
station_id TEXT,
|
|
pump_id TEXT,
|
|
pump_name TEXT,
|
|
control_type TEXT,
|
|
min_speed_hz REAL DEFAULT 20.0,
|
|
max_speed_hz REAL DEFAULT 60.0,
|
|
default_setpoint_hz REAL DEFAULT 35.0,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (station_id, pump_id),
|
|
FOREIGN KEY (station_id) REFERENCES stations(station_id)
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE pump_plans (
|
|
plan_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
station_id TEXT,
|
|
pump_id TEXT,
|
|
target_flow_m3h REAL,
|
|
target_power_kw REAL,
|
|
target_level_m REAL,
|
|
suggested_speed_hz REAL,
|
|
interval_start TIMESTAMP,
|
|
interval_end TIMESTAMP,
|
|
plan_version INTEGER,
|
|
plan_status TEXT DEFAULT 'ACTIVE',
|
|
plan_created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
plan_updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
optimization_run_id TEXT,
|
|
FOREIGN KEY (station_id, pump_id) REFERENCES pumps(station_id, pump_id)
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE pump_feedback (
|
|
feedback_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
station_id TEXT,
|
|
pump_id TEXT,
|
|
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
actual_speed_hz REAL,
|
|
actual_power_kw REAL,
|
|
actual_flow_m3h REAL,
|
|
wet_well_level_m REAL,
|
|
pump_running BOOLEAN,
|
|
alarm_active BOOLEAN,
|
|
alarm_code TEXT,
|
|
FOREIGN KEY (station_id, pump_id) REFERENCES pumps(station_id, pump_id)
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE emergency_stop_events (
|
|
event_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
triggered_by TEXT,
|
|
reason TEXT,
|
|
station_id TEXT,
|
|
pump_id TEXT,
|
|
event_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
cleared_by TEXT,
|
|
cleared_timestamp TIMESTAMP,
|
|
cleared_notes TEXT,
|
|
FOREIGN KEY (station_id, pump_id) REFERENCES pumps(station_id, pump_id)
|
|
)
|
|
""")
|
|
|
|
# Insert test data
|
|
cursor.execute("""
|
|
INSERT INTO stations (station_id, station_name, location) VALUES
|
|
('STATION_001', 'Main Pump Station', 'Downtown Area'),
|
|
('STATION_002', 'Secondary Station', 'Industrial Zone')
|
|
""")
|
|
|
|
cursor.execute("""
|
|
INSERT INTO pumps (station_id, pump_id, pump_name, control_type, min_speed_hz, max_speed_hz, default_setpoint_hz) VALUES
|
|
('STATION_001', 'PUMP_001', 'Main Pump 1', 'DIRECT_SPEED', 20.0, 60.0, 35.0),
|
|
('STATION_001', 'PUMP_002', 'Main Pump 2', 'LEVEL_CONTROLLED', 20.0, 60.0, 35.0),
|
|
('STATION_002', 'PUMP_001', 'Secondary Pump 1', 'POWER_CONTROLLED', 20.0, 60.0, 35.0)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
INSERT INTO pump_plans (
|
|
station_id, pump_id, target_flow_m3h, target_power_kw, target_level_m,
|
|
suggested_speed_hz, interval_start, interval_end, plan_version, optimization_run_id
|
|
) VALUES
|
|
('STATION_001', 'PUMP_001', 150.0, NULL, NULL, 42.5,
|
|
datetime('now', '-1 hour'), datetime('now', '+1 hour'), 1, 'OPT_RUN_001'),
|
|
('STATION_001', 'PUMP_002', NULL, NULL, 2.5, 38.0,
|
|
datetime('now', '-1 hour'), datetime('now', '+1 hour'), 1, 'OPT_RUN_001'),
|
|
('STATION_002', 'PUMP_001', NULL, 18.5, NULL, 40.0,
|
|
datetime('now', '-1 hour'), datetime('now', '+1 hour'), 1, 'OPT_RUN_001')
|
|
""")
|
|
|
|
cursor.execute("""
|
|
INSERT INTO pump_feedback (
|
|
station_id, pump_id, actual_speed_hz, actual_power_kw, actual_flow_m3h,
|
|
wet_well_level_m, pump_running, alarm_active
|
|
) VALUES
|
|
('STATION_001', 'PUMP_001', 42.5, 16.2, 148.5, 1.8, 1, 0),
|
|
('STATION_001', 'PUMP_002', 38.0, 14.8, 135.2, 2.3, 1, 0),
|
|
('STATION_002', 'PUMP_001', 40.0, 18.3, 142.1, 1.9, 1, 0)
|
|
""")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
return temp_db.name
|
|
|
|
def run_tests(test_path, coverage_dir, test_type):
|
|
"""Run tests with detailed output and coverage."""
|
|
print_header(f"RUNNING {test_type.upper()} TESTS")
|
|
|
|
cmd = [
|
|
'python', '-m', 'pytest', test_path,
|
|
'-v',
|
|
'--tb=long',
|
|
'--cov=src',
|
|
'--cov-report=term-missing',
|
|
f'--cov-report=html:{coverage_dir}',
|
|
'--durations=10', # Show 10 slowest tests
|
|
'--color=yes'
|
|
]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
|
|
# Print output
|
|
print(result.stdout)
|
|
if result.stderr:
|
|
print_color(Colors.YELLOW, f"STDERR:\n{result.stderr}")
|
|
|
|
return result.returncode, result.stdout
|
|
|
|
def main():
|
|
"""Main test runner function."""
|
|
print_header("CALEJO CONTROL ADAPTER - COMPREHENSIVE TEST RUNNER")
|
|
|
|
# Create coverage directories
|
|
coverage_dirs = {
|
|
'unit': 'htmlcov_unit',
|
|
'integration': 'htmlcov_integration',
|
|
'combined': 'htmlcov_combined'
|
|
}
|
|
|
|
for dir_name in coverage_dirs.values():
|
|
if os.path.exists(dir_name):
|
|
shutil.rmtree(dir_name)
|
|
|
|
# Create SQLite test database
|
|
print_info("Creating SQLite test database...")
|
|
test_db_path = create_sqlite_test_db()
|
|
|
|
# Set environment variables
|
|
os.environ['TEST_DATABASE_URL'] = f'sqlite:///{test_db_path}'
|
|
os.environ['TEST_MODE'] = 'true'
|
|
|
|
print_success(f"Test database created: {test_db_path}")
|
|
|
|
# Run unit tests
|
|
unit_exit_code, unit_output = run_tests(
|
|
'tests/unit/',
|
|
coverage_dirs['unit'],
|
|
'UNIT'
|
|
)
|
|
|
|
# Run integration tests
|
|
integration_exit_code, integration_output = run_tests(
|
|
'tests/integration/',
|
|
coverage_dirs['integration'],
|
|
'INTEGRATION'
|
|
)
|
|
|
|
# Generate combined coverage report
|
|
print_header("GENERATING COMBINED COVERAGE REPORT")
|
|
|
|
cmd = [
|
|
'python', '-m', 'pytest',
|
|
'--cov=src',
|
|
'--cov-report=html:htmlcov_combined',
|
|
'--cov-report=term-missing',
|
|
'tests/'
|
|
]
|
|
|
|
subprocess.run(cmd)
|
|
|
|
# Clean up test database
|
|
os.unlink(test_db_path)
|
|
print_success("Test database cleaned up")
|
|
|
|
# Generate test summary
|
|
print_header("TEST RESULTS SUMMARY")
|
|
|
|
# Count tests from output
|
|
def count_tests(output):
|
|
passed = output.count('PASSED')
|
|
failed = output.count('FAILED')
|
|
errors = output.count('ERROR')
|
|
skipped = output.count('SKIPPED')
|
|
return passed, failed, errors, skipped
|
|
|
|
unit_passed, unit_failed, unit_errors, unit_skipped = count_tests(unit_output)
|
|
integration_passed, integration_failed, integration_errors, integration_skipped = count_tests(integration_output)
|
|
|
|
total_passed = unit_passed + integration_passed
|
|
total_failed = unit_failed + integration_failed
|
|
total_errors = unit_errors + integration_errors
|
|
total_skipped = unit_skipped + integration_skipped
|
|
total_tests = total_passed + total_failed + total_errors + total_skipped
|
|
|
|
print_info(f"Unit Tests: {unit_passed} passed, {unit_failed} failed, {unit_errors} errors, {unit_skipped} skipped")
|
|
print_info(f"Integration Tests: {integration_passed} passed, {integration_failed} failed, {integration_errors} errors, {integration_skipped} skipped")
|
|
print_info(f"Total Tests: {total_passed} passed, {total_failed} failed, {total_errors} errors, {total_skipped} skipped")
|
|
|
|
if unit_exit_code == 0 and integration_exit_code == 0:
|
|
print_success("🎉 ALL TESTS PASSED! 🎉")
|
|
print_info(f"Coverage reports generated in: {', '.join(coverage_dirs.values())}")
|
|
return 0
|
|
else:
|
|
print_error("❌ SOME TESTS FAILED ❌")
|
|
if unit_exit_code != 0:
|
|
print_error(f"Unit tests failed with exit code: {unit_exit_code}")
|
|
if integration_exit_code != 0:
|
|
print_error(f"Integration tests failed with exit code: {integration_exit_code}")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
exit_code = main()
|
|
sys.exit(exit_code)
|
|
except KeyboardInterrupt:
|
|
print_warning("\nTest run interrupted by user")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print_error(f"Unexpected error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1) |