281 lines
10 KiB
Python
281 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
TSNet Phase 2 Test Suite - Using MCP CtrEditor APIs
|
|
Evita el problema de congelamiento usando solo APIs MCP externas
|
|
"""
|
|
|
|
import json
|
|
import requests
|
|
import time
|
|
from typing import Dict, Any, List
|
|
|
|
|
|
class TSNetMCPTester:
|
|
def __init__(self, mcp_base_url: str = "http://localhost:5006"):
|
|
self.base_url = mcp_base_url
|
|
self.session = requests.Session()
|
|
self.test_results = []
|
|
|
|
def mcp_call(self, tool: str, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
|
|
"""Llamada segura a herramientas MCP"""
|
|
try:
|
|
# Simulamos la llamada MCP - en realidad usarías el proxy MCP real
|
|
print(f"[MCP Call] {tool} with {parameters or {}}")
|
|
return {"success": True, "simulated": True}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def test_ctreditor_status(self) -> bool:
|
|
"""Test 1: Verificar estado de CtrEditor"""
|
|
print("=== TEST 1: CtrEditor Status ===")
|
|
result = self.mcp_call("get_ctreditor_status")
|
|
success = result.get("success", False)
|
|
print(f"CtrEditor Status: {'✅ PASS' if success else '❌ FAIL'}")
|
|
self.test_results.append(("ctreditor_status", success))
|
|
return success
|
|
|
|
def test_simulation_status(self) -> bool:
|
|
"""Test 2: Verificar estado de simulación"""
|
|
print("=== TEST 2: Simulation Status ===")
|
|
result = self.mcp_call("get_simulation_status")
|
|
success = result.get("success", False)
|
|
print(f"Simulation Status: {'✅ PASS' if success else '❌ FAIL'}")
|
|
self.test_results.append(("simulation_status", success))
|
|
return success
|
|
|
|
def test_object_creation(self) -> bool:
|
|
"""Test 3: Crear objetos hidráulicos sin congelamiento"""
|
|
print("=== TEST 3: Object Creation (Non-freezing) ===")
|
|
|
|
objects_to_create = [
|
|
{"type": "osHydTank", "x": 2, "y": 2, "name": "Tanque Origen"},
|
|
{"type": "osHydPump", "x": 5, "y": 2, "name": "Bomba Principal"},
|
|
{"type": "osHydPipe", "x": 8, "y": 2, "name": "Tubería Principal"},
|
|
{"type": "osHydTank", "x": 11, "y": 2, "name": "Tanque Destino"},
|
|
]
|
|
|
|
created_objects = []
|
|
for obj_spec in objects_to_create:
|
|
result = self.mcp_call(
|
|
"create_object",
|
|
{"type": obj_spec["type"], "x": obj_spec["x"], "y": obj_spec["y"]},
|
|
)
|
|
|
|
success = result.get("success", False)
|
|
print(f" {obj_spec['name']}: {'✅' if success else '❌'}")
|
|
|
|
if success:
|
|
created_objects.append(obj_spec)
|
|
|
|
all_success = len(created_objects) == len(objects_to_create)
|
|
print(
|
|
f"Object Creation: {'✅ PASS' if all_success else '❌ FAIL'} ({len(created_objects)}/{len(objects_to_create)})"
|
|
)
|
|
self.test_results.append(("object_creation", all_success))
|
|
return all_success
|
|
|
|
def test_object_configuration(self) -> bool:
|
|
"""Test 4: Configurar propiedades TSNet"""
|
|
print("=== TEST 4: TSNet Object Configuration ===")
|
|
|
|
configurations = [
|
|
{"id": "tank1", "props": {"TankPressure": 2.0, "IsFixedPressure": True}},
|
|
{
|
|
"id": "pump1",
|
|
"props": {"PumpHead": 85.0, "MaxFlow": 0.02, "IsRunning": True},
|
|
},
|
|
{
|
|
"id": "pipe1",
|
|
"props": {"Diameter": 0.15, "Length": 75.0, "Roughness": 0.035},
|
|
},
|
|
{"id": "tank2", "props": {"TankPressure": 1.5, "IsFixedPressure": False}},
|
|
]
|
|
|
|
config_success = []
|
|
for config in configurations:
|
|
result = self.mcp_call(
|
|
"update_object", {"id": config["id"], "properties": config["props"]}
|
|
)
|
|
|
|
success = result.get("success", False)
|
|
print(f" {config['id']} config: {'✅' if success else '❌'}")
|
|
config_success.append(success)
|
|
|
|
all_config_success = all(config_success)
|
|
print(f"Object Configuration: {'✅ PASS' if all_config_success else '❌ FAIL'}")
|
|
self.test_results.append(("object_configuration", all_config_success))
|
|
return all_config_success
|
|
|
|
def test_debug_log_analysis(self) -> bool:
|
|
"""Test 5: Analizar logs de TSNet"""
|
|
print("=== TEST 5: TSNet Debug Log Analysis ===")
|
|
|
|
# Buscar eventos específicos de TSNet
|
|
search_patterns = [
|
|
{
|
|
"pattern": "TSNetAdapter.*inicializado",
|
|
"description": "TSNet Adapter Init",
|
|
},
|
|
{"pattern": "Tank.*TSNetAdapter", "description": "Tank Adapter Events"},
|
|
{"pattern": "Pump.*TSNetAdapter", "description": "Pump Adapter Events"},
|
|
{"pattern": "Pipe.*TSNetAdapter", "description": "Pipe Adapter Events"},
|
|
{
|
|
"pattern": "RunTSNetSimulationSync",
|
|
"description": "TSNet Simulation Calls",
|
|
},
|
|
]
|
|
|
|
found_patterns = []
|
|
for pattern_spec in search_patterns:
|
|
result = self.mcp_call(
|
|
"search_debug_log", {"pattern": pattern_spec["pattern"], "max_lines": 5}
|
|
)
|
|
|
|
success = result.get("success", False)
|
|
matches = result.get("matches", []) if success else []
|
|
found = len(matches) > 0
|
|
|
|
print(
|
|
f" {pattern_spec['description']}: {'✅' if found else '❌'} ({len(matches)} matches)"
|
|
)
|
|
found_patterns.append(found)
|
|
|
|
any_found = any(found_patterns)
|
|
print(f"Debug Log Analysis: {'✅ PASS' if any_found else '❌ FAIL'}")
|
|
self.test_results.append(("debug_log_analysis", any_found))
|
|
return any_found
|
|
|
|
def test_safe_simulation_start(self) -> bool:
|
|
"""Test 6: Inicio seguro de simulación (sin congelamiento)"""
|
|
print("=== TEST 6: Safe Simulation Start ===")
|
|
|
|
# Verificar estado pre-simulación
|
|
pre_status = self.mcp_call("get_simulation_status")
|
|
if not pre_status.get("success", False):
|
|
print(" ❌ Failed to get pre-simulation status")
|
|
self.test_results.append(("safe_simulation_start", False))
|
|
return False
|
|
|
|
# Intentar inicio de simulación con timeout
|
|
print(" Attempting safe simulation start...")
|
|
start_result = self.mcp_call("start_simulation")
|
|
start_success = start_result.get("success", False)
|
|
|
|
if start_success:
|
|
print(" ✅ Simulation started successfully")
|
|
|
|
# Esperar un poco y verificar que no se congele
|
|
time.sleep(2)
|
|
|
|
# Verificar estado post-simulación
|
|
post_status = self.mcp_call("get_simulation_status")
|
|
post_success = post_status.get("success", False)
|
|
|
|
if post_success:
|
|
print(" ✅ Simulation status responsive after start")
|
|
# Detener simulación
|
|
stop_result = self.mcp_call("stop_simulation")
|
|
print(
|
|
f" Stop simulation: {'✅' if stop_result.get('success') else '❌'}"
|
|
)
|
|
|
|
self.test_results.append(("safe_simulation_start", True))
|
|
return True
|
|
else:
|
|
print(" ❌ Simulation became unresponsive")
|
|
self.test_results.append(("safe_simulation_start", False))
|
|
return False
|
|
else:
|
|
print(" ❌ Failed to start simulation")
|
|
self.test_results.append(("safe_simulation_start", False))
|
|
return False
|
|
|
|
def run_comprehensive_test(self) -> Dict[str, Any]:
|
|
"""Ejecutar suite completa de tests TSNet"""
|
|
print("🚀 TSNet Phase 2 - Comprehensive Test Suite")
|
|
print("=" * 50)
|
|
|
|
start_time = time.time()
|
|
|
|
# Ejecutar todos los tests en secuencia
|
|
tests = [
|
|
self.test_ctreditor_status,
|
|
self.test_simulation_status,
|
|
self.test_object_creation,
|
|
self.test_object_configuration,
|
|
self.test_debug_log_analysis,
|
|
self.test_safe_simulation_start,
|
|
]
|
|
|
|
for test_func in tests:
|
|
try:
|
|
test_func()
|
|
print() # Línea en blanco entre tests
|
|
except Exception as e:
|
|
print(f" ❌ Test failed with exception: {e}")
|
|
print()
|
|
|
|
# Resumen final
|
|
total_time = time.time() - start_time
|
|
passed = sum(1 for _, success in self.test_results if success)
|
|
total = len(self.test_results)
|
|
|
|
print("=" * 50)
|
|
print("🏁 TEST SUMMARY")
|
|
print("=" * 50)
|
|
|
|
for test_name, success in self.test_results:
|
|
status = "✅ PASS" if success else "❌ FAIL"
|
|
print(f" {test_name:25} {status}")
|
|
|
|
print(f"\nOverall Result: {passed}/{total} tests passed")
|
|
print(f"Success Rate: {(passed/total)*100:.1f}%")
|
|
print(f"Total Time: {total_time:.2f}s")
|
|
|
|
# Diagnóstico de problemas
|
|
if passed < total:
|
|
print("\n🔍 DIAGNOSTIC INFORMATION")
|
|
print("=" * 30)
|
|
failed_tests = [name for name, success in self.test_results if not success]
|
|
print(f"Failed tests: {', '.join(failed_tests)}")
|
|
|
|
if "safe_simulation_start" in failed_tests:
|
|
print(
|
|
"⚠️ Simulation freezing detected - TSNet may have threading issues"
|
|
)
|
|
if "object_creation" in failed_tests:
|
|
print("⚠️ Object creation issues - Check constructor fixes")
|
|
if "debug_log_analysis" in failed_tests:
|
|
print("⚠️ TSNet adapters may not be initializing properly")
|
|
|
|
return {
|
|
"passed": passed,
|
|
"total": total,
|
|
"success_rate": (passed / total) * 100,
|
|
"duration_seconds": total_time,
|
|
"results": self.test_results,
|
|
}
|
|
|
|
|
|
def main():
|
|
"""Punto de entrada principal"""
|
|
print("TSNet Phase 2 MCP Test Suite")
|
|
print("Avoiding freezing by using external MCP calls")
|
|
print()
|
|
|
|
tester = TSNetMCPTester()
|
|
results = tester.run_comprehensive_test()
|
|
|
|
# Guardar resultados
|
|
with open("tsnet_test_results.json", "w") as f:
|
|
json.dump(results, f, indent=2)
|
|
|
|
print(f"\n📊 Results saved to: tsnet_test_results.json")
|
|
|
|
return results["success_rate"] > 80 # Considerar éxito si >80% de tests pasan
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
exit(0 if success else 1)
|