MaverickMCP

""" Test runner validation for parallel research functionality test suites. This module validates that all test suites follow pytest best practices and async patterns without triggering circular imports during validation. """ import ast import re from pathlib import Path from typing import Any class TestSuiteValidator: """Validator for test suite structure and patterns.""" def __init__(self, test_file_path: str): self.test_file_path = Path(test_file_path) self.content = self.test_file_path.read_text() self.tree = ast.parse(self.content) def validate_pytest_patterns(self) -> dict[str, Any]: """Validate pytest patterns and best practices.""" results = { "has_pytest_markers": False, "has_async_tests": False, "has_fixtures": False, "has_proper_imports": False, "has_class_based_tests": False, "test_count": 0, "async_test_count": 0, "fixture_count": 0, } # Check imports for node in ast.walk(self.tree): if isinstance(node, ast.ImportFrom): if node.module == "pytest": results["has_proper_imports"] = True elif isinstance(node, ast.Import): for alias in node.names: if alias.name == "pytest": results["has_proper_imports"] = True # Check for pytest markers, fixtures, and test functions for node in ast.walk(self.tree): if isinstance(node, ast.FunctionDef): # Check for test functions if node.name.startswith("test_"): results["test_count"] += 1 # Check for async tests if isinstance(node, ast.AsyncFunctionDef): results["has_async_tests"] = True results["async_test_count"] += 1 # Check for fixtures for decorator in node.decorator_list: if isinstance(decorator, ast.Attribute): if decorator.attr == "fixture": results["has_fixtures"] = True results["fixture_count"] += 1 elif isinstance(decorator, ast.Name): if decorator.id == "fixture": results["has_fixtures"] = True results["fixture_count"] += 1 elif isinstance(node, ast.AsyncFunctionDef): if node.name.startswith("test_"): results["test_count"] += 1 results["has_async_tests"] = True results["async_test_count"] += 1 # Check for pytest markers marker_pattern = r"@pytest\.mark\.\w+" if re.search(marker_pattern, self.content): results["has_pytest_markers"] = True # Check for class-based tests for node in ast.walk(self.tree): if isinstance(node, ast.ClassDef): if node.name.startswith("Test"): results["has_class_based_tests"] = True break return results def validate_async_patterns(self) -> dict[str, Any]: """Validate async/await patterns.""" results = { "proper_async_await": True, "has_asyncio_imports": False, "async_fixtures_marked": True, "issues": [], } # Check for asyncio imports if "import asyncio" in self.content or "from asyncio" in self.content: results["has_asyncio_imports"] = True # Check async function patterns for node in ast.walk(self.tree): if isinstance(node, ast.AsyncFunctionDef): # Check if async test functions are properly marked if node.name.startswith("test_"): for decorator in node.decorator_list: if isinstance(decorator, ast.Attribute): if ( hasattr(decorator.value, "attr") and decorator.value.attr == "mark" and decorator.attr == "asyncio" ): pass elif isinstance(decorator, ast.Call): if ( isinstance(decorator.func, ast.Attribute) and hasattr(decorator.func.value, "attr") and decorator.func.value.attr == "mark" and decorator.func.attr == "asyncio" ): pass # Not all test environments require explicit asyncio marking # Modern pytest-asyncio auto-detects async tests return results def validate_mock_usage(self) -> dict[str, Any]: """Validate mock usage patterns.""" results = { "has_mocks": False, "has_async_mocks": False, "has_patch_usage": False, "proper_mock_imports": False, } # Check mock imports mock_imports = ["Mock", "AsyncMock", "MagicMock", "patch"] for imp in mock_imports: if ( f"from unittest.mock import {imp}" in self.content or f"import {imp}" in self.content ): results["proper_mock_imports"] = True results["has_mocks"] = True if imp == "AsyncMock": results["has_async_mocks"] = True if imp == "patch": results["has_patch_usage"] = True return results class TestParallelResearchTestSuites: """Test the test suites for parallel research functionality.""" def test_parallel_research_orchestrator_tests_structure(self): """Test structure of ParallelResearchOrchestrator test suite.""" test_file = Path(__file__).parent / "test_parallel_research_orchestrator.py" assert test_file.exists(), "ParallelResearchOrchestrator test file should exist" validator = TestSuiteValidator(str(test_file)) results = validator.validate_pytest_patterns() assert results["test_count"] > 0, "Should have test functions" assert results["has_async_tests"], "Should have async tests" assert results["has_fixtures"], "Should have fixtures" assert results["has_class_based_tests"], "Should have class-based tests" assert results["async_test_count"] > 0, "Should have async test functions" def test_deep_research_parallel_execution_tests_structure(self): """Test structure of DeepResearchAgent parallel execution test suite.""" test_file = Path(__file__).parent / "test_deep_research_parallel_execution.py" assert test_file.exists(), "DeepResearchAgent parallel test file should exist" validator = TestSuiteValidator(str(test_file)) results = validator.validate_pytest_patterns() assert results["test_count"] > 0, "Should have test functions" assert results["has_async_tests"], "Should have async tests" assert results["has_fixtures"], "Should have fixtures" assert results["has_class_based_tests"], "Should have class-based tests" def test_orchestration_logging_tests_structure(self): """Test structure of OrchestrationLogger test suite.""" test_file = Path(__file__).parent / "test_orchestration_logging.py" assert test_file.exists(), "OrchestrationLogger test file should exist" validator = TestSuiteValidator(str(test_file)) results = validator.validate_pytest_patterns() assert results["test_count"] > 0, "Should have test functions" assert results["has_async_tests"], "Should have async tests" assert results["has_fixtures"], "Should have fixtures" assert results["has_class_based_tests"], "Should have class-based tests" def test_parallel_research_integration_tests_structure(self): """Test structure of parallel research integration test suite.""" test_file = Path(__file__).parent / "test_parallel_research_integration.py" assert test_file.exists(), ( "Parallel research integration test file should exist" ) validator = TestSuiteValidator(str(test_file)) results = validator.validate_pytest_patterns() assert results["test_count"] > 0, "Should have test functions" assert results["has_async_tests"], "Should have async tests" assert results["has_fixtures"], "Should have fixtures" assert results["has_class_based_tests"], "Should have class-based tests" assert results["has_pytest_markers"], ( "Should have pytest markers (like @pytest.mark.integration)" ) def test_async_patterns_validation(self): """Test that async patterns are properly implemented across all test suites.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_orchestration_logging.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): validator = TestSuiteValidator(str(file_path)) results = validator.validate_async_patterns() assert results["proper_async_await"], ( f"Async patterns should be correct in {test_file}" ) assert results["has_asyncio_imports"], ( f"Should import asyncio in {test_file}" ) def test_mock_usage_patterns(self): """Test that mock usage patterns are consistent across test suites.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_orchestration_logging.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): validator = TestSuiteValidator(str(file_path)) results = validator.validate_mock_usage() assert results["has_mocks"], f"Should use mocks in {test_file}" assert results["proper_mock_imports"], ( f"Should have proper mock imports in {test_file}" ) # For async-heavy test files, should use AsyncMock if test_file in [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_parallel_research_integration.py", ]: assert results["has_async_mocks"], ( f"Should use AsyncMock in {test_file}" ) def test_test_coverage_completeness(self): """Test that test coverage is comprehensive for parallel research functionality.""" # Define expected test categories for each component expected_test_categories = { "test_parallel_research_orchestrator.py": [ "config", "task", "orchestrator", "distribution", "result", "integration", ], "test_deep_research_parallel_execution.py": [ "agent", "subagent", "execution", "synthesis", "integration", ], "test_orchestration_logging.py": [ "logger", "decorator", "context", "utility", "integrated", "load", ], "test_parallel_research_integration.py": [ "endtoend", "scalability", "logging", "error", "data", ], } for test_file, expected_categories in expected_test_categories.items(): file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text().lower() for category in expected_categories: assert category in content, ( f"Should have {category} tests in {test_file}" ) def test_docstring_quality(self): """Test that test files have proper docstrings.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_orchestration_logging.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text() # Should have module docstring assert '"""' in content, f"Should have docstrings in {test_file}" # Should describe what is being tested docstring_keywords = ["test", "functionality", "cover", "suite"] first_docstring = content.split('"""')[1].lower() assert any( keyword in first_docstring for keyword in docstring_keywords ), f"Module docstring should describe testing purpose in {test_file}" def test_import_safety(self): """Test that imports are safe and avoid circular dependencies.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_orchestration_logging.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text() # Should not have circular import patterns lines = content.split("\n") import_lines = [ line for line in lines if line.strip().startswith(("import ", "from ")) ] # Basic validation that imports are structured assert len(import_lines) > 0, ( f"Should have import statements in {test_file}" ) # Should import pytest pytest_imported = any("pytest" in line for line in import_lines) assert pytest_imported, f"Should import pytest in {test_file}" def test_fixture_best_practices(self): """Test that fixtures follow best practices.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_orchestration_logging.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text() # If file has fixtures, they should be properly structured if "@pytest.fixture" in content: # Should have fixture decorators assert "def " in content, ( f"Fixtures should be functions in {test_file}" ) # Common fixture patterns should be present fixture_patterns = ["yield", "return", "Mock", "config"] has_fixture_pattern = any( pattern in content for pattern in fixture_patterns ) assert has_fixture_pattern, ( f"Should have proper fixture patterns in {test_file}" ) def test_error_handling_coverage(self): """Test that error handling scenarios are covered.""" test_files = [ "test_parallel_research_orchestrator.py", "test_deep_research_parallel_execution.py", "test_parallel_research_integration.py", ] for test_file in test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text().lower() # Should test error scenarios error_keywords = [ "error", "exception", "timeout", "failure", "fallback", ] has_error_tests = any(keyword in content for keyword in error_keywords) assert has_error_tests, f"Should test error scenarios in {test_file}" def test_performance_testing_coverage(self): """Test that performance characteristics are tested.""" performance_test_files = [ "test_parallel_research_orchestrator.py", "test_parallel_research_integration.py", ] for test_file in performance_test_files: file_path = Path(__file__).parent / test_file if file_path.exists(): content = file_path.read_text().lower() # Should test performance characteristics perf_keywords = [ "performance", "timing", "efficiency", "concurrent", "parallel", ] has_perf_tests = any(keyword in content for keyword in perf_keywords) assert has_perf_tests, ( f"Should test performance characteristics in {test_file}" ) def test_integration_test_markers(self): """Test that integration tests are properly marked.""" integration_file = ( Path(__file__).parent / "test_parallel_research_integration.py" ) if integration_file.exists(): content = integration_file.read_text() # Should have integration markers assert "@pytest.mark.integration" in content, ( "Should mark integration tests" ) # Should have integration test classes integration_patterns = ["TestParallel", "Integration", "EndToEnd"] has_integration_classes = any( pattern in content for pattern in integration_patterns ) assert has_integration_classes, "Should have integration test classes"