feat(engine): complete stage 1 completion and stage 2 self-healing engine runs

Author: pras-ops

scrapewizard/cli/commands/engine.py

@@ -0,0 +1,93 @@
+import asyncio
+import json
+import typer
+from pathlib import Path
+from scrapewizard.engine.recorder import InteractiveRecorder
+from scrapewizard.engine.sandbox import SandboxRunner
+
+app = typer.Typer(help="Self-healing test automation engine commands")
+
+def async_command(f):
+    """Decorator to run async Typer commands synchronously."""
+    import functools
+    @functools.wraps(f)
+    def wrapper(*args, **kwargs):
+        return asyncio.run(f(*args, **kwargs))
+    return wrapper
+
+@app.command(name="record")
+@async_command
+async def record(
+    url: str = typer.Option(..., "--url", "-u", help="Target URL to start recording from"),
+    output: str = typer.Option("flow.json", "--output", "-o", help="Path to save the generated flow.json"),
+    screenshots: str = typer.Option("screenshots", "--screenshots", "-s", help="Directory to save crop screenshots")
+):
+    """
+    Open a headed browser to record user interactions on a page.
+    Saves the flow steps and element fingerprints to a flow.json file.
+    """
+    typer.echo(f"Starting interactive recording on {url}...")
+    recorder = InteractiveRecorder(output_path=output, screenshots_dir=screenshots, headless=False)
+    await recorder.start(url)
+    typer.echo(f"Successfully saved flow recording to {output}")
+
+@app.command(name="test")
+@async_command
+async def test(
+    flow_path: str = typer.Argument(..., help="Path to the flow.json file to execute"),
+    artifacts: str = typer.Option(None, "--artifacts", "-a", help="Directory to save run artifacts"),
+    headless: bool = typer.Option(True, "--headless/--headed", help="Run the browser in headless or headed mode")
+):
+    """
+    Run an automated headless sandbox execution of the recorded flow.json.
+    Validates console/network errors, visual diffs, and accessibility violations.
+    """
+    flow_file = Path(flow_path)
+    if not flow_file.exists():
+        typer.echo(f"Error: flow file not found at {flow_path}", err=True)
+        raise typer.Exit(code=1)
+        
+    try:
+        with open(flow_file, "r", encoding="utf-8") as f:
+            flow_data = json.load(f)
+    except Exception as e:
+        typer.echo(f"Error: failed to parse JSON in {flow_path}: {e}", err=True)
+        raise typer.Exit(code=1)
+        
+    # Construct test definition from flow data directly
+    from scrapewizard.engine.test_generator import TestGenerator
+    generator = TestGenerator(flow_path)
+    test_def = generator.generate()
+    
+    typer.echo(f"Running sandbox execution for {flow_path}...")
+    runner = SandboxRunner(artifacts_dir=artifacts, headless=headless)
+    result = await runner.run(test_def)
+    
+    # Print results
+    typer.echo(f"\nExecution finished in {result.duration_ms} ms. Status: {result.status.upper()}")
+    typer.echo(f"Artifacts saved to: {result.artifacts_dir}\n")
+    
+    for idx, step in enumerate(result.step_results):
+        status_symbol = "✅" if step.status == "passed" else "❌"
+        typer.echo(f" {status_symbol} Step {idx + 1}: {step.step_name} - {step.status.upper()} ({step.duration_ms} ms)")
+        if step.error_message:
+            typer.echo(f"    Error: {step.error_message}")
+        if step.console_errors:
+            typer.echo(f"    Console Errors ({len(step.console_errors)}):")
+            for err in step.console_errors:
+                typer.echo(f"      - {err}")
+        if step.network_errors:
+            typer.echo(f"    Network Failures ({len(step.network_errors)}):")
+            for err in step.network_errors:
+                typer.echo(f"      - {err}")
+        if step.a11y_violations:
+            typer.echo(f"    A11y Violations ({len(step.a11y_violations)}):")
+            for violation in step.a11y_violations:
+                typer.echo(f"      - [{violation['impact']}] {violation['id']}: {violation['help']}")
+                
+    if result.status != "passed":
+        typer.echo("\n❌ Run FAILED.")
+        raise typer.Exit(code=1)
+    else:
+        typer.echo("\n✅ Run PASSED.")
+        raise typer.Exit(code=0)

scrapewizard/cli/main.py

@@ -1,5 +1,5 @@
 import typer
