| """ |
| Code Tester Module - Run tests and generate coverage reports. |
| |
| This module handles: |
| - Running pytest tests |
| - Generating code coverage reports |
| - Running linting checks (pylint, flake8) |
| - Collecting test results |
| """ |
|
|
| import os |
| import subprocess |
| import logging |
| from pathlib import Path |
| from typing import Dict, Any, List |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class CodeTester: |
| """Run tests and generate coverage reports.""" |
| |
| def __init__(self, config: Dict[str, Any], project_root: str): |
| """ |
| Initialize the code tester. |
| |
| Args: |
| config: Configuration dictionary for code tester |
| project_root: Root directory of the project |
| """ |
| self.config = config |
| self.project_root = Path(project_root) |
| self.results = { |
| "pytest_results": {}, |
| "coverage_results": {}, |
| "pylint_results": {}, |
| "flake8_results": {}, |
| "total_tests": 0, |
| "passed_tests": 0, |
| "failed_tests": 0, |
| "coverage_percentage": 0.0, |
| "errors": [] |
| } |
| |
| def run(self) -> Dict[str, Any]: |
| """ |
| Run all code tests. |
| |
| Returns: |
| Dictionary containing results of code tests |
| """ |
| logger.info("Starting code tests...") |
| |
| if not self.config.get("enabled", True): |
| logger.info("Code tester is disabled in configuration") |
| return self.results |
| |
| |
| if self.config.get("run_pytest", True): |
| self._run_pytest() |
| |
| |
| if self.config.get("run_pylint", True): |
| self._run_pylint() |
| |
| |
| if self.config.get("run_flake8", True): |
| self._run_flake8() |
| |
| logger.info(f"Code tests completed. {self.results['passed_tests']}/{self.results['total_tests']} tests passed") |
| return self.results |
| |
| def _run_pytest(self): |
| """Run pytest with coverage.""" |
| logger.info("Running pytest...") |
| |
| try: |
| test_dirs = self.config.get("test_directories", ["backend"]) |
| pytest_args = self.config.get("pytest_args", ["-v"]) |
| generate_coverage = self.config.get("generate_coverage", True) |
| |
| |
| cmd = ["pytest"] |
| cmd.extend(pytest_args) |
| |
| if generate_coverage: |
| cmd.extend([ |
| "--cov=.", |
| "--cov-report=html:quality_reports/coverage", |
| "--cov-report=term" |
| ]) |
| |
| |
| for test_dir in test_dirs: |
| test_path = self.project_root / test_dir |
| if test_path.exists(): |
| cmd.append(str(test_path)) |
| |
| result = subprocess.run( |
| cmd, |
| capture_output=True, |
| text=True, |
| cwd=self.project_root |
| ) |
| |
| |
| output_lines = result.stdout.split('\n') |
| |
| |
| for line in output_lines: |
| if " passed" in line or " failed" in line: |
| |
| parts = line.split() |
| for i, part in enumerate(parts): |
| if part == "passed": |
| try: |
| self.results["passed_tests"] = int(parts[i-1]) |
| except (ValueError, IndexError): |
| pass |
| elif part == "failed": |
| try: |
| self.results["failed_tests"] = int(parts[i-1]) |
| except (ValueError, IndexError): |
| pass |
| |
| |
| if "TOTAL" in line and "%" in line: |
| parts = line.split() |
| for part in parts: |
| if "%" in part: |
| try: |
| self.results["coverage_percentage"] = float(part.replace("%", "")) |
| except ValueError: |
| pass |
| |
| self.results["total_tests"] = self.results["passed_tests"] + self.results["failed_tests"] |
| |
| self.results["pytest_results"] = { |
| "status": "PASSED" if result.returncode == 0 else "FAILED", |
| "return_code": result.returncode, |
| "output": result.stdout, |
| "total_tests": self.results["total_tests"], |
| "passed": self.results["passed_tests"], |
| "failed": self.results["failed_tests"] |
| } |
| |
| if result.returncode == 0: |
| logger.info(f"β Pytest passed - {self.results['passed_tests']} tests") |
| else: |
| logger.warning(f"β Pytest failed - {self.results['failed_tests']} failures") |
| |
| if generate_coverage: |
| logger.info(f"Code coverage: {self.results['coverage_percentage']}%") |
| |
| except FileNotFoundError: |
| error_msg = "pytest is not installed. Install with: pip install pytest pytest-cov" |
| logger.error(error_msg) |
| self.results["errors"].append(error_msg) |
| except Exception as e: |
| logger.error(f"Error running pytest: {str(e)}") |
| self.results["errors"].append(f"pytest error: {str(e)}") |
| |
| def _run_pylint(self): |
| """Run pylint for code quality checks.""" |
| logger.info("Running pylint...") |
| |
| try: |
| |
| python_files = [] |
| for test_dir in self.config.get("test_directories", ["backend"]): |
| test_path = self.project_root / test_dir |
| if test_path.exists(): |
| python_files.extend(test_path.glob("**/*.py")) |
| |
| if not python_files: |
| logger.warning("No Python files found for pylint") |
| return |
| |
| cmd = [ |
| "pylint", |
| *[str(f) for f in python_files[:20]] |
| ] |
| |
| result = subprocess.run( |
| cmd, |
| capture_output=True, |
| text=True, |
| cwd=self.project_root |
| ) |
| |
| |
| output_lines = result.stdout.split('\n') |
| score = 0.0 |
| |
| for line in output_lines: |
| if "Your code has been rated at" in line: |
| try: |
| score_str = line.split("rated at ")[1].split("/")[0] |
| score = float(score_str) |
| except (IndexError, ValueError): |
| pass |
| |
| threshold = self.config.get("pylint_threshold", 7.0) |
| |
| self.results["pylint_results"] = { |
| "status": "PASSED" if score >= threshold else "FAILED", |
| "score": score, |
| "threshold": threshold, |
| "output": result.stdout[:1000] |
| } |
| |
| if score >= threshold: |
| logger.info(f"β Pylint passed - Score: {score}/10") |
| else: |
| logger.warning(f"β Pylint failed - Score: {score}/10 (threshold: {threshold})") |
| |
| except FileNotFoundError: |
| error_msg = "pylint is not installed. Install with: pip install pylint" |
| logger.error(error_msg) |
| self.results["errors"].append(error_msg) |
| except Exception as e: |
| logger.error(f"Error running pylint: {str(e)}") |
| self.results["errors"].append(f"pylint error: {str(e)}") |
| |
| def _run_flake8(self): |
| """Run flake8 for style checks.""" |
| logger.info("Running flake8...") |
| |
| try: |
| test_dirs = self.config.get("test_directories", ["backend"]) |
| |
| cmd = ["flake8"] |
| |
| for test_dir in test_dirs: |
| test_path = self.project_root / test_dir |
| if test_path.exists(): |
| cmd.append(str(test_path)) |
| |
| result = subprocess.run( |
| cmd, |
| capture_output=True, |
| text=True, |
| cwd=self.project_root |
| ) |
| |
| |
| issues = result.stdout.split('\n') |
| issue_count = len([line for line in issues if line.strip()]) |
| |
| self.results["flake8_results"] = { |
| "status": "PASSED" if result.returncode == 0 else "FAILED", |
| "issue_count": issue_count, |
| "output": result.stdout[:1000] |
| } |
| |
| if result.returncode == 0: |
| logger.info("β Flake8 passed - No style issues") |
| else: |
| logger.warning(f"β Flake8 found {issue_count} style issues") |
| |
| except FileNotFoundError: |
| error_msg = "flake8 is not installed. Install with: pip install flake8" |
| logger.error(error_msg) |
| self.results["errors"].append(error_msg) |
| except Exception as e: |
| logger.error(f"Error running flake8: {str(e)}") |
| self.results["errors"].append(f"flake8 error: {str(e)}") |
| |
| def get_summary(self) -> str: |
| """Get a summary of code test results.""" |
| summary = f""" |
| Code Test Summary: |
| - Total tests: {self.results['total_tests']} |
| - Passed: {self.results['passed_tests']} |
| - Failed: {self.results['failed_tests']} |
| - Code coverage: {self.results['coverage_percentage']}% |
| |
| Quality Checks: |
| - Pylint score: {self.results.get('pylint_results', {}).get('score', 'N/A')}/10 |
| - Flake8 issues: {self.results.get('flake8_results', {}).get('issue_count', 'N/A')} |
| """ |
| return summary |
|
|