379 lines
12 KiB
Python
379 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Standalone test script for mock SCADA and optimizer services
|
||
|
|
This script can test the services without requiring Docker
|
||
|
|
"""
|
||
|
|
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
import requests
|
||
|
|
import json
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
def run_command(cmd, check=True):
|
||
|
|
"""Run a shell command and return output"""
|
||
|
|
try:
|
||
|
|
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=check)
|
||
|
|
return result.stdout, result.stderr, result.returncode
|
||
|
|
except subprocess.CalledProcessError as e:
|
||
|
|
return e.stdout, e.stderr, e.returncode
|
||
|
|
|
||
|
|
def check_python_dependencies():
|
||
|
|
"""Check if required Python packages are installed"""
|
||
|
|
required_packages = ['flask', 'requests']
|
||
|
|
missing = []
|
||
|
|
|
||
|
|
for package in required_packages:
|
||
|
|
try:
|
||
|
|
__import__(package)
|
||
|
|
except ImportError:
|
||
|
|
missing.append(package)
|
||
|
|
|
||
|
|
if missing:
|
||
|
|
print(f"❌ Missing required packages: {', '.join(missing)}")
|
||
|
|
print("Install with: pip install flask requests")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print("✅ All required Python packages are installed")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def start_mock_services():
|
||
|
|
"""Start the mock services in background"""
|
||
|
|
print("🚀 Starting mock services...")
|
||
|
|
|
||
|
|
# Start SCADA service
|
||
|
|
scada_cmd = "cd tests/mock_services && python mock_scada_server.py > /tmp/scada.log 2>&1 &"
|
||
|
|
stdout, stderr, code = run_command(scada_cmd)
|
||
|
|
|
||
|
|
# Start Optimizer service
|
||
|
|
optimizer_cmd = "cd tests/mock_services && python mock_optimizer_server.py > /tmp/optimizer.log 2>&1 &"
|
||
|
|
stdout, stderr, code = run_command(optimizer_cmd)
|
||
|
|
|
||
|
|
print("✅ Mock services started in background")
|
||
|
|
print(" SCADA logs: /tmp/scada.log")
|
||
|
|
print(" Optimizer logs: /tmp/optimizer.log")
|
||
|
|
|
||
|
|
def wait_for_services():
|
||
|
|
"""Wait for services to be ready"""
|
||
|
|
print("⏳ Waiting for services to be ready...")
|
||
|
|
|
||
|
|
max_wait = 30
|
||
|
|
start_time = time.time()
|
||
|
|
|
||
|
|
while time.time() - start_time < max_wait:
|
||
|
|
try:
|
||
|
|
scada_ready = requests.get("http://localhost:8081/health", timeout=2).status_code == 200
|
||
|
|
optimizer_ready = requests.get("http://localhost:8082/health", timeout=2).status_code == 200
|
||
|
|
|
||
|
|
if scada_ready and optimizer_ready:
|
||
|
|
print("✅ All services are ready!")
|
||
|
|
return True
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
print(f" Waiting... ({int(time.time() - start_time)}/{max_wait} seconds)")
|
||
|
|
time.sleep(2)
|
||
|
|
|
||
|
|
print("❌ Services not ready within timeout period")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def test_scada_service():
|
||
|
|
"""Test SCADA service functionality"""
|
||
|
|
print("\n📊 Testing SCADA Service...")
|
||
|
|
|
||
|
|
tests_passed = 0
|
||
|
|
total_tests = 0
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Test health endpoint
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8081/health")
|
||
|
|
if response.status_code == 200 and response.json().get("status") == "healthy":
|
||
|
|
print(" ✅ Health check passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Health check failed")
|
||
|
|
|
||
|
|
# Test data endpoint
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8081/api/v1/data")
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "data" in data and "equipment" in data:
|
||
|
|
print(" ✅ Data retrieval passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Data structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Data endpoint failed")
|
||
|
|
|
||
|
|
# Test specific data tag
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8081/api/v1/data/temperature")
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "value" in data and "unit" in data:
|
||
|
|
print(" ✅ Specific data tag passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Specific data structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Specific data endpoint failed")
|
||
|
|
|
||
|
|
# Test equipment control
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.post(
|
||
|
|
"http://localhost:8081/api/v1/control/pump_1",
|
||
|
|
json={"command": "START"}
|
||
|
|
)
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "current_status" in data and data["current_status"] == "START":
|
||
|
|
print(" ✅ Equipment control passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Equipment control response invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Equipment control failed")
|
||
|
|
|
||
|
|
# Test alarms
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8081/api/v1/alarms")
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "alarms" in data:
|
||
|
|
print(" ✅ Alarms endpoint passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Alarms structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Alarms endpoint failed")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" ❌ SCADA test error: {e}")
|
||
|
|
|
||
|
|
print(f" 📈 SCADA tests: {tests_passed}/{total_tests} passed")
|
||
|
|
return tests_passed, total_tests
|
||
|
|
|
||
|
|
def test_optimizer_service():
|
||
|
|
"""Test optimizer service functionality"""
|
||
|
|
print("\n🧠 Testing Optimizer Service...")
|
||
|
|
|
||
|
|
tests_passed = 0
|
||
|
|
total_tests = 0
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Test health endpoint
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8082/health")
|
||
|
|
if response.status_code == 200 and response.json().get("status") == "healthy":
|
||
|
|
print(" ✅ Health check passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Health check failed")
|
||
|
|
|
||
|
|
# Test models endpoint
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8082/api/v1/models")
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "models" in data and "energy_optimization" in data["models"]:
|
||
|
|
print(" ✅ Models endpoint passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Models structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Models endpoint failed")
|
||
|
|
|
||
|
|
# Test energy optimization
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.post(
|
||
|
|
"http://localhost:8082/api/v1/optimize/energy_optimization",
|
||
|
|
json={"power_load": 450, "time_of_day": 14, "production_rate": 95}
|
||
|
|
)
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "result" in data and "optimization_id" in data:
|
||
|
|
print(" ✅ Energy optimization passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Optimization response invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Energy optimization failed")
|
||
|
|
|
||
|
|
# Test forecast
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.post(
|
||
|
|
"http://localhost:8082/api/v1/forecast",
|
||
|
|
json={"hours": 12}
|
||
|
|
)
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "forecast" in data and len(data["forecast"]) == 12:
|
||
|
|
print(" ✅ Forecast passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Forecast structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Forecast failed")
|
||
|
|
|
||
|
|
# Test history
|
||
|
|
total_tests += 1
|
||
|
|
response = requests.get("http://localhost:8082/api/v1/history")
|
||
|
|
if response.status_code == 200:
|
||
|
|
data = response.json()
|
||
|
|
if "history" in data and "total_optimizations" in data:
|
||
|
|
print(" ✅ History endpoint passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ History structure invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ History endpoint failed")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" ❌ Optimizer test error: {e}")
|
||
|
|
|
||
|
|
print(f" 📈 Optimizer tests: {tests_passed}/{total_tests} passed")
|
||
|
|
return tests_passed, total_tests
|
||
|
|
|
||
|
|
def test_end_to_end_workflow():
|
||
|
|
"""Test end-to-end workflow"""
|
||
|
|
print("\n🔄 Testing End-to-End Workflow...")
|
||
|
|
|
||
|
|
tests_passed = 0
|
||
|
|
total_tests = 0
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Get SCADA data
|
||
|
|
total_tests += 1
|
||
|
|
scada_response = requests.get("http://localhost:8081/api/v1/data")
|
||
|
|
if scada_response.status_code == 200:
|
||
|
|
scada_data = scada_response.json()
|
||
|
|
power_value = scada_data["data"]["power"]["value"]
|
||
|
|
print(" ✅ SCADA data retrieved")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ SCADA data retrieval failed")
|
||
|
|
return tests_passed, total_tests
|
||
|
|
|
||
|
|
# Run optimization based on SCADA data
|
||
|
|
total_tests += 1
|
||
|
|
opt_response = requests.post(
|
||
|
|
"http://localhost:8082/api/v1/optimize/energy_optimization",
|
||
|
|
json={
|
||
|
|
"power_load": power_value,
|
||
|
|
"time_of_day": datetime.now().hour,
|
||
|
|
"production_rate": 95
|
||
|
|
}
|
||
|
|
)
|
||
|
|
if opt_response.status_code == 200:
|
||
|
|
opt_data = opt_response.json()
|
||
|
|
if "result" in opt_data and "recommended_actions" in opt_data["result"]:
|
||
|
|
print(" ✅ Optimization based on SCADA data passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Optimization response invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Optimization failed")
|
||
|
|
|
||
|
|
# Test control based on optimization
|
||
|
|
total_tests += 1
|
||
|
|
control_response = requests.post(
|
||
|
|
"http://localhost:8081/api/v1/control/compressor",
|
||
|
|
json={"command": "START"}
|
||
|
|
)
|
||
|
|
if control_response.status_code == 200:
|
||
|
|
control_data = control_response.json()
|
||
|
|
if "current_status" in control_data:
|
||
|
|
print(" ✅ Control based on workflow passed")
|
||
|
|
tests_passed += 1
|
||
|
|
else:
|
||
|
|
print(" ❌ Control response invalid")
|
||
|
|
else:
|
||
|
|
print(" ❌ Control failed")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" ❌ End-to-end test error: {e}")
|
||
|
|
|
||
|
|
print(f" 📈 End-to-end tests: {tests_passed}/{total_tests} passed")
|
||
|
|
return tests_passed, total_tests
|
||
|
|
|
||
|
|
def stop_services():
|
||
|
|
"""Stop the mock services"""
|
||
|
|
print("\n🛑 Stopping mock services...")
|
||
|
|
|
||
|
|
# Find and kill the processes
|
||
|
|
run_command("pkill -f 'python mock_scada_server.py'", check=False)
|
||
|
|
run_command("pkill -f 'python mock_optimizer_server.py'", check=False)
|
||
|
|
|
||
|
|
print("✅ Mock services stopped")
|
||
|
|
|
||
|
|
def main():
|
||
|
|
"""Main test runner"""
|
||
|
|
print("🧪 Standalone Mock Services Test Runner")
|
||
|
|
print("=" * 50)
|
||
|
|
|
||
|
|
# Check dependencies
|
||
|
|
if not check_python_dependencies():
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
# Create mock services directory if needed
|
||
|
|
import os
|
||
|
|
os.makedirs("tests/mock_services", exist_ok=True)
|
||
|
|
|
||
|
|
# Check if mock service files exist
|
||
|
|
if not os.path.exists("tests/mock_services/mock_scada_server.py"):
|
||
|
|
print("❌ Mock service files not found. Run setup script first:")
|
||
|
|
print(" ./scripts/setup-test-environment.sh")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
# Start services
|
||
|
|
start_mock_services()
|
||
|
|
|
||
|
|
# Wait for services
|
||
|
|
if not wait_for_services():
|
||
|
|
stop_services()
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
# Run tests
|
||
|
|
total_passed = 0
|
||
|
|
total_tests = 0
|
||
|
|
|
||
|
|
# Test SCADA
|
||
|
|
passed, tests = test_scada_service()
|
||
|
|
total_passed += passed
|
||
|
|
total_tests += tests
|
||
|
|
|
||
|
|
# Test Optimizer
|
||
|
|
passed, tests = test_optimizer_service()
|
||
|
|
total_passed += passed
|
||
|
|
total_tests += tests
|
||
|
|
|
||
|
|
# Test End-to-End
|
||
|
|
passed, tests = test_end_to_end_workflow()
|
||
|
|
total_passed += passed
|
||
|
|
total_tests += tests
|
||
|
|
|
||
|
|
# Stop services
|
||
|
|
stop_services()
|
||
|
|
|
||
|
|
# Print summary
|
||
|
|
print("\n" + "=" * 50)
|
||
|
|
print("📊 TEST SUMMARY")
|
||
|
|
print("=" * 50)
|
||
|
|
print(f"Total Tests: {total_tests}")
|
||
|
|
print(f"Tests Passed: {total_passed}")
|
||
|
|
print(f"Tests Failed: {total_tests - total_passed}")
|
||
|
|
print(f"Success Rate: {(total_passed/total_tests)*100:.1f}%")
|
||
|
|
|
||
|
|
if total_passed == total_tests:
|
||
|
|
print("\n🎉 ALL TESTS PASSED!")
|
||
|
|
print("Mock services are working correctly!")
|
||
|
|
else:
|
||
|
|
print(f"\n❌ {total_tests - total_passed} TESTS FAILED")
|
||
|
|
print("Check the logs above for details")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|