-from scrapewizard.cli.commands import setup, scrape, utils
+from scrapewizard.cli.commands import setup, scrape, utils, engine
 from scrapewizard.core.logging import Logger
 
 app = typer.Typer(
@@ -16,6 +16,8 @@
 app.command()(utils.clean)
 app.command()(utils.doctor)
 app.command()(utils.resume)
+app.command(name="record")(engine.record)
+app.command(name="test")(engine.test)
 
 VERSION = "1.2.0"
 

scrapewizard/engine/checks/__init__.py

@@ -0,0 +1,4 @@
+# ScrapeWizard Step Checks Package
+from .console_network import ConsoleNetworkTracker
+from .visual import perform_visual_check
+from .a11y import perform_a11y_check

scrapewizard/engine/checks/a11y.py

@@ -0,0 +1,60 @@
+import os
+from pathlib import Path
+from typing import List, Dict, Any
+from scrapewizard.core.logging import log
+
+# Read the bundled script content once and cache it in memory
+AXE_SCRIPT_PATH = Path(__file__).parent / "axe.min.js"
+AXE_SCRIPT_CONTENT = None
+
+if AXE_SCRIPT_PATH.exists():
+    with open(AXE_SCRIPT_PATH, "r", encoding="utf-8") as f:
+        AXE_SCRIPT_CONTENT = f.read()
+
+async def perform_a11y_check(page) -> List[Dict[str, Any]]:
+    """
+    Inject axe-core into the target Playwright page, run analysis,
+    and return a structured list of accessibility violations.
+    """
+    global AXE_SCRIPT_CONTENT
+    
+    if not AXE_SCRIPT_CONTENT:
+        if AXE_SCRIPT_PATH.exists():
+            with open(AXE_SCRIPT_PATH, "r", encoding="utf-8") as f:
+                AXE_SCRIPT_CONTENT = f.read()
+        else:
+            log("axe.min.js not found, skipping accessibility check", level="warning")
+            return []
+
+    try:
+        # Inject the axe-core library
+        await page.evaluate(AXE_SCRIPT_CONTENT)
+        
+        # Run accessibility analysis
+        # axe.run() returns a promise, so we evaluate it asynchronously
+        results = await page.evaluate("async () => { return await axe.run(); }")
+        
+        violations = []
+        for violation in results.get("violations", []):
+            nodes = []
+            for node in violation.get("nodes", []):
+                nodes.append({
+                    "html": node.get("html"),
+                    "target": node.get("target"),
+                    "failure_summary": node.get("failureSummary")
+                })
+                
+            violations.append({
+                "id": violation.get("id"),
+                "impact": violation.get("impact"),
+                "description": violation.get("description"),
+                "help": violation.get("help"),
+                "help_url": violation.get("helpUrl"),
+                "nodes": nodes
+            })
+            
+        return violations
+        
+    except Exception as e:
+        log(f"Accessibility validation failed: {e}", level="warning")
+        return []

scrapewizard/engine/checks/axe.min.js

@@ -0,0 +1,33 @@
+class ConsoleNetworkTracker:
+    """
+    Captures console errors and network response failures (4xx/5xx status codes)
+    on a Playwright page during a test execution block.
+    """
+    def __init__(self, page):
+        self.page = page
+        self.console_errors = []
+        self.network_errors = []
+        self._setup_listeners()
+
+    def _setup_listeners(self):
+        def handle_console(msg):
+            if msg.type == "error":
+                location = msg.location or {}
+                url = location.get("url") or "unknown"
+                line = location.get("lineNumber") or 0
+                self.console_errors.append(f"[{url}:{line}] {msg.text}")
+        
+        def handle_response(response):
+            if response.status >= 400:
+                self.network_errors.append(f"[{response.status}] {response.url}")
+
+        self.page.on("console", handle_console)
+        self.page.on("response", handle_response)
+
+    def flush(self):
+        """Return the collected errors and clear the tracking buffers."""
+        console = list(self.console_errors)
+        network = list(self.network_errors)
+        self.console_errors.clear()
+        self.network_errors.clear()
+        return console, network

scrapewizard/engine/checks/console_network.py

@@ -0,0 +1,70 @@
+import os
+from pathlib import Path
+from typing import Tuple, Optional
+from PIL import Image, ImageChops
+
+def perform_visual_check(
+    current_screenshot_path: str,
+    step_name: str,
+    baseline_dir: str,
+    diff_dir: str,
+    threshold: float = 0.05
+) -> Tuple[float, Optional[str]]:
+    """
+    Perform a visual check comparing the current step's screenshot to a saved baseline.
+    If the baseline does not exist, copies current screenshot as baseline and returns (0.0, None).
+    Otherwise, computes the percentage of differing pixels.
+    If the difference is above threshold, saves a difference image in diff_dir.
+    Returns: (diff_score: float, diff_image_path: Optional[str])
+    """
+    baseline_path = Path(baseline_dir) / f"{step_name}.png"
+    
+    if not Path(current_screenshot_path).exists():
+        return 0.0, None
+
+    if not baseline_path.exists():
+        # Set the current run screenshot as the baseline
+        baseline_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            img = Image.open(current_screenshot_path)
+            img.save(baseline_path)
+        except Exception:
+            pass
+        return 0.0, None
+        
+    # Open and compare
+    try:
+        img1 = Image.open(baseline_path).convert("RGB")
+        img2 = Image.open(current_screenshot_path).convert("RGB")
+    except Exception:
+        # If image parsing fails, return 0 diff score
+        return 0.0, None
+    
+    if img1.size != img2.size:
+        img2 = img2.resize(img1.size)
+        
+    diff = ImageChops.difference(img1, img2)
+    stat = diff.getbbox()
+    if stat is None:
+        return 0.0, None
+        
+    # Calculate score (ratio of non-matching pixels)
+    gray_diff = diff.convert("L")
+    hist = gray_diff.histogram()
+    total_pixels = gray_diff.size[0] * gray_diff.size[1]
+    matching_pixels = hist[0]
+    differing_pixels = total_pixels - matching_pixels
+    diff_score = differing_pixels / total_pixels
+    
+    diff_image_path = None
+    if diff_score > threshold:
+        # Save diff image
+        diff_path = Path(diff_dir) / f"diff_{step_name}.png"
+        diff_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            diff.save(diff_path)
+            diff_image_path = str(diff_path)
+        except Exception:
+            pass
+        
+    return diff_score, diff_image_path