Add new async testing files

Signed-off-by: Sebastian <[email protected]>

Author: Sebastian

async_testing/async_validator.py

@@ -0,0 +1,183 @@
+"""
+Validate async code patterns and detect common pitfalls.
+"""
+
+import ast
+import argparse
+import json
+from pathlib import Path
+from typing import List, Dict, Any
+
+class AsyncCodeValidator:
+    """Validate async code for common patterns and pitfalls."""
+    
+    def __init__(self):
+        self.issues = []
+        self.suggestions = []
+        
+    def validate_directory(self, source_dir: Path) -> Dict[str, Any]:
+        """Validate all Python files in directory."""
+        
+        validation_results = {
+            'files_checked': 0,
+            'issues_found': 0,
+            'suggestions': 0,
+            'details': []
+        }
+        
+        python_files = list(source_dir.rglob("*.py"))
+        
+        for file_path in python_files:
+            if self._should_skip_file(file_path):
+                continue
+                
+            file_results = self._validate_file(file_path)
+            validation_results['details'].append(file_results)
+            validation_results['files_checked'] += 1
+            validation_results['issues_found'] += len(file_results['issues'])
+            validation_results['suggestions'] += len(file_results['suggestions'])
+            
+        return validation_results
+        
+    def _validate_file(self, file_path: Path) -> Dict[str, Any]:
+        """Validate a single Python file."""
+        
+        file_results = {
+            'file': str(file_path),
+            'issues': [],
+            'suggestions': []
+        }
+        
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                source_code = f.read()
+                
+            tree = ast.parse(source_code, filename=str(file_path))
+            
+            # Analyze AST for async patterns
+            validator = AsyncPatternVisitor(file_path)
+            validator.visit(tree)
+            
+            file_results['issues'] = validator.issues
+            file_results['suggestions'] = validator.suggestions
+            
+        except Exception as e:
+            file_results['issues'].append({
+                'type': 'parse_error',
+                'message': f"Failed to parse file: {str(e)}",
+                'line': 0
+            })
+            
+        return file_results
+    
+    
+    def _should_skip_file(self, file_path: Path) -> bool:
+        """Determine if a file should be skipped (e.g., __init__.py files)."""
+        return file_path.name == "__init__.py"
+        
+class AsyncPatternVisitor(ast.NodeVisitor):
+    """AST visitor to detect async patterns and issues."""
+    
+    def __init__(self, file_path: Path):
+        self.file_path = file_path
+        self.issues = []
+        self.suggestions = []
+        self.in_async_function = False
+        
+    def visit_AsyncFunctionDef(self, node):
+        """Visit async function definitions."""
+        
+        self.in_async_function = True
+        
+        # Check for blocking operations in async functions
+        self._check_blocking_operations(node)
+        
+        # Check for proper error handling
+        self._check_error_handling(node)
+        
+        self.generic_visit(node)
+        self.in_async_function = False
+        
+    def visit_Call(self, node):
+        """Visit function calls."""
+        
+        if self.in_async_function:
+            # Check for potentially unawaited async calls
+            self._check_unawaited_calls(node)
+            
+            # Check for blocking I/O operations
+            self._check_blocking_io(node)
+            
+        self.generic_visit(node)
+        
+    def _check_blocking_operations(self, node):
+        """Check for blocking operations in async functions."""
+        
+        blocking_patterns = [
+            'time.sleep',
+            'requests.get', 'requests.post',
+            'subprocess.run', 'subprocess.call',
+            'open'  # File I/O without async
+        ]
+        
+        for child in ast.walk(node):
+            if isinstance(child, ast.Call):
+                call_name = self._get_call_name(child)
+                if call_name in blocking_patterns:
+                    self.issues.append({
+                        'type': 'blocking_operation',
+                        'message': f"Blocking operation '{call_name}' in async function",
+                        'line': child.lineno,
+                        'suggestion': f"Use async equivalent of {call_name}"
+                    })
+                    
+    def _check_unawaited_calls(self, node):
+        """Check for potentially unawaited async calls."""
+        
+        # Look for calls that might return coroutines
+        async_patterns = [
+            'aiohttp', 'asyncio', 'asyncpg',
+            'websockets', 'motor'  # Common async libraries
+        ]
+        
+        call_name = self._get_call_name(node)
+        
+        for pattern in async_patterns:
+            if pattern in call_name:
+                # Check if this call is awaited
+                parent = getattr(node, 'parent', None)
+                if not isinstance(parent, ast.Await):
+                    self.suggestions.append({
+                        'type': 'potentially_unawaited',
+                        'message': f"Call to '{call_name}' might need await",
+                        'line': node.lineno
+                    })
+                    break
+                    
+    def _get_call_name(self, node):
+        """Extract the name of a function call."""
+        
+        if isinstance(node.func, ast.Name):
+            return node.func.id
+        elif isinstance(node.func, ast.Attribute):
+            if isinstance(node.func.value, ast.Name):
+                return f"{node.func.value.id}.{node.func.attr}"
+            else:
+                return node.func.attr
+        return "unknown"
+    
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Validate async code patterns and detect common pitfalls.")
+    parser.add_argument("--source", type=Path, required=True, help="Source directory to validate.")
+    parser.add_argument("--report", type=Path, required=True, help="Path to the output validation report.")
+    
+    args = parser.parse_args()
+    
+    validator = AsyncCodeValidator()
+    results = validator.validate_directory(args.source)
+    
+    with open(args.report, 'w') as f:
+        json.dump(results, f, indent=4)
+    
+    print(f"Validation report saved to {args.report}")

