Module 2: Expand Your Agent Fleet

# === Code Review Analyzer ===
# Contract-name: xpander_run_action
# Purpose: Automated code review and quality analysis

import os
import re
import ast
import subprocess
from pathlib import Path
from typing import Dict, List, Any

def xpander_run_action(file_path: str, review_type: str = "comprehensive"):
    """
    Perform automated code review and quality analysis
    
    Args:
        file_path: Path to the code file to review
        review_type: Type of review ("security", "performance", "comprehensive")
        
    Returns:
        dict: Review results with issues and recommendations
    """
    
    try:
        if not os.path.exists(file_path):
            return {"error": f"File not found: {file_path}"}
        
        with open(file_path, 'r', encoding='utf-8') as f:
            code_content = f.read()
        
        file_ext = Path(file_path).suffix.lower()
        
        if file_ext == '.py':
            return review_python_code(code_content, review_type)
        elif file_ext in ['.js', '.ts']:
            return review_javascript_code(code_content, review_type)
        else:
            return review_generic_code(code_content, review_type)
            
    except Exception as e:
        return {"error": f"Review failed: {str(e)}"}

def review_python_code(code: str, review_type: str) -> Dict[str, Any]:
    """Comprehensive Python code review"""
    issues = []
    suggestions = []
    metrics = {"complexity": 0, "maintainability": 100, "security_score": 100}
    
    # Security checks
    if review_type in ["security", "comprehensive"]:
        security_issues = check_python_security(code)
        issues.extend(security_issues)
        if security_issues:
            metrics["security_score"] -= len(security_issues) * 10
    
    # Performance checks  
    if review_type in ["performance", "comprehensive"]:
        performance_issues = check_python_performance(code)
        issues.extend(performance_issues)
    
    # Code quality checks
    if review_type == "comprehensive":
        quality_issues = check_python_quality(code)
        issues.extend(quality_issues)
        
        # Calculate complexity
        try:
            tree = ast.parse(code)
            complexity = calculate_complexity(tree)
            metrics["complexity"] = complexity
            
            if complexity > 10:
                issues.append({
                    "type": "complexity",
                    "severity": "high",
                    "message": f"Cyclomatic complexity too high: {complexity}",
                    "suggestion": "Consider breaking down complex functions into smaller ones"
                })
        except SyntaxError:
            issues.append({
                "type": "syntax",
                "severity": "critical", 
                "message": "Syntax error in code",
                "suggestion": "Fix syntax errors before proceeding"
            })
    
    # Generate overall recommendations
    if len(issues) == 0:
        suggestions.append("Code quality is excellent! No issues found.")
    elif len(issues) <= 3:
        suggestions.append("Minor issues found. Consider addressing them for better code quality.")
    else:
        suggestions.append("Multiple issues found. Recommend comprehensive refactoring.")
    
    return {
        "file_type": "python",
        "issues": issues,
        "suggestions": suggestions,
        "metrics": metrics,
        "summary": f"Found {len(issues)} issues across {review_type} review"
    }

def check_python_security(code: str) -> List[Dict[str, str]]:
    """Check for common Python security issues"""
    issues = []
    
    # Check for dangerous imports
    dangerous_patterns = [
        (r'import\s+pickle', "Using pickle can be unsafe with untrusted data"),
        (r'eval\s*\(', "eval() can execute arbitrary code - very dangerous"),
        (r'exec\s*\(', "exec() can execute arbitrary code - very dangerous"),
        (r'subprocess\.call\([^)]*shell=True', "shell=True in subprocess can be dangerous"),
        (r'os\.system\s*\(', "os.system() can be vulnerable to injection attacks"),
    ]
    
    for pattern, message in dangerous_patterns:
        if re.search(pattern, code):
            issues.append({
                "type": "security",
                "severity": "high",
                "message": message,
                "suggestion": "Use safer alternatives or validate inputs thoroughly"
            })
    
    # Check for hardcoded secrets
    secret_patterns = [
        (r'password\s*=\s*["\'][^"\']+["\']', "Hardcoded password detected"),
        (r'api_key\s*=\s*["\'][^"\']+["\']', "Hardcoded API key detected"),
        (r'secret\s*=\s*["\'][^"\']+["\']', "Hardcoded secret detected"),
    ]
    
    for pattern, message in secret_patterns:
        if re.search(pattern, code, re.IGNORECASE):
            issues.append({
                "type": "security",
                "severity": "critical",
                "message": message,
                "suggestion": "Use environment variables or secure key management"
            })
    
    return issues

