122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
"""
|
|
End-to-end workflow tests for Calejo Control Adapter
|
|
|
|
Tests basic system workflows from database operations to component integration.
|
|
"""
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
import asyncio
|
|
|
|
from src.database.flexible_client import FlexibleDatabaseClient
|
|
from src.core.auto_discovery import AutoDiscovery
|
|
from src.core.optimization_manager import OptimizationPlanManager
|
|
|
|
|
|
class TestEndToEndWorkflow:
|
|
"""Test basic system workflows."""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_client(self):
|
|
"""Create database client for testing."""
|
|
client = FlexibleDatabaseClient(
|
|
database_url="sqlite:///:memory:",
|
|
pool_size=5,
|
|
max_overflow=10,
|
|
pool_timeout=30
|
|
)
|
|
|
|
# Connect to database
|
|
await client.connect()
|
|
|
|
# Create test tables
|
|
client.create_tables()
|
|
|
|
# Insert test data
|
|
self._insert_test_data(client)
|
|
|
|
return client
|
|
|
|
def _insert_test_data(self, db_client):
|
|
"""Insert realistic test data for end-to-end testing."""
|
|
# Insert pump stations
|
|
db_client.execute(
|
|
"""INSERT INTO pump_stations (station_id, station_name, location) VALUES
|
|
('STATION_001', 'Station A', 'Location A')"""
|
|
)
|
|
|
|
# Insert pumps
|
|
db_client.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', 'Pump 1', 'DIRECT_SPEED', 20.0, 60.0, 35.0)"""
|
|
)
|
|
|
|
# Insert optimization plan
|
|
db_client.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, plan_status, optimization_run_id
|
|
) VALUES (
|
|
'STATION_001', 'PUMP_001', 150.0, NULL, NULL, 42.5,
|
|
datetime('now', '-1 hour'), datetime('now', '+1 hour'), 1, 'ACTIVE', 'OPT_RUN_001'
|
|
)"""
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_operations(self, db_client):
|
|
"""Test basic database operations."""
|
|
# Test getting pump stations
|
|
stations = db_client.get_pump_stations()
|
|
assert len(stations) == 1
|
|
assert stations[0]['station_name'] == 'Station A'
|
|
|
|
# Test getting pumps
|
|
pumps = db_client.get_pumps(station_id='STATION_001')
|
|
assert len(pumps) == 1
|
|
assert pumps[0]['pump_name'] == 'Pump 1'
|
|
|
|
# Test getting current plan
|
|
plan = db_client.get_current_plan(station_id='STATION_001', pump_id='PUMP_001')
|
|
assert plan is not None
|
|
assert plan['suggested_speed_hz'] == 42.5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_discovery_workflow(self, db_client):
|
|
"""Test auto-discovery workflow."""
|
|
auto_discovery = AutoDiscovery(db_client)
|
|
|
|
# Run auto-discovery
|
|
await auto_discovery.discover()
|
|
|
|
# Auto-discovery should find stations and pumps
|
|
stations = auto_discovery.get_stations()
|
|
assert len(stations) == 1
|
|
assert stations['STATION_001']['station_name'] == 'Station A'
|
|
|
|
pumps = auto_discovery.get_pumps(station_id='STATION_001')
|
|
assert len(pumps) == 1
|
|
assert pumps[0]['pump_name'] == 'Pump 1'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_optimization_workflow(self, db_client):
|
|
"""Test optimization workflow."""
|
|
# Test that we can query the pump_plans table directly
|
|
plans = db_client.execute_query(
|
|
"""SELECT station_id, pump_id, suggested_speed_hz
|
|
FROM pump_plans
|
|
WHERE station_id = 'STATION_001' AND pump_id = 'PUMP_001'"""
|
|
)
|
|
assert len(plans) == 1
|
|
assert plans[0]['suggested_speed_hz'] == 42.5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_health_check(self, db_client):
|
|
"""Test database health monitoring."""
|
|
# Test health check
|
|
health = db_client.health_check()
|
|
assert health is True
|
|
|
|
# Test connection stats
|
|
stats = db_client.get_connection_stats()
|
|
assert 'connection_attempts' in stats
|
|
assert 'failed_connections' in stats |