learningcircuit--local-deep-research/tests/run_all_tests.py

#!/usr/bin/env python3
"""
Comprehensive Test Runner for Local Deep Research

This script orchestrates all test types and provides different test profiles
for various scenarios (CI/CD, development, full validation).

Usage:
    python tests/run_all_tests.py [profile]

Profiles:
    fast          - Quick health checks and unit tests (< 30s)
    standard      - Standard development testing (< 5min)
    full          - Optimized full testing, no redundant runs (< 10min)
    comprehensive - Legacy full testing, includes redundant runs (< 15min)
    ci            - CI/CD optimized tests
    unit-only     - Unit tests only, no server dependencies (< 2min)
"""

import argparse
import os
import subprocess
from subprocess import CalledProcessError
import sys
import time
from pathlib import Path
from typing import List, Tuple
from functools import cache


@cache
def find_exe(tool_name: str) -> Path:
    """
    Finds a particular command-line tool.

    Args:
        tool_name: The name of the tool.

    Returns:
        The path to the tool.

    """
    which_output = ""
    try:
        which_result = subprocess.run(
            ["/usr/bin/which", tool_name], check=True, capture_output=True
        )
        which_output = which_result.stdout.decode("utf8")
    except CalledProcessError as err:
        if err.returncode != 1:  # pragma: no cover
            # If it returns 1, it's probably because the executable does not
            # exist. Otherwise, something weird happened.
            raise err
        pass
    if not which_output:
        raise OSError(f"Could not find '{tool_name}'. Is it installed?")

    tool_path = Path(which_output.rstrip("\n"))
    return tool_path