async_testing/benchmarks.py

@@ -0,0 +1,72 @@
+"""
+Run async performance benchmarks and output results.
+"""
+import asyncio
+import time
+import json
+import argparse
+from pathlib import Path
+from typing import Any, Dict
+
+class AsyncBenchmark:
+    """Run async performance benchmarks."""
+    
+    def __init__(self, iterations: int):
+        self.iterations = iterations
+        self.results: Dict[str, Any] = {
+            'iterations': self.iterations,
+            'benchmarks': []
+        }
+    
+    async def run_benchmarks(self) -> None:
+        """Run all benchmarks."""
+        
+        # Example benchmarks
+        await self._benchmark_example("Example Benchmark 1", self.example_benchmark_1)
+        await self._benchmark_example("Example Benchmark 2", self.example_benchmark_2)
+    
+    async def _benchmark_example(self, name: str, benchmark_func) -> None:
+        """Run a single benchmark and record its performance."""
+        
+        start_time = time.perf_counter()
+        
+        for _ in range(self.iterations):
+            await benchmark_func()
+        
+        end_time = time.perf_counter()
+        total_time = end_time - start_time
+        avg_time = total_time / self.iterations
+        
+        self.results['benchmarks'].append({
+            'name': name,
+            'total_time': total_time,
+            'average_time': avg_time
+        })
+        
+    async def example_benchmark_1(self) -> None:
+        """An example async benchmark function."""
+        await asyncio.sleep(0.001)
+    
+    async def example_benchmark_2(self) -> None:
+        """Another example async benchmark function."""
+        await asyncio.sleep(0.002)
+    
+    def save_results(self, output_path: Path) -> None:
+        """Save benchmark results to a file."""
+        
+        with open(output_path, 'w') as f:
+            json.dump(self.results, f, indent=4)
+        
+        print(f"Benchmark results saved to {output_path}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Run async performance benchmarks.")
+    parser.add_argument("--output", type=Path, required=True, help="Path to the output benchmark results file.")
+    parser.add_argument("--iterations", type=int, default=1000, help="Number of iterations to run each benchmark.")
+    
+    args = parser.parse_args()
+    
+    benchmark = AsyncBenchmark(args.iterations)
+    asyncio.run(benchmark.run_benchmarks())
+    benchmark.save_results(args.output)

async_testing/config.yaml

@@ -0,0 +1,25 @@
+async_linting:
+  ruff_rules: ["F", "E", "B", "ASYNC"]
+  flake8_plugins: ["flake8-bugbear", "flake8-async"]
+  mypy_config:
+    warn_unused_coroutine: true
+    strict: true
+  
+profiling:
+  output_dir: "async_testing/profiles"
+  snakeviz_port: 8080
+  profile_scenarios:
+    - "websocket_stress_test" 
+    - "database_query_performance"
+    - "concurrent_mcp_calls"
+    
+monitoring:
+  aiomonitor_port: 50101
+  debug_mode: true
+  task_tracking: true
+  
+performance_thresholds:
+  websocket_connection: 100  # ms
+  database_query: 50         # ms
+  mcp_rpc_call: 100          # ms
+  concurrent_users: 100      # simultaneous connections

