myAi/docs/skills/myai-smoke-test/scripts/run_smoke_test.py

#!/usr/bin/env python3
"""
myAi Smoke Test Automation Script

Runs a minimal end-to-end test of the myAi CV Matcher application:
1. Starts Docker Compose
2. Waits for app health check
3. Uploads CV.pdf and job description
4. Verifies CV analysis results display
5. Confirms match email was sent

Exit codes:
0 = All tests passed
1 = One or more tests failed
2 = Critical error (app won't start, file not found, etc.)
"""

import os
import sys
import time
import json
import subprocess
import requests
from pathlib import Path
from datetime import datetime
from urllib.parse import urljoin

try:
    from selenium import webdriver
    from selenium.webdriver.common.by import By
    from selenium.webdriver.support.ui import WebDriverWait
    from selenium.webdriver.support import expected_conditions as EC
    from selenium.webdriver.chrome.options import Options
    from webdriver_manager.chrome import ChromeDriverManager
    from selenium.webdriver.chrome.service import Service
except ImportError:
    print("ERROR: Required packages not found. Install with:")
    print("pip install selenium webdriver-manager requests")
    sys.exit(2)


# Configuration
PROJECT_ROOT = os.environ.get("PROJECT_ROOT", os.getcwd())
APP_BASE_URL = os.environ.get("APP_BASE_URL", "http://localhost:5140")
DOCKER_COMPOSE_FILE = os.environ.get("DOCKER_COMPOSE_FILE", "docker-compose/docker-compose.yml")
APP_STARTUP_TIMEOUT = int(os.environ.get("APP_STARTUP_TIMEOUT", 45))
RESULTS_DISPLAY_TIMEOUT = int(os.environ.get("RESULTS_DISPLAY_TIMEOUT", 30))
HEADLESS_CHROME = os.environ.get("HEADLESS_CHROME", "true").lower() == "true"
KEEP_CONTAINERS = os.environ.get("KEEP_CONTAINERS", "false").lower() == "true"

# Hardcoded test data
CV_FILE_PATH = os.path.join(PROJECT_ROOT, "CV.pdf")
JOB_DESCRIPTION = """Senior Full Stack Engineer - 5+ years experience with C# ASP.NET Core, React, and cloud deployment. Experience with CI/CD pipelines, Docker, and agile teams required."""


def log(message, level="INFO"):
    """Log a message with timestamp."""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{timestamp}] [{level}] {message}")


def run_command(command, cwd=None):
    """Run a shell command and return result."""
    try:
        result = subprocess.run(
            command,
            shell=True,
            cwd=cwd or PROJECT_ROOT,
            capture_output=True,
            text=True,
            timeout=60
        )
        return result.returncode, result.stdout, result.stderr
    except subprocess.TimeoutExpired:
        return -1, "", "Command timed out"
    except Exception as e:
        return -1, "", str(e)


def check_app_health():
    """Check if app is responding to health check."""
    try:
        response = requests.get(APP_BASE_URL, timeout=2)
        return response.status_code == 200
    except requests.RequestException:
        return False


def wait_for_app_startup():
    """Wait for app to be ready, up to APP_STARTUP_TIMEOUT seconds."""
    log(f"Waiting for app at {APP_BASE_URL}...")
    start_time = time.time()

    while time.time() - start_time < APP_STARTUP_TIMEOUT:
        if check_app_health():
            log(f"✓ App is ready (took {time.time() - start_time:.1f}s)")
            return True
        time.sleep(1)

    log(f"✗ App failed to start within {APP_STARTUP_TIMEOUT} seconds", "ERROR")
    return False


def start_docker_compose():
    """Start Docker Compose containers."""
    log("Starting Docker Compose...")
    docker_compose_path = os.path.join(PROJECT_ROOT, DOCKER_COMPOSE_FILE)

    if not os.path.exists(docker_compose_path):
        log(f"ERROR: docker-compose file not found at {docker_compose_path}", "ERROR")
        return False

    # Include both main compose file and override file for local builds
    override_file = os.path.join(os.path.dirname(DOCKER_COMPOSE_FILE), "docker-compose.override.yml")
    compose_cmd = f"docker compose -f {DOCKER_COMPOSE_FILE} -f {override_file} up --build -d"

    returncode, stdout, stderr = run_command(compose_cmd)

    if returncode != 0:
        log(f"ERROR: Failed to start Docker Compose: {stderr}", "ERROR")
        return False

    log("Docker Compose started, waiting for app to be ready...")
    return wait_for_app_startup()