class TestRunner:
    def __init__(self):
        self.project_root = Path(__file__).parent.parent.absolute()
        self.tests_dir = self.project_root / "tests"
        self.results: List[Tuple[str, bool, float]] = []
        self.server_process = None

    def log(self, message: str, level: str = "INFO"):
        """Log a message with timestamp."""
        timestamp = time.strftime("%H:%M:%S")
        prefix = (
            "✅" if level == "SUCCESS" else "❌" if level == "ERROR" else "📋"
        )
        print(f"[{timestamp}] {prefix} {message}")

    def run_command(
        self, cmd: List[str], name: str, cwd: str = None, timeout: int = 300
    ) -> bool:
        """Run a command and track its result."""
        self.log(f"Running {name}...")
        start_time = time.time()

        try:
            # Set environment
            env = os.environ.copy()
            env["PYTHONPATH"] = str(self.project_root)

            result = subprocess.run(
                cmd,
                cwd=cwd or self.project_root,
                env=env,
                capture_output=True,
                text=True,
                timeout=timeout,
            )

            duration = time.time() - start_time
            success = result.returncode == 0

            if success:
                self.log(
                    f"{name} completed successfully ({duration:.1f}s)",
                    "SUCCESS",
                )
            else:
                self.log(f"{name} failed ({duration:.1f}s)", "ERROR")
                print(f"\n{'=' * 60}")
                print(f"DETAILED ERROR OUTPUT FOR: {name}")
                print(f"{'=' * 60}")

                # Print detailed output for failures with full context
                if result.stdout:
                    print("STDOUT:")
                    lines = result.stdout.strip().split("\n")

                    # Show test summary/results
                    for line in lines[-15:]:  # Check last 15 lines for summary
                        if any(
                            keyword in line.lower()
                            for keyword in [
                                "failed",
                                "passed",
                                "error",
                                "collected",
                                "warnings",
                            ]
                        ):
                            print(f"  {line}")

                    # Show specific failed tests with more context
                    print("\nFAILED TESTS WITH CONTEXT:")
                    for i, line in enumerate(lines):
                        if "FAILED" in line and "::" in line:
                            print(f"  {line.strip()}")
                            # Show next few lines for error details
                            for j in range(i + 1, min(i + 10, len(lines))):
                                if lines[j].strip() and not lines[j].startswith(
                                    "===="
                                ):
                                    print(f"    {lines[j].strip()}")
                                if (
                                    "KeyError" in lines[j]
                                    or "Exception" in lines[j]
                                ):
                                    break

                    # Show full stack traces for exceptions
                    print("\nFULL STACK TRACES:")
                    in_traceback = False
                    traceback_lines = []
                    for line in lines:
                        if "Traceback (most recent call last):" in line:
                            in_traceback = True
                            traceback_lines = [line]
                        elif in_traceback:
                            traceback_lines.append(line)
                            if line.strip() and not line.startswith(
                                (" ", "\t", "File ", "  ")
                            ):
                                # End of traceback
                                for tb_line in traceback_lines:
                                    print(f"  {tb_line}")
                                print()
                                in_traceback = False
                                traceback_lines = []
                        elif any(
                            error_type in line
                            for error_type in [
                                "KeyError:",
                                "ImportError:",
                                "ModuleNotFoundError:",
                                "AttributeError:",
                                "Exception:",
                            ]
                        ):
                            print(f"  ERROR: {line.strip()}")

                    # Show import errors with context
                    print("\nIMPORT ERRORS:")
                    for i, line in enumerate(lines):
                        if (
                            "ImportError" in line
                            or "ModuleNotFoundError" in line
                        ):
                            print(f"  {line.strip()}")
                            # Show preceding line for context
                            if i > 0:
                                print(f"    Context: {lines[i - 1].strip()}")

                    # Show KeyError details specifically
                    print("\nKEYERROR DETAILS:")
                    for i, line in enumerate(lines):
                        if "KeyError" in line:
                            print(f"  {line.strip()}")
                            # Show surrounding context
                            for j in range(
                                max(0, i - 3), min(len(lines), i + 4)
                            ):
                                if j != i:
                                    print(
                                        f"    [{j - i:+d}] {lines[j].strip()}"
                                    )

                    # Show pytest short test summary if available
                    print("\nPYTEST SHORT SUMMARY:")
                    in_summary = False
                    for line in lines:
                        if "short test summary info" in line.lower():
                            in_summary = True
                        elif in_summary:
                            if line.startswith("="):
                                break
                            print(f"  {line.strip()}")

                if result.stderr:
                    print("\nSTDERR:")
                    stderr_lines = result.stderr.strip().split("\n")
                    for line in stderr_lines[-10:]:  # Last 10 lines
                        print(f"  {line}")

                print(f"{'=' * 60}\n")

            self.results.append((name, success, duration))
            return success

        except subprocess.TimeoutExpired:
            duration = time.time() - start_time
            self.log(f"{name} timed out after {timeout}s", "ERROR")
            self.results.append((name, False, duration))
            return False
        except Exception as e:
            duration = time.time() - start_time
            self.log(f"{name} failed with exception: {e}", "ERROR")
            self.results.append((name, False, duration))
            return False

    def check_server_running(self) -> bool:
        """Check if the web server is running."""
        try:
            import requests

            response = requests.get(
                "http://127.0.0.1:5000/api/v1/health", timeout=5
            )
            return response.status_code == 200
        except Exception:
            return False

    def start_server(self) -> bool:
        """Start the web server using pdm run ldr-web."""
        if self.check_server_running():
            self.log("Server already running at http://127.0.0.1:5000")
            return True

        self.log("Starting web server with 'pdm run ldr-web'...")
        try:
            import subprocess
            import time

            # Start server in background
            self.server_process = subprocess.Popen(
                ["pdm", "run", "ldr-web"],
                cwd=self.project_root,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )

            # Wait for server to start (max 45 seconds)
            for i in range(45):
                time.sleep(1)  # allow: unmarked-sleep
                if self.check_server_running():
                    self.log(
                        f"Server started successfully after {i + 1} seconds"
                    )
                    return True
                if self.server_process.poll() is not None:
                    # Process died
                    stdout, stderr = self.server_process.communicate()
                    self.log(
                        f"Server failed to start. Exit code: {self.server_process.returncode}",
                        "ERROR",
                    )
                    if stderr:
                        print("STDERR:", stderr[-500:])
                    return False

            self.log("Server startup timed out after 45 seconds", "ERROR")
            # Try to get output for debugging
            if self.server_process.poll() is None:
                self.server_process.terminate()
                try:
                    stdout, stderr = self.server_process.communicate(timeout=5)
                    if stderr:
                        print("STDERR after timeout:", stderr[-500:])
                except Exception:
                    pass
            return False

        except Exception as e:
            self.log(f"Failed to start server: {e}", "ERROR")
            return False

    def stop_server(self):
        """Stop the web server if we started it."""
        if hasattr(self, "server_process") and self.server_process:
            self.log("Stopping web server...")
            self.server_process.terminate()
            try:
                self.server_process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                self.server_process.kill()
                self.server_process.wait()

    def run_health_checks(self, start_server_if_needed: bool = True) -> bool:
        """Run health check tests."""
        self.log("=== HEALTH CHECKS ===")

        # Start server if needed for health checks
        if start_server_if_needed and not self.check_server_running():
            if not self.start_server():
                self.log("⚠️  Could not start server for health checks", "ERROR")
                return False

        # Try Python version first
        python_success = self.run_command(
            [sys.executable, "tests/health_check/run_quick_health_check.py"],
            "Health Check (Python)",
            timeout=120,
        )

        # Try shell version as backup
        if not python_success:
            shell_script = (
                self.tests_dir / "health_check" / "test_endpoints_health.sh"
            )
            if shell_script.exists():
                self.run_command(
                    ["bash", str(shell_script)],
                    "Health Check (Shell)",
                    timeout=120,
                )

        return python_success

    def run_unit_tests(self) -> bool:
        """Run unit and feature tests."""
        self.log("=== UNIT & FEATURE TESTS ===")

        # Add extra debugging for CI environments
        pytest_args = [
            find_exe("pdm"),
            "run",
            "pytest",
            "-v",  # Verbose output
            "tests/test_settings_manager.py",
            "tests/test_google_pse.py",
            "tests/test_wikipedia_url_security.py",
            "tests/test_search_engines_enhanced.py",
            "tests/test_utils.py",
            "tests/rate_limiting/",  # Rate limiting test suite
            "tests/retriever_integration/",  # LangChain retriever integration tests
            "tests/feature_tests/",
            "tests/fix_tests/",
            "tests/infrastructure_tests/",  # Infrastructure and architecture tests
            "tests/content_fetcher/",  # Content fetcher tests (URL classifier, HTML downloader)
            # Critical stability tests - prevent breaking changes
            "tests/database/test_encryption_constants.py",  # Encryption params MUST NOT change
            "tests/database/test_schema_stability.py",  # Table names MUST NOT change
            "tests/performance/database/test_backwards_compatibility.py",  # DB compatibility tests
            "--cov=src",
            "--cov-report=term-missing",
        ]

        # In CI, add even more debugging
        if os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS"):
            pytest_args.extend(
                [
                    "--tb=short",  # Short traceback format
                    "--durations=10",  # Show 10 slowest tests
                ]
            )
            self.log("Running with enhanced CI debugging options")

            # Log environment info for debugging
            self.log(
                f"CI Environment detected: CI={os.environ.get('CI')}, GITHUB_ACTIONS={os.environ.get('GITHUB_ACTIONS')}"
            )
            self.log(f"Python path: {sys.executable}")
            self.log(f"Working directory: {os.getcwd()}")

        return self.run_command(
            pytest_args,
            "Unit & Feature Tests",
            timeout=300,
        )

    def run_integration_tests(self) -> bool:
        """Run integration tests (external dependencies)."""
        self.log("=== INTEGRATION TESTS ===")

        return self.run_command(
            [
                sys.executable,
                "-m",
                "pytest",
                "-v",
                "tests/searxng/",
                "-k",
                "not slow",
            ],
            "Integration Tests",
            timeout=120,
        )

    def run_ui_tests(self, start_server_if_needed: bool = True) -> bool:
        """Run UI tests with Puppeteer."""
        self.log("=== UI TESTS ===")

        if start_server_if_needed and not self.check_server_running():
            if not self.start_server():
                self.log(
                    "⚠️  Could not start server - skipping UI tests", "ERROR"
                )
                return False
        elif not self.check_server_running():
            self.log(
                "⚠️  Warning: Server not running at http://127.0.0.1:5000",
                "ERROR",
            )
            self.log("Please start the server with: pdm run ldr-web")
            return False

        # Check for Node.js and run UI tests
        try:
            subprocess.run(
                ["node", "--version"], capture_output=True, check=True
            )
        except (subprocess.CalledProcessError, FileNotFoundError):
            self.log("Node.js not found - skipping UI tests", "ERROR")
            return False

        # Check if we're in a headless environment (CI)
        cmd = ["node", "run_all_tests.js"]
        if os.environ.get("CI") or not os.environ.get("DISPLAY"):
            # If no display is available, use xvfb-run if available
            try:
                subprocess.run(
                    ["which", "xvfb-run"], capture_output=True, check=True
                )
                cmd = ["xvfb-run", "-a", "-s", "-screen 0 1920x1080x24"] + cmd
                self.log(
                    "Running UI tests with xvfb-run for headless environment"
                )
            except (subprocess.CalledProcessError, FileNotFoundError):
                self.log("Warning: No display available and xvfb-run not found")
                # Set headless environment variable for Puppeteer
                os.environ["PUPPETEER_HEADLESS"] = "true"

        return self.run_command(
            cmd,
            "UI Tests (Puppeteer)",
            cwd=self.tests_dir / "ui_tests",
            timeout=300,
        )

    def run_all_pytest_tests(self) -> bool:
        """Run all pytest tests with coverage (unit, integration, feature tests)."""
        self.log("=== ALL PYTEST TESTS ===")

        # Run all tests with coverage
        cmd = [
            sys.executable,
            "-m",
            "pytest",
            "--verbose",
            "--color=yes",
            "--cov=src",
            "--cov-report=term",
            "--cov-report=html:coverage_html",
            "--cov-config=.coveragerc",
        ]
        timeout = 600  # 10 minutes for all tests

        if os.environ.get("CI"):
            cmd.extend(
                [
                    "--tb=line",  # Show one line per failure for easier debugging
                    "--timeout=60",  # Per-test timeout of 60 seconds
                    # Removed -x and --maxfail to show all test results
                ]
            )
            self.log(
                "Running in CI mode with verbose output - showing all test results"
            )

        return self.run_command(cmd, "All Pytest Tests", timeout=timeout)

    def print_summary(self):
        """Print test execution summary."""
        self.log("=" * 60)
        self.log("TEST EXECUTION SUMMARY")
        self.log("=" * 60)

        total_tests = len(self.results)
        passed = sum(1 for _, success, _ in self.results if success)
        failed = total_tests - passed
        total_time = sum(duration for _, _, duration in self.results)

        for name, success, duration in self.results:
            status = "✅ PASS" if success else "❌ FAIL"
            self.log(f"{status} {name} ({duration:.1f}s)")

        self.log("-" * 60)
        self.log(
            f"📊 Total: {total_tests} | ✅ Passed: {passed} | ❌ Failed: {failed}"
        )
        self.log(f"⏱️  Total Time: {total_time:.1f}s")
        self.log(f"📈 Success Rate: {(passed / total_tests * 100):.1f}%")

        if failed == 0:
            self.log("🎉 All tests passed!", "SUCCESS")
        else:
            self.log(f"⚠️  {failed} test(s) failed", "ERROR")

        return failed == 0


