Relint

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

async_testing/async_validator.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 """
 Validate async code patterns and detect common pitfalls.
 """
@@ -10,116 +11,116 @@
 
 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)
@@ -130,18 +131,18 @@ def _check_blocking_operations(self, node):
                         '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
@@ -153,10 +154,10 @@ def _check_unawaited_calls(self, node):
                         '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):
@@ -165,19 +166,19 @@ def _get_call_name(self, node):
             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}")
+
+    print(f"Validation report saved to {args.report}")

async_testing/benchmarks.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 """
 Run async performance benchmarks and output results.
 """
@@ -10,63 +11,63 @@
 
 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)
+    benchmark.save_results(args.output)