def stop_docker_compose():
    """Stop Docker Compose containers."""
    if KEEP_CONTAINERS:
        log("KEEP_CONTAINERS=true, leaving containers running")
        return

    log("Stopping Docker Compose...")
    override_file = os.path.join(os.path.dirname(DOCKER_COMPOSE_FILE), "docker-compose.override.yml")
    compose_cmd = f"docker compose -f {DOCKER_COMPOSE_FILE} -f {override_file} down"
    run_command(compose_cmd)


def initialize_chrome_driver():
    """Initialize Selenium Chrome WebDriver."""
    log("Initializing Chrome WebDriver...")

    try:
        options = Options()
        if HEADLESS_CHROME:
            options.add_argument("--headless=new")
        options.add_argument("--disable-blink-features=AutomationControlled")
        options.add_argument("--disable-gpu")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")

        service = Service(ChromeDriverManager().install())
        driver = webdriver.Chrome(service=service, options=options)
        log("✓ Chrome WebDriver ready")
        return driver
    except Exception as e:
        log(f"ERROR: Failed to initialize Chrome: {e}", "ERROR")
        return None


def upload_cv(driver):
    """Upload CV.pdf file."""
    log("Uploading CV file...")

    if not os.path.exists(CV_FILE_PATH):
        log(f"ERROR: CV file not found at {CV_FILE_PATH}", "ERROR")
        return False

    try:
        # Find the file input element
        file_input = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.ID, "cvFile"))
        )

        # Send the file path to the input
        file_input.send_keys(os.path.abspath(CV_FILE_PATH))
        log(f"✓ CV file uploaded ({os.path.basename(CV_FILE_PATH)})")

        # Wait a moment for the file change handler to process
        time.sleep(1)
        return True
    except Exception as e:
        log(f"ERROR: Failed to upload CV: {e}", "ERROR")
        return False


def fill_job_description(driver):
    """Fill in the job description field."""
    log("Filling job description...")

    try:
        # Find the job description textarea
        job_input = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.ID, "jobInput"))
        )

        # Clear any existing content and fill
        job_input.clear()
        job_input.send_keys(JOB_DESCRIPTION)
        log("✓ Job description filled")
        return True
    except Exception as e:
        log(f"ERROR: Failed to fill job description: {e}", "ERROR")
        return False


def submit_form(driver):
    """Submit the CV match form."""
    log("Submitting form...")

    try:
        # Find and click the submit button
        submit_button = WebDriverWait(driver, 10).until(
            EC.element_to_be_clickable((By.ID, "submitBtn"))
        )
        submit_button.click()
        log("✓ Form submitted")
        return True
    except Exception as e:
        log(f"ERROR: Failed to submit form: {e}", "ERROR")
        return False