def check_python_performance(code: str) -> List[Dict[str, str]]:
    """Check for common Python performance issues"""
    issues = []
    
    performance_patterns = [
        (r'for\s+\w+\s+in\s+range\(len\([^)]+\)\):', "Use enumerate() instead of range(len())"),
        (r'\+\s*=.*\[.*\]', "String concatenation in loop - use join() instead"),
        (r'\.append\(.*\)\s*$', "Multiple appends - consider list comprehension"),
    ]
    
    for pattern, message in performance_patterns:
        if re.search(pattern, code, re.MULTILINE):
            issues.append({
                "type": "performance",
                "severity": "medium",
                "message": message,
                "suggestion": "Optimize for better performance"
            })
    
    return issues

def check_python_quality(code: str) -> List[Dict[str, str]]:
    """Check for code quality issues"""
    issues = []
    
    # Check for missing docstrings
    if 'def ' in code and '"""' not in code and "'''" not in code:
        issues.append({
            "type": "documentation",
            "severity": "medium",
            "message": "Functions missing docstrings",
            "suggestion": "Add comprehensive docstrings to all functions"
        })
    
    # Check for long lines
    long_lines = [i+1 for i, line in enumerate(code.split('\n')) if len(line) > 88]
    if long_lines:
        issues.append({
            "type": "style",
            "severity": "low",
            "message": f"Lines too long: {long_lines}",
            "suggestion": "Keep lines under 88 characters for better readability"
        })
    
    # Check for magic numbers
    magic_numbers = re.findall(r'\b\d{2,}\b', code)
    if magic_numbers:
        issues.append({
            "type": "maintainability",
            "severity": "medium", 
            "message": "Magic numbers found",
            "suggestion": "Define constants for magic numbers"
        })
    
    return issues

def calculate_complexity(tree) -> int:
    """Calculate cyclomatic complexity"""
    complexity = 1  # Base complexity
    
    for node in ast.walk(tree):
        if isinstance(node, (ast.If, ast.While, ast.For, ast.AsyncFor)):
            complexity += 1
        elif isinstance(node, ast.FunctionDef):
            complexity += 1
    
    return complexity

def review_javascript_code(code: str, review_type: str) -> Dict[str, Any]:
    """Basic JavaScript code review"""
    issues = []
    
    # Basic JavaScript checks
    if 'eval(' in code:
        issues.append({
            "type": "security",
            "severity": "high",
            "message": "eval() usage detected",
            "suggestion": "Avoid eval() - use safer alternatives"
        })
    
    if 'var ' in code:
        issues.append({
            "type": "style",
            "severity": "low",
            "message": "var usage detected",
            "suggestion": "Use let or const instead of var"
        })
    
    return {
        "file_type": "javascript",
        "issues": issues,
        "suggestions": ["Consider using ESLint for more comprehensive analysis"],
        "metrics": {"complexity": len(issues)},
        "summary": f"Basic JavaScript review found {len(issues)} issues"
    }

def review_generic_code(code: str, review_type: str) -> Dict[str, Any]:
    """Generic code review for unsupported file types"""
    issues = []
    
    # Basic checks that apply to any code
    lines = code.split('\n')
    if len(lines) > 500:
        issues.append({
            "type": "maintainability",
            "severity": "medium",
            "message": "File is very large",
            "suggestion": "Consider splitting into smaller files"
        })
    
    return {
        "file_type": "generic",
        "issues": issues,
        "suggestions": ["Manual review recommended for this file type"],
        "metrics": {"lines": len(lines)},
        "summary": f"Generic review completed"
    }

# === Test Generator ===
# Contract-name: xpander_run_action
# Purpose: Generate comprehensive test cases for code

import os
import ast
import re
from typing import Dict, List, Any

def xpander_run_action(source_file: str, test_framework: str = "pytest"):
    """
    Generate test cases for a given source file
    
    Args:
        source_file: Path to the source code file
        test_framework: Testing framework to use ("pytest", "unittest", "jest")
        
    Returns:
        dict: Generated test code and test coverage recommendations
    """
    
    try:
        if not os.path.exists(source_file):
            return {"error": f"Source file not found: {source_file}"}
        
        with open(source_file, 'r') as f:
            source_code = f.read()
        
        file_ext = os.path.splitext(source_file)[1].lower()
        
        if file_ext == '.py':
            return generate_python_tests(source_code, test_framework)
        elif file_ext in ['.js', '.ts']:
            return generate_javascript_tests(source_code, test_framework)
        else:
            return {"error": f"Unsupported file type: {file_ext}"}
            
    except Exception as e:
        return {"error": f"Test generation failed: {str(e)}"}