async_testing/monitor_runner.py

@@ -0,0 +1,95 @@
+"""
+Runtime async monitoring with aiomonitor integration.
+"""
+import asyncio
+from typing import Any, Dict
+import aiomonitor
+import argparse
+
+class AsyncMonitor:
+    """Monitor live async operations in mcpgateway."""
+    
+    def __init__(self, webui_port: int = 50101, console_port: int = 50102, host: str = "localhost"):
+        self.webui_port = webui_port
+        self.console_port = console_port
+        self.host = host
+        self.monitor = None
+        self.running = False
+        
+    async def start_monitoring(self, console_enabled: bool = True):
+        """Start aiomonitor for live async debugging."""
+        
+        print(f"👁️  Starting aiomonitor on http://{self.host}:{self.webui_port}")
+        
+        # Configure aiomonitor
+        self.monitor = aiomonitor.Monitor(
+            asyncio.get_event_loop(),
+            host=self.host,
+            webui_port=self.webui_port,
+            console_port=self.console_port,     # TODO: FIX CONSOLE NOT CONNECTING TO PORT
+            console_enabled=console_enabled,
+            locals={'monitor': self}
+        )
+        
+        self.monitor.start()
+        self.running = True
+        
+        if console_enabled:
+            print(f"🌐 aiomonitor console available at: http://{self.host}:{self.console_port}")
+            print("📊 Available commands: ps, where, cancel, signal, console")
+            print("🔍 Use 'ps' to list running tasks")
+            print("📍 Use 'where <task_id>' to see task stack trace")
+        
+        # Keep monitoring running
+        try:
+            while self.running:
+                await asyncio.sleep(1)
+                
+                # Periodic task summary
+                tasks = [t for t in asyncio.all_tasks() if not t.done()]
+                if len(tasks) % 100 == 0 and len(tasks) > 0:
+                    print(f"📈 Current active tasks: {len(tasks)}")
+                    
+        except KeyboardInterrupt: # TODO: FIX STACK TRACE STILL APPEARING ON CTRL-C
+            print("\n🛑 Stopping aiomonitor...")
+        finally:
+            self.monitor.close()
+            
+    def stop_monitoring(self):
+        """Stop the monitoring."""
+        self.running = False
+        
+    async def get_task_summary(self) -> Dict[str, Any]:
+        """Get summary of current async tasks."""
+        
+        tasks = asyncio.all_tasks()
+        
+        summary: Dict[str, Any] = {
+            'total_tasks': len(tasks),
+            'running_tasks': len([t for t in tasks if not t.done()]),
+            'completed_tasks': len([t for t in tasks if t.done()]),
+            'cancelled_tasks': len([t for t in tasks if t.cancelled()]),
+            'task_details': []
+        }
+        
+        for task in tasks:
+            if not task.done():
+                summary['task_details'].append({
+                    'name': getattr(task, '_name', 'unnamed'),
+                    'state': task._state.name if hasattr(task, '_state') else 'unknown',
+                    'coro': str(task._coro) if hasattr(task, '_coro') else 'unknown'
+                })
+                
+        return summary
+    
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Run aiomonitor for live async debugging.")
+    parser.add_argument("--host", type=str, default="localhost", help="Host to run aiomonitor on.")
+    parser.add_argument("--webui_port", type=int, default=50101, help="Port to run aiomonitor on.")
+    parser.add_argument("--console_port", type=int, default=50102, help="Port to run aiomonitor on.")
+    parser.add_argument("--console-enabled", action="store_true", help="Enable console for aiomonitor.")
+    
+    args = parser.parse_args()
+    
+    monitor = AsyncMonitor(webui_port=args.webui_port, console_port=args.console_port, host=args.host)
+    asyncio.run(monitor.start_monitoring(console_enabled=args.console_enabled))