#!/usr/bin/env python3 """ TSNet Phase 2 - Test Suite Completo Tests exhaustivos para validar la funcionalidad de simulación TSNet """ import requests import json import time import sys from datetime import datetime class TSNetTestSuite: def __init__(self, base_url="http://localhost:5006"): self.base_url = base_url self.test_results = [] self.created_objects = [] def log_test(self, test_name, passed, details=""): """Registrar resultado de test""" self.test_results.append({ 'test': test_name, 'passed': passed, 'details': details, 'timestamp': datetime.now().isoformat() }) status = "✅ PASS" if passed else "❌ FAIL" print(f"{status}: {test_name}") if details: print(f" Details: {details}") def send_mcp_request(self, method, params=None, timeout=10): """Enviar request MCP con manejo de errores""" try: payload = { 'jsonrpc': '2.0', 'method': method, 'id': int(time.time() * 1000) } if params: payload['params'] = params response = requests.post(self.base_url, json=payload, timeout=timeout) if response.status_code == 200: result = response.json() if 'error' in result: return False, f"MCP Error: {result['error']}" return True, result.get('result', {}) else: return False, f"HTTP {response.status_code}: {response.text[:200]}" except Exception as e: return False, f"Exception: {str(e)}" def test_01_mcp_connectivity(self): """Test 1: Conectividad básica MCP""" success, result = self.send_mcp_request('get_ctreditor_status') if success: status = result.get('connection_status', 'unknown') self.log_test("MCP Connectivity", status == 'available', f"Status: {status}") else: self.log_test("MCP Connectivity", False, result) return success def test_02_clear_workspace(self): """Test 2: Limpiar workspace""" success, result = self.send_mcp_request('list_objects') if success: objects = result.get('objects', []) if objects: # Eliminar objetos existentes object_ids = [str(obj['id']['Value']) for obj in objects] del_success, del_result = self.send_mcp_request('delete_objects', {'ids': object_ids}) self.log_test("Clear Workspace", del_success, f"Deleted {len(object_ids)} objects") else: self.log_test("Clear Workspace", True, "Workspace already empty") else: self.log_test("Clear Workspace", False, result) return success def test_03_create_hydraulic_system(self): """Test 3: Crear sistema hidráulico completo""" components = [ {'type': 'osHydTank', 'x': 1.0, 'y': 1.0, 'name': 'Source Tank'}, {'type': 'osHydPump', 'x': 3.0, 'y': 1.0, 'name': 'Main Pump'}, {'type': 'osHydPipe', 'x': 5.0, 'y': 1.0, 'name': 'Pipe 1'}, {'type': 'osHydTank', 'x': 7.0, 'y': 1.0, 'name': 'Target Tank'}, ] created_count = 0 for comp in components: success, result = self.send_mcp_request('create_object', { 'type': comp['type'], 'x': comp['x'], 'y': comp['y'] }) if success: created_count += 1 self.created_objects.append(comp['name']) total_components = len(components) all_created = created_count == total_components self.log_test("Create Hydraulic System", all_created, f"Created {created_count}/{total_components} components") return all_created def test_04_list_created_objects(self): """Test 4: Verificar objetos creados""" success, result = self.send_mcp_request('list_objects') if success: objects = result.get('objects', []) hydraulic_objects = [obj for obj in objects if 'osHyd' in obj.get('type', '')] self.log_test("List Created Objects", len(hydraulic_objects) > 0, f"Found {len(hydraulic_objects)} hydraulic objects") # Mostrar detalles de objetos hidráulicos for obj in hydraulic_objects: obj_type = obj.get('type', 'Unknown') obj_id = obj.get('id', {}).get('Value', 'No ID') print(f" - {obj_type} (ID: {obj_id})") return len(hydraulic_objects) > 0 else: self.log_test("List Created Objects", False, result) return False def test_05_tsnet_basic_integration(self): """Test 5: Integración básica TSNet""" code = """ try: print("Testing TSNet basic integration...") app.TestTSNetIntegrationSync() result = "SUCCESS: Basic TSNet integration test completed" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=30) test_passed = success and "SUCCESS" in str(result) self.log_test("TSNet Basic Integration", test_passed, str(result)[:200]) return test_passed def test_06_tsnet_full_simulation(self): """Test 6: Simulación completa TSNet""" code = """ try: print("Testing TSNet full simulation...") # Verificar objetos hidráulicos disponibles hydraulic_objects = [] for obj in app.ObjetosSimulables: if obj.GetType().Name.startswith("osHyd"): hydraulic_objects.append({ 'name': obj.Nombre, 'type': obj.GetType().Name, 'id': obj.Id.Value if obj.Id else None }) print(f"Found {len(hydraulic_objects)} hydraulic objects:") for hobj in hydraulic_objects: print(f" - {hobj['name']} ({hobj['type']}) ID: {hobj['id']}") # Ejecutar simulación TSNet app.RunTSNetSimulationSync() result = f"SUCCESS: Full TSNet simulation completed with {len(hydraulic_objects)} objects" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=45) test_passed = success and "SUCCESS" in str(result) self.log_test("TSNet Full Simulation", test_passed, str(result)[:300]) return test_passed def test_07_adapter_validation(self): """Test 7: Validación de adaptadores TSNet""" code = """ try: print("Testing TSNet adapters validation...") # Verificar adaptadores creados adapter_count = { 'tanks': len(app.tsnetSimulationManager._tankAdapters) if hasattr(app.tsnetSimulationManager, '_tankAdapters') else 0, 'pumps': len(app.tsnetSimulationManager._pumpAdapters) if hasattr(app.tsnetSimulationManager, '_pumpAdapters') else 0, 'pipes': len(app.tsnetSimulationManager._pipeAdapters) if hasattr(app.tsnetSimulationManager, '_pipeAdapters') else 0 } print(f"Adapters created: {adapter_count}") # Validar configuraciones config_errors = app.tsnetSimulationManager.ValidateAllConfigurations() print(f"Configuration errors: {len(config_errors)}") for error in config_errors: print(f" - {error}") total_adapters = sum(adapter_count.values()) result = f"SUCCESS: {total_adapters} adapters, {len(config_errors)} config errors" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=30) test_passed = success and "SUCCESS" in str(result) self.log_test("Adapter Validation", test_passed, str(result)[:300]) return test_passed def test_08_network_rebuild(self): """Test 8: Reconstrucción de red hidráulica""" code = """ try: print("Testing network rebuild...") # Reconstruir red app.tsnetSimulationManager.RebuildNetwork() # Verificar estado de la red network_state = "Network rebuilt successfully" if hasattr(app.tsnetSimulationManager, 'Network'): network_state += " - Network object exists" result = f"SUCCESS: {network_state}" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=30) test_passed = success and "SUCCESS" in str(result) self.log_test("Network Rebuild", test_passed, str(result)[:200]) return test_passed def test_09_object_registration_stress(self): """Test 9: Stress test de registro de objetos""" code = """ try: print("Testing object registration stress...") # Resetear manager app.tsnetSimulationManager.ResetAllCalculatedValues() # Registrar todos los objetos hidráulicos múltiples veces registration_count = 0 for iteration in range(3): for obj in app.ObjetosSimulables: if obj.GetType().Name.startswith("osHyd"): try: app.tsnetSimulationManager.RegisterHydraulicObject(obj) registration_count += 1 except Exception as reg_ex: print(f"Registration error for {obj.Nombre}: {reg_ex}") result = f"SUCCESS: {registration_count} registrations completed" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=30) test_passed = success and "SUCCESS" in str(result) self.log_test("Object Registration Stress", test_passed, str(result)[:200]) return test_passed def test_10_performance_timing(self): """Test 10: Medición de rendimiento""" code = """ import time try: print("Testing performance timing...") # Medir tiempo de simulación completa start_time = time.time() # Resetear app.tsnetSimulationManager.ResetAllCalculatedValues() # Registrar objetos reg_start = time.time() hydraulic_count = 0 for obj in app.ObjetosSimulables: if obj.GetType().Name.startswith("osHyd"): app.tsnetSimulationManager.RegisterHydraulicObject(obj) hydraulic_count += 1 reg_time = time.time() - reg_start # Validar configuraciones val_start = time.time() config_errors = app.tsnetSimulationManager.ValidateAllConfigurations() val_time = time.time() - val_start # Reconstruir red net_start = time.time() app.tsnetSimulationManager.RebuildNetwork() net_time = time.time() - net_start total_time = time.time() - start_time result = f"SUCCESS: Total: {total_time:.3f}s, Registration: {reg_time:.3f}s, Validation: {val_time:.3f}s, Network: {net_time:.3f}s ({hydraulic_count} objects)" except Exception as e: result = f"ERROR: {str(e)}" import traceback result += "\\n" + traceback.format_exc() print(result) """ success, result = self.send_mcp_request('execute_python', {'code': code}, timeout=30) test_passed = success and "SUCCESS" in str(result) self.log_test("Performance Timing", test_passed, str(result)[:300]) return test_passed def run_all_tests(self): """Ejecutar todos los tests""" print("🧪 TSNet Phase 2 - Complete Test Suite") print("=" * 60) print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print() # Lista de tests a ejecutar tests = [ self.test_01_mcp_connectivity, self.test_02_clear_workspace, self.test_03_create_hydraulic_system, self.test_04_list_created_objects, self.test_05_tsnet_basic_integration, self.test_06_tsnet_full_simulation, self.test_07_adapter_validation, self.test_08_network_rebuild, self.test_09_object_registration_stress, self.test_10_performance_timing ] # Ejecutar tests for i, test_func in enumerate(tests, 1): print(f"\n--- Test {i:02d}: {test_func.__doc__.split(':')[1].strip()} ---") try: test_func() except Exception as e: self.log_test(f"Test {i:02d} Execution", False, f"Exception: {str(e)}") # Pequeña pausa entre tests time.sleep(1) # Resumen final self.print_summary() def print_summary(self): """Imprimir resumen de resultados""" print("\n" + "=" * 60) print("📊 TEST SUMMARY") print("=" * 60) passed_tests = [r for r in self.test_results if r['passed']] failed_tests = [r for r in self.test_results if not r['passed']] print(f"Total Tests: {len(self.test_results)}") print(f"Passed: {len(passed_tests)} ✅") print(f"Failed: {len(failed_tests)} ❌") print(f"Success Rate: {len(passed_tests)/len(self.test_results)*100:.1f}%") if failed_tests: print("\n❌ Failed Tests:") for test in failed_tests: print(f" • {test['test']}: {test['details'][:100]}") if len(passed_tests) == len(self.test_results): print("\n🎉 ALL TESTS PASSED! TSNet Phase 2 is fully functional.") elif len(passed_tests) >= len(self.test_results) * 0.8: print("\n✅ Most tests passed. TSNet Phase 2 is mostly functional.") else: print("\n⚠️ Multiple test failures. Please review the implementation.") print(f"\nCompleted at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") def main(): test_suite = TSNetTestSuite() test_suite.run_all_tests() if __name__ == "__main__": main()