def main():
    parser = argparse.ArgumentParser(
        description="Run Local Deep Research tests"
    )
    parser.add_argument(
        "profile",
        nargs="?",
        default="standard",
        choices=[
            "fast",
            "standard",
            "full",
            "comprehensive",
            "ci",
            "unit-only",
        ],
        help="Test profile to run",
    )
    parser.add_argument(
        "--no-server-start",
        action="store_true",
        help="Skip automatic server startup - assume server is already running",
    )

    args = parser.parse_args()
    runner = TestRunner()

    runner.log(f"🚀 Starting {args.profile.upper()} test profile")
    start_time = time.time()

    success = True

    try:
        if args.profile == "fast":
            # Quick tests for rapid feedback (< 30s)
            success &= runner.run_health_checks(
                start_server_if_needed=not args.no_server_start
            )
            success &= runner.run_unit_tests()

        elif args.profile == "standard":
            # Standard development testing (< 5min)
            success &= runner.run_health_checks(
                start_server_if_needed=not args.no_server_start
            )
            success &= runner.run_unit_tests()
            success &= runner.run_ui_tests(
                start_server_if_needed=not args.no_server_start
            )

        elif args.profile == "full":
            # Optimized full testing - no duplicate test runs (< 10min)
            success &= runner.run_health_checks(
                start_server_if_needed=not args.no_server_start
            )
            # Run all pytest tests (includes unit, integration, feature tests)
            success &= runner.run_all_pytest_tests()
            # UI tests require special setup so run separately
            success &= runner.run_ui_tests(
                start_server_if_needed=not args.no_server_start
            )

        elif args.profile == "comprehensive":
            # Legacy comprehensive testing - runs some tests multiple times (< 15min)
            # Kept for backwards compatibility
            success &= runner.run_health_checks(
                start_server_if_needed=not args.no_server_start
            )
            success &= runner.run_all_pytest_tests()
            success &= (
                runner.run_integration_tests()
            )  # Redundant but kept for compatibility
            success &= runner.run_ui_tests(
                start_server_if_needed=not args.no_server_start
            )

        elif args.profile == "ci":
            # CI/CD optimized tests
            # Skip health checks in CI as they require a running server
            # which is not available in the CI environment
            success &= runner.run_unit_tests()
            # Skip UI tests in CI unless explicitly configured

        elif args.profile == "unit-only":
            # Unit tests only - no server dependencies
            success &= runner.run_unit_tests()

    finally:
        # Always cleanup server if we started it
        runner.stop_server()

    total_duration = time.time() - start_time
    runner.log(f"🏁 Test suite completed in {total_duration:.1f}s")

    overall_success = runner.print_summary()

    sys.exit(0 if overall_success else 1)


if __name__ == "__main__":
    main()