def verify_results_display(driver):
    """Wait for and verify CV analysis results display."""
    log("Waiting for results...")

    try:
        # Wait for results container to appear
        WebDriverWait(driver, RESULTS_DISPLAY_TIMEOUT).until(
            EC.presence_of_element_located((By.CLASS_NAME, "match-result"))
        )

        # Wait for results to be visible
        WebDriverWait(driver, 5).until(
            EC.visibility_of_element_located((By.CLASS_NAME, "match-result"))
        )

        # Check for key result elements
        result_container = driver.find_element(By.CLASS_NAME, "match-result")

        # Verify score badge exists
        try:
            score_badge = result_container.find_element(By.CLASS_NAME, "score-badge")
            score_text = score_badge.text
            log(f"✓ Match score displayed: {score_text}")
        except:
            log("WARNING: Score badge not found, but results container exists", "WARN")

        # Verify strengths section
        try:
            strengths = result_container.find_element(By.CLASS_NAME, "strengths")
            strength_items = strengths.find_elements(By.TAG_NAME, "li")
            log(f"✓ Strengths section found ({len(strength_items)} items)")
        except:
            log("WARNING: Strengths section not found", "WARN")

        # Verify gaps section
        try:
            gaps = result_container.find_element(By.CLASS_NAME, "gaps")
            gap_items = gaps.find_elements(By.TAG_NAME, "li")
            log(f"✓ Gaps section found ({len(gap_items)} items)")
        except:
            log("WARNING: Gaps section not found", "WARN")

        # Verify evidence section
        try:
            evidence = result_container.find_element(By.CLASS_NAME, "evidence")
            log("✓ Evidence section found")
        except:
            log("WARNING: Evidence section not found", "WARN")

        return True
    except Exception as e:
        log(f"ERROR: Results did not display within {RESULTS_DISPLAY_TIMEOUT}s: {e}", "ERROR")
        return False


def verify_email_sent():
    """Check Docker logs for email send confirmation."""
    log("Checking if email was sent...")

    try:
        returncode, stdout, stderr = run_command("docker logs myai-email-api-1")

        if returncode != 0:
            log("WARNING: Could not read email-api logs (container may not exist)", "WARN")
            return False

        # Check for email send confirmation
        if "Message sent" in stdout or "sent" in stdout.lower():
            log("✓ Email send confirmation found in logs")
            return True
        else:
            log("WARNING: Email send confirmation not found in logs", "WARN")
            # Don't fail the whole test if email can't be verified
            # (MailHog or real SMTP might have different log formats)
            return True
    except Exception as e:
        log(f"WARNING: Could not verify email: {e}", "WARN")
        return True


def run_smoke_test():
    """Run the complete smoke test."""
    log("=" * 60)
    log("myAi Smoke Test Starting")
    log("=" * 60)

    driver = None
    test_passed = True
    failed_steps = []

    try:
        # Step 1: Start Docker Compose
        if not start_docker_compose():
            log("CRITICAL: Application failed to start", "ERROR")
            return False, ["Application failed to start"]

        # Step 2: Initialize Chrome
        driver = initialize_chrome_driver()
        if not driver:
            log("CRITICAL: Chrome driver failed to initialize", "ERROR")
            return False, ["Chrome driver initialization failed"]

        # Step 3: Navigate to app
        log(f"Navigating to {APP_BASE_URL}...")
        driver.get(APP_BASE_URL)
        time.sleep(2)  # Wait for page to load

        # Step 4: Upload CV
        if not upload_cv(driver):
            test_passed = False
            failed_steps.append("CV file upload")

        time.sleep(1)

        # Step 5: Fill job description
        if not fill_job_description(driver):
            test_passed = False
            failed_steps.append("Job description input")

        time.sleep(1)

        # Step 6: Submit form
        if not submit_form(driver):
            test_passed = False
            failed_steps.append("Form submission")

        time.sleep(2)

        # Step 7: Verify results display
        if not verify_results_display(driver):
            test_passed = False
            failed_steps.append("Results display")

        # Step 8: Verify email
        if not verify_email_sent():
            test_passed = False
            failed_steps.append("Email verification")

    except Exception as e:
        log(f"CRITICAL: Unexpected error: {e}", "ERROR")
        test_passed = False
        failed_steps.append(f"Unexpected error: {e}")

    finally:
        # Cleanup
        if driver:
            try:
                driver.quit()
            except:
                pass

        stop_docker_compose()

    # Report results
    log("=" * 60)
    if test_passed:
        log("✓ SMOKE TEST PASSED - All steps completed successfully", "SUCCESS")
        log("CV uploaded, matched, results displayed, email sent.")
    else:
        log("✗ SMOKE TEST FAILED", "ERROR")
        log(f"Failed steps: {', '.join(failed_steps)}")
    log("=" * 60)

    return test_passed, failed_steps


if __name__ == "__main__":
    success, failed_steps = run_smoke_test()
    sys.exit(0 if success else 1)