rentgen/scripts/test_verify.py

#!/usr/bin/env python3
"""
test_verify.py - Verifies extension and exits (doesn't wait forever)

This script is a Python version of test_verify.sh that:
- Starts Firefox with the extension
- Verifies extension loaded without errors
- EXITS after verification (instead of waiting forever)

Used for automated tests in CI/Docker.
Uses guard clauses pattern for cleaner flow control.
"""

import sys
import time
import subprocess
import os
import signal
import json
from pathlib import Path


def print_header(text: str) -> None:
    """Print formatted header."""
    print(f"\n{text}")


def print_success(text: str) -> None:
    """Print success message."""
    print(f"✓ {text}")


def print_error(text: str) -> None:
    """Print error message."""
    print(f"✗ {text}")


def start_xvfb() -> int:
    """Start Xvfb virtual X server. Returns PID."""
    print_header("Starting Xvfb on display :99...")

    xvfb = subprocess.Popen(
        ["Xvfb", ":99", "-screen", "0", "1024x768x24"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )

    os.environ["DISPLAY"] = ":99"
    time.sleep(2)

    print_success(f"Xvfb started with PID: {xvfb.pid}")
    return xvfb.pid


def start_webext() -> tuple[int, Path]:
    """Start web-ext and log output to file. Returns (PID, log_path)."""
    print_header("Starting web-ext run with verbose logging...")
    print("=" * 40)

    log_path = Path("/tmp/web-ext.log")
    log_file = open(log_path, "w")

    # Enable Marionette on port 2828
    # Use --arg to pass arguments to Firefox
    webext = subprocess.Popen(
        ["npx", "web-ext", "run", "--verbose",
         "--arg=-marionette",
         "--arg=--marionette-port",
         "--arg=2828"],
        stdout=log_file,
        stderr=subprocess.STDOUT
    )

    return webext.pid, log_path


def wait_for_extension_install(log_path: Path, timeout_seconds: int = 60) -> bool:
    """Wait for extension installation confirmation in logs."""
    print_header("Waiting for extension to install...")

    for i in range(timeout_seconds):
        if not log_path.exists():
            time.sleep(1)
            continue

        content = log_path.read_text()

        # Check for Firefox crashes or errors
        if "firefox crashed" in content.lower() or "panic" in content.lower():
            print_error("Firefox crashed or panicked!")
            print("Last 20 lines of log:")
            print("\n".join(content.split("\n")[-20:]))
            return False

        # Check if Marionette is mentioned
        if i == 10 and "marionette" in content.lower():
            print_success("Marionette mentioned in logs")

        # Look for any of these success indicators
        success_indicators = [
            "Installed /app as a temporary add-on",
            "as a temporary add-on",
            "Extension added",
            "Installed temporary add-on",
            "installed from",
            "Installing manifest at"
        ]

        if any(indicator in content for indicator in success_indicators):
            print("=" * 40)
            print_success("Extension installed!")
            print_success("Firefox is running in headless mode")
            print_success("Extension: rentgen@internet-czas-dzialac.pl")

            # Check for Marionette port in logs
            if "2828" in content:
                print_success("Marionette port 2828 found in logs")
            return True

        # Print progress every 10 seconds
        if i > 0 and i % 10 == 0:
            print(f"Still waiting... ({i}s)")

        time.sleep(1)

    # If we timeout, print the last part of the log
    print_error("Timeout! Last 30 lines of log:")
    if log_path.exists():
        content = log_path.read_text()
        print("\n".join(content.split("\n")[-30:]))

    return False


def check_javascript_errors(log_path: Path) -> list[str]:
    """Check for JavaScript errors in extension code. Returns list of errors."""
    print_header("Checking for JavaScript errors in extension code...")

    content = log_path.read_text()

    # Filter out unrelated Firefox errors
    error_patterns = [
        "JavaScript error.*background.js",
        "SyntaxError.*background",
        "ReferenceError.*background"
    ]

    errors = []
    for line in content.split("\n"):
        # Skip BackupService and RSLoader errors (Firefox internal)
        if "BackupService" in line or "RSLoader" in line:
            continue

        for pattern in error_patterns:
            import re
            if re.search(pattern, line, re.IGNORECASE):
                errors.append(line)
                break

    return errors


def check_content_script_marker_in_logs(log_path: Path) -> tuple[bool, str]:
    """Check if background received message from content script in web-ext logs.
    This proves: background script → tabs.create() → content script injection → runtime.sendMessage() → background received
    Returns (success, message)."""
    try:
        content = log_path.read_text()

        # Look for background test marker (content script sends message to background, which logs it)
        import re
        pattern = r'\[RENTGEN_BACKGROUND_TEST\] Received verification from content script: (.+)'
        match = re.search(pattern, content)

        if match:
            try:
                import json
                data = json.loads(match.group(1))
                title = data.get('title', '')
                timestamp = data.get('timestamp', 0)
                url = data.get('url', '')

                # Verify page title contains "Example" (proves page loaded correctly)
                if 'Example' in title or 'example' in title.lower():
                    return True, f"Content script verified! Title: '{title}', URL: {url}, timestamp: {timestamp}"
                else:
                    return False, f"Page title unexpected: '{title}' (expected 'Example')"
            except json.JSONDecodeError:
                return False, f"Found marker but couldn't parse JSON: {match.group(1)}"

        return False, "No background test marker found in logs (content script may not have executed)"

    except Exception as e:
        return False, f"Log check failed: {e}"


def extract_debugger_port(log_path: Path) -> str | None:
    """Extract debugger port from web-ext logs."""
    content = log_path.read_text()
    import re
    match = re.search(r"start-debugger-server (\d+)", content)
    if match:
        return match.group(1)
    return None


def test_with_marionette() -> tuple[bool, str]:
    """Test using Firefox Marionette (WebDriver) protocol."""
    try:
        from marionette_driver.marionette import Marionette
        from marionette_driver import errors
        import time

        print_header("Testing with Marionette WebDriver...")

        # Try to connect to Marionette on default port
        client = Marionette(host='localhost', port=2828)

        try:
            client.start_session()
            print_success("Connected to Marionette")

            # Stage 1: Test simple JavaScript execution
            print_header("Stage 1: Testing JavaScript execution (2 + 3)...")
            result = client.execute_script("return 2 + 3;")

            if result == 5:
                print_success(f"JavaScript execution works! 2 + 3 = {result}")
            else:
                return False, f"Unexpected result from 2+3: {result}"

            # Stage 2: Navigate to example.com
            print_header("Stage 2: Navigating to example.com...")
            client.navigate("https://example.com")
            time.sleep(3)  # Wait for page to load

            title = client.title
            print_success(f"Page loaded, title: {title}")

            if "Example" not in title:
                return False, f"Unexpected page title: {title}"

            # Stage 3: Test bidirectional communication with extension
            print_header("Stage 3: Testing bidirectional extension communication...")

            # Generate a test value for verification
            test_value = 17  # Arbitrary test value
            expected_result = (test_value * 2) + 3  # Should be 37

            # Dispatch event to content script and wait for background script response
            verification_result = client.execute_script("""
                const testValue = arguments[0];
                const expectedResult = arguments[1];

                // First check if content script is injected
                const injected = document.body.getAttribute('data-rentgen-injected');
                if (injected !== 'true') {
                    return { success: false, error: 'Content script not injected' };
                }

                // Dispatch test request to content script
                document.dispatchEvent(new CustomEvent('rentgen_test_request', {
                    detail: {
                        value: testValue,
                        timestamp: Date.now()
                    }
                }));

                // Wait for the full round-trip communication
                return new Promise((resolve) => {
                    let attempts = 0;
                    const checkInterval = setInterval(() => {
                        attempts++;

                        // Check if content script processed the message
                        const verified = document.body.getAttribute('data-rentgen-verified');
                        const computed = document.body.getAttribute('data-rentgen-computed');
                        const formula = document.body.getAttribute('data-rentgen-formula');
                        const bgTimestamp = document.body.getAttribute('data-rentgen-background-timestamp');
                        const error = document.body.getAttribute('data-rentgen-error');

                        if (verified === 'true' && computed) {
                            clearInterval(checkInterval);
                            resolve({
                                success: true,
                                verified: true,
                                computed: parseInt(computed),
                                formula: formula,
                                backgroundTimestamp: bgTimestamp,
                                expectedResult: expectedResult,
                                testValue: testValue
                            });
                        } else if (verified === 'false' || attempts > 20) {
                            clearInterval(checkInterval);
                            resolve({
                                success: false,
                                error: error || 'Timeout waiting for background response',
                                attempts: attempts,
                                verified: verified
                            });
                        }
                    }, 100);  // Check every 100ms, up to 2 seconds
                });
            """, script_args=[test_value, expected_result], script_timeout=3000)

            # Analyze the results
            if verification_result.get('success'):
                computed = verification_result.get('computed')
                expected = verification_result.get('expectedResult')
                formula = verification_result.get('formula', '')

                print_success("Content script successfully communicated with background!")
                print_success(f"Background computed: {formula}")

                if computed == expected:
                    print_success(f"Computation verified! {test_value} → {computed} (expected {expected})")
                    return True, f"Full extension stack verified! Background computed ({test_value}*2)+3={computed}"
                else:
                    print_error(f"Computation mismatch: got {computed}, expected {expected}")
                    return False, f"Background returned wrong value: {computed} != {expected}"
            else:
                error = verification_result.get('error', 'Unknown error')
                print_error(f"Extension communication failed: {error}")
                return False, f"Content script/background communication failed: {error}"

        finally:
            try:
                client.close()
            except:
                pass

    except ImportError:
        return False, "marionette_driver not installed"
    except ConnectionRefusedError:
        return False, "Cannot connect to Marionette on port 2828 (Firefox may not have Marionette enabled)"
    except Exception as e:
        return False, f"Marionette test failed: {e}"


def cleanup(xvfb_pid: int, webext_pid: int) -> None:
    """Cleanup processes."""
    try:
        os.kill(webext_pid, signal.SIGTERM)
    except ProcessLookupError:
        pass

    try:
        os.kill(xvfb_pid, signal.SIGTERM)
    except ProcessLookupError:
        pass


def main() -> int:
    """Main test verification logic using guard clauses."""

    # Start Xvfb
    xvfb_pid = start_xvfb()

    # Start web-ext
    webext_pid, log_path = start_webext()

    # Guard: Check if extension installed
    if not wait_for_extension_install(log_path):
        print_error("Extension failed to install within 60s")
        cleanup(xvfb_pid, webext_pid)
        return 1

    # Give extension time to initialize
    time.sleep(3)

    # Guard: Check for JavaScript errors
    errors = check_javascript_errors(log_path)
    if errors:
        print()
        print("=" * 40)
        print("✗✗✗ CRITICAL ERROR ✗✗✗")
        print("=" * 40)
        print("Found JavaScript errors in background.js!")
        print("Extension installed but CODE DID NOT EXECUTE!")
        print()
        print("Errors:")
        for error in errors[:10]:
            print(f"  {error}")
        print("=" * 40)

        cleanup(xvfb_pid, webext_pid)
        return 1

    print_success("NO JavaScript errors in background.js")

    # Functional test: Verify extension code execution via Marionette
    print_header("Functional test: Verifying extension code execution...")

    # Give extension time to initialize
    time.sleep(5)

    # Test with Marionette WebDriver
    marionette_success, marionette_msg = test_with_marionette()

    # Guard: Check if Marionette tests passed
    if not marionette_success:
        print_error("Marionette test failed")
        print_error(f"Reason: {marionette_msg}")
        print_error("Extension installed but could not verify execution via Marionette")
        cleanup(xvfb_pid, webext_pid)
        return 1

    print_success("Extension code VERIFIED executing!")
    print()
    print(f"Test results: {marionette_msg}")
    print()
    print("This proves UNDENIABLY:")
    print("  - Firefox Marionette WebDriver works")
    print("  - JavaScript execution works (2+3=5)")
    print("  - Page navigation works (example.com loaded)")
    print("  - Content script is injected and listening for events")
    print("  - Content script sends messages to background script")
    print("  - Background script receives messages and computes results")
    print("  - Background script sends responses back to content script")
    print("  - Full bidirectional communication verified with computed proof")
    print("  - The extension is not just installed but ACTIVELY EXECUTING")

    # Show process info
    print()
    print_success("Process info:")
    subprocess.run(["ps", "aux"], stdout=subprocess.PIPE, text=True, check=False)
    result = subprocess.run(
        ["bash", "-c", "ps aux | grep -E '(firefox|Xvfb)' | grep -v grep | head -3"],
        capture_output=True,
        text=True
    )
    print(result.stdout)

    print("=" * 40)
    print("Extension is VERIFIED working!")
    print("=" * 40)

    # Cleanup
    cleanup(xvfb_pid, webext_pid)

    print("Test completed successfully.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted by user")
        sys.exit(130)
    except Exception as e:
        print(f"Unexpected error: {e}", file=sys.stderr)
        sys.exit(1)