def generate_python_tests(source_code: str, framework: str) -> Dict[str, Any]:
    """Generate Python test cases"""
    
    try:
        tree = ast.parse(source_code)
        functions = extract_functions(tree)
        classes = extract_classes(tree)
        
        if framework == "pytest":
            test_code = generate_pytest_code(functions, classes)
        else:
            test_code = generate_unittest_code(functions, classes)
        
        test_recommendations = generate_test_recommendations(functions, classes)
        
        return {
            "test_code": test_code,
            "test_file_name": "test_" + os.path.basename(source_file),
            "framework": framework,
            "functions_found": len(functions),
            "classes_found": len(classes),
            "recommendations": test_recommendations,
            "coverage_suggestions": generate_coverage_suggestions(functions, classes)
        }
        
    except SyntaxError as e:
        return {"error": f"Source code has syntax errors: {str(e)}"}

def extract_functions(tree) -> List[Dict[str, Any]]:
    """Extract function definitions from AST"""
    functions = []
    
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            func_info = {
                "name": node.name,
                "args": [arg.arg for arg in node.args.args],
                "returns": bool(any(isinstance(n, ast.Return) for n in ast.walk(node))),
                "has_docstring": bool(ast.get_docstring(node)),
                "line_number": node.lineno
            }
            functions.append(func_info)
    
    return functions

def extract_classes(tree) -> List[Dict[str, Any]]:
    """Extract class definitions from AST"""
    classes = []
    
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            methods = [n.name for n in node.body if isinstance(n, ast.FunctionDef)]
            class_info = {
                "name": node.name,
                "methods": methods,
                "has_init": "__init__" in methods,
                "line_number": node.lineno
            }
            classes.append(class_info)
    
    return classes

def generate_pytest_code(functions: List[Dict], classes: List[Dict]) -> str:
    """Generate pytest test code"""
    
    test_code = """import pytest
from unittest.mock import Mock, patch
# Import your modules here
# from your_module import YourClass, your_function

"""
    
    # Generate function tests
    for func in functions:
        test_code += f"""
def test_{func['name']}_basic():
    \"\"\"Test basic functionality of {func['name']}\"\"\"
    # Arrange
    # TODO: Set up test data
    
    # Act
    # TODO: Call {func['name']} with test data
    # result = {func['name']}({', '.join(['test_' + arg for arg in func['args']])})
    
    # Assert
    # TODO: Verify expected behavior
    # assert result == expected_value
    pass

def test_{func['name']}_edge_cases():
    \"\"\"Test edge cases for {func['name']}\"\"\"
    # TODO: Test edge cases like empty inputs, None values, etc.
    pass

def test_{func['name']}_error_handling():
    \"\"\"Test error handling for {func['name']}\"\"\"
    # TODO: Test that appropriate exceptions are raised
    # with pytest.raises(ValueError):
    #     {func['name']}(invalid_input)
    pass
"""
    
    # Generate class tests
    for cls in classes:
        test_code += f"""
class Test{cls['name']}:
    \"\"\"Test suite for {cls['name']}\"\"\"
    
    def setup_method(self):
        \"\"\"Set up test fixtures before each test method.\"\"\"
        # TODO: Initialize test instance
        # self.instance = {cls['name']}()
        pass
    
    def test_initialization(self):
        \"\"\"Test {cls['name']} initialization\"\"\"
        # TODO: Test object creation
        pass
"""
        
        for method in cls['methods']:
            if method != '__init__':
                test_code += f"""
    def test_{method}(self):
        \"\"\"Test {method} method\"\"\"
        # TODO: Test {method} functionality
        pass
"""
    
    return test_code

def generate_test_recommendations(functions: List[Dict], classes: List[Dict]) -> List[str]:
    """Generate test recommendations"""
    recommendations = []
    
    if not functions and not classes:
        recommendations.append("No functions or classes found to test")
        return recommendations
    
    recommendations.append(f"Create tests for {len(functions)} functions and {len(classes)} classes")
    
    # Check for functions without docstrings
    undocumented = [f['name'] for f in functions if not f['has_docstring']]
    if undocumented:
        recommendations.append(f"Add docstrings to functions: {', '.join(undocumented)}")
    
    # Recommend specific test types
    recommendations.extend([
        "Include positive test cases (happy path)",
        "Include negative test cases (error conditions)",
        "Test edge cases and boundary values",
        "Mock external dependencies",
        "Verify error messages and exception types",
        "Test with different data types and sizes"
    ])
    
    return recommendations

def generate_coverage_suggestions(functions: List[Dict], classes: List[Dict]) -> List[str]:
    """Generate test coverage suggestions"""
    suggestions = [
        "Aim for >90% code coverage",
        "Use coverage.py to measure test coverage",
        "Focus on testing business logic thoroughly",
        "Don't forget to test error handling paths"
    ]
    
    if classes:
        suggestions.append("Test all public methods of each class")
        suggestions.append("Test class initialization with various parameters")
    
    if functions:
        suggestions.append("Test functions with various input combinations")
        suggestions.append("Verify return values and side effects")
    
    return suggestions