base integration tests

pyproject.toml

@@ -139,6 +139,7 @@ module = [
     "scrubadub.*",
     "traceloop.*",
     "browser_use.*",
+    "pytest_check.*",
     "cdp_use.*",
     "aiohttp.*",
     "websockets.*",
@@ -155,6 +156,7 @@ disallow_subclassing_any = false
 module = ["tests.*"]
 disallow_untyped_decorators = false
 disallow_untyped_defs = false
+disallow_untyped_calls = false
 
 # ============================================================================
 # Ruff Configuration (Fast Python Linter & Formatter)
@@ -241,10 +243,13 @@ ignore = [
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = [
-    "S106",   # Possible hardcoded password
-    "S108",   # Possible insecure usage of temporary file/directory
-    "ARG001", # Unused function argument
+    "S106",    # Possible hardcoded password
+    "S108",    # Possible insecure usage of temporary file/directory
+    "S603",    # subprocess call with untrusted input
+    "S607",    # Partial executable path
+    "ARG001",  # Unused function argument
     "PLR2004", # Magic value used in comparison
+    "PLW0603", # Global statement (acceptable for test UI state)
 ]
 "strix/tools/**/*.py" = [
     "ARG001", # Unused function argument (tools may have unused args for interface consistency)
@@ -351,6 +356,10 @@ python_files = ["test_*.py", "*_test.py"]
 python_functions = ["test_*"]
 python_classes = ["Test*"]
 asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "session"
+markers = [
+    "integration: end-to-end tests requiring Docker and network access",
+]
 
 [tool.coverage.run]
 source = ["strix"]

tests/integration/conftest.py

@@ -0,0 +1,173 @@
+import asyncio
+import logging
+import os
+import subprocess
+import threading
+import uuid
+from pathlib import Path
+
+import pytest
+
+from strix.agents.state import AgentState
+from strix.runtime.docker_runtime import DockerRuntime
+from strix.tools.browser.browser_actions import browser_action
+from strix.tools.context import set_current_agent_id
+
+from . import console as ui
+
+
+def pytest_addoption(parser):
+    parser.addoption("--pretty", action="store_true", default=False, help="Clean TUI output")
+
+
+def _preflight():
+    missing = [v for v in ("STRIX_LLM", "LLM_API_KEY") if not os.environ.get(v)]
+    if missing:
+        pytest.exit(
+            f"Missing required env vars: {', '.join(missing)}\n"
+            "  STRIX_LLM=<model, eg: openrouter/xyz>\n"
+            "  LLM_API_KEY=<api key>",
+            returncode=1,
+        )
+
+    image = os.environ.get("STRIX_IMAGE", "")
+    if not image:
+        ui.log_warn("STRIX_IMAGE not set — falling back to config default")
+    elif "dev" not in image:
+        ui.log_warn(f"STRIX_IMAGE={image} — consider using strix-sandbox:dev for integration tests")
+
+
+_preflight()
+
+# in pretty mode, silence all loggers and suppress pytest's own terminal output
+if ui.is_pretty():
+    for _name in (
+        "strix.tests.integration",
+        "strix.tests.integration.browser",
+        "strix.tests.integration.container",
+        "strix.tools.browser.browser_actions",
+        "strix.tools.browser.browser_manager",
+    ):
+        _lg = logging.getLogger(_name)
+        _lg.handlers = []
+        _lg.propagate = False
+        _lg.setLevel(logging.CRITICAL)
+
+_SESSION_AGENT_ID = f"integration-{uuid.uuid4().hex[:8]}"
+
+_bg_loop = asyncio.new_event_loop()
+_bg_thread = threading.Thread(target=_bg_loop.run_forever, daemon=True)
+_bg_thread.start()
+
+
+def _run_in_bg(coro):
+    future = asyncio.run_coroutine_threadsafe(coro, _bg_loop)
+    return future.result(timeout=120)
+
+
+@pytest.hookimpl(trylast=True)
+def pytest_configure(config):
+    if not ui.is_pretty():
+        return
+    # redirect pytest's stdout writes to devnull so "PASSED"/"FAILED"/dots don't show
+    terminal = config.pluginmanager.get_plugin("terminalreporter")
+    if terminal is not None:
+        terminal._tw = type(terminal._tw)(Path(os.devnull).open("w"))  # noqa: SIM115
+
+
+def pytest_report_teststatus(report, config):
+    if not ui.is_pretty():
+        return None
+    if report.when == "call":
+        return "", "", ""
+    return None
+
+
+# -- fixtures ------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _ui_lifecycle():
+    ui.start()
+    yield
+    ui.stop()
+
+
+@pytest.fixture(scope="session")
+def docker_runtime():
+    ui.status("Creating DockerRuntime…")
+    runtime = DockerRuntime()
+    yield runtime
+    runtime.cleanup()
+
+
+@pytest.fixture(scope="session")
+def sandbox_info(docker_runtime):
+    agent_id = _SESSION_AGENT_ID
+    ui.status(f"Creating sandbox ({agent_id})…")
+    info = _run_in_bg(docker_runtime.create_sandbox(agent_id))
+    ui.log(f"Sandbox ready: container={info['workspace_id'][:12]} api={info['api_url']}")
+
+    yield info
+
+    subprocess.Popen(
+        ["docker", "rm", "-f", info["workspace_id"]],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    docker_runtime._scan_container = None
+    docker_runtime._tool_server_port = None
+    docker_runtime._tool_server_token = None
+    docker_runtime._caido_port = None
+
+
+@pytest.fixture(scope="session")
+def agent_state(sandbox_info):
+    return AgentState(
+        agent_id=_SESSION_AGENT_ID,
+        sandbox_id=sandbox_info["workspace_id"],
+        sandbox_token=sandbox_info["auth_token"],
+        sandbox_info=sandbox_info,
+    )
+
+
+@pytest.fixture(scope="session")
+def browser_session(agent_state):
+    set_current_agent_id(_SESSION_AGENT_ID)
+    ui.status("Launching browser…")
+    result = _run_in_bg(browser_action(action="launch", agent_state=agent_state))
+    if "error" in result:
+        pytest.fail(f"Browser launch failed: {result}")
+    ui.log(f"Browser ready: mode={result.get('mode')}")
+    yield agent_state
+    from strix.tools.browser.browser_manager import _manager
+
+    _manager.sessions.pop(_SESSION_AGENT_ID, None)
+    _bg_loop.call_soon_threadsafe(_bg_loop.stop)
+
+
+def pytest_runtest_logstart(nodeid, location):
+    name = nodeid.split("::")[-1]
+    ui.test_start(name)
+
+
+def pytest_runtest_logreport(report):
+    if report.when != "call":
+        return
+    name = report.nodeid.split("::")[-1]
+    if report.passed:
+        ui.test_passed(name)
+    elif report.failed:
+        ui.test_failed(name, details=report.longreprtext)
+
+
+@pytest.fixture(autouse=True)
+def _set_agent_context():
+    set_current_agent_id(_SESSION_AGENT_ID)
+
+
+@pytest.fixture
+def browser(browser_session):
+    from .helpers import Browser
+
+    return Browser(browser_session)

tests/integration/console.py

@@ -0,0 +1,176 @@
+import sys
+import threading
+
+
+_pretty = "--pretty" in sys.argv
+_lock = threading.Lock()
+
+_failures: list[dict[str, str | dict[str, object] | None]] = []
+_passed = 0
+_failed = 0
+
+_live = None
+_spinner = None
+_console = None
+
+if _pretty:
+    from rich.console import Console
+    from rich.live import Live
+    from rich.markup import escape as rich_escape
+    from rich.spinner import Spinner
+
+    _console = Console(stderr=True, force_terminal=True)
+
+
+def is_pretty():
+    return _pretty
+
+
+def start():
+    global _live, _spinner
+    if not _pretty:
+        return
+    _spinner = Spinner("dots", text="Starting…")
+    _live = Live(
+        _spinner,
+        console=_console,
+        refresh_per_second=12,
+    )
+    _live.start()
+
+
+def stop():
+    global _live
+    if _live is not None:
+        _live.stop()
+        _live = None
+    if _pretty and (_passed or _failed):
+        _print_summary()
+
+
+def status(msg):
+    if not _pretty:
+        return
+    with _lock:
+        if _spinner is not None:
+            _spinner.update(text=f"  {msg}")
+
+
+def _print_line(text, **kwargs):
+    with _lock:
+        if _live is not None:
+            _live.console.print(text, **kwargs)
+        elif _console is not None:
+            _console.print(text, **kwargs)
+
+
+def test_start(name):
+    if not _pretty:
+        return
+    _print_line(f"  [bold]●[/] [bold]{name}[/]")
+
+
+def log(msg, style="dim"):
+    if not _pretty:
+        return
+    _print_line(f"      [dim]│[/] {msg}", style=style, highlight=False)
+
+
+def log_error(msg):
+    if not _pretty:
+        return
+    _print_line(f"      [red]│[/] {msg}", style="bold red", highlight=False)
+
+
+def log_warn(msg):
+    if not _pretty:
+        return
+    _print_line(f"      [yellow]│[/] {msg}", style="yellow", highlight=False)
+
+
+def test_passed(name):
+    global _passed
+    _passed += 1
+    if not _pretty:
+        return
+    _print_line("      [bold green]╰─ ✓ pass[/]")
+
+
+def test_failed(name, details=""):
+    global _failed
+    _failed += 1
+    if details:
+        _failures.append({"name": name, "details": details})
+    if not _pretty:
+        return
+    _print_line("      [bold red]╰─ ✗ fail[/]")
+
+
+def record_failure(
+    test_name,
+    label,
+    reason,
+    result=None,
+    screenshot_path=None,
+):
+    _failures.append(
+        {
+            "name": test_name,
+            "label": label,
+            "reason": reason,
+            "result": result,
+            "screenshot": screenshot_path,
+        }
+    )
+
+
+def _print_summary():
+    if _console is None:
+        return
+    _console.print()
+    _console.rule(style="dim")
+    _console.print()
+
+    if _failures:
+        _console.print("  [bold red]failures[/]")
+        _console.print()
+        for f in _failures:
+            name = f.get("name", "?")
+            reason = str(f.get("reason", f.get("details", "")))
+            result = f.get("result")
+            screenshot = f.get("screenshot")
+
+            _console.print(f"  [bold red]── {name} ──[/]")
+
+            if reason:
+                for line in reason.split("\n"):
+                    _console.print(f"     [red]│[/] {rich_escape(line)}")
+
+            if result and isinstance(result, dict):
+                _console.print("     [red]│[/]")
+                _console.print("     [red]╰─▶[/] [dim]result:[/]")
+                for k, v in result.items():
+                    if k == "screenshot":
+                        continue
+                    v_str = str(v)
+                    if len(v_str) > 120:
+                        v_str = v_str[:120] + "…"
+                    _console.print(f"          [dim]{k}:[/] {rich_escape(v_str)}")
+
+            if screenshot:
+                _console.print(f"     [dim]screenshot:[/] [underline]{rich_escape(screenshot)}[/]")
+
+            _console.print()
+
+    total = _passed + _failed
+    if _failed:
+        _console.print(
+            f"  [bold]result:[/] [bold red]FAILED[/]  "
+            f"{_passed} passed, {_failed} failed  "
+            f"({total} total)",
+        )
+    else:
+        _console.print(
+            f"  [bold]result:[/] [bold green]ok[/]  {_passed} passed  ({total} total)",
+        )
+    _console.print()

tests/integration/helpers.py

@@ -0,0 +1,85 @@
+import base64
+import inspect
+import shutil
+from pathlib import Path
+
+from pytest_check import check
+
+from . import console as ui
+from .conftest import _run_in_bg
+
+
+SCREENSHOTS_DIR = Path(__file__).parent / "screenshots"
+
+
+def setup_screenshots_dir():
+    if SCREENSHOTS_DIR.exists():
+        shutil.rmtree(SCREENSHOTS_DIR)
+    SCREENSHOTS_DIR.mkdir(exist_ok=True)
+
+
+class Browser:
+    def __init__(self, agent_state):
+        self._state = agent_state
+
+    def __getattr__(self, action):
+        from strix.tools.browser.browser_actions import browser_action
+
+        def call(**kwargs):
+            result = _run_in_bg(
+                browser_action(
+                    action=action,
+                    agent_state=self._state,
+                    **kwargs,
+                )
+            )
+            if "error" in result:
+                Fail(result).error(result["error"])
+            return result
+
+        return call
+
+
+def _caller_test_name():
+    for frame in inspect.stack():
+        if frame.function.startswith("test_"):
+            return frame.function
+    return "unknown"
+
+
+def _save_screenshot(result, name):
+    b64 = result.get("screenshot")
+    if not b64 or not isinstance(b64, str) or len(b64) < 100:
+        return None
+    path = SCREENSHOTS_DIR / f"{name}.png"
+    path.write_bytes(base64.b64decode(b64))
+    return str(path)
+
+
+class Fail:
+    def __init__(self, result=None):
+        self._result = result
+        self._name = _caller_test_name()
+        self._screenshot = _save_screenshot(result, self._name) if result else None
+
+    def expected(self, value):
+        self._expected = value
+        return self
+
+    def got(self, value):
+        self._emit(f"expected {self._expected!r}, got {value!r}")
+
+    def error(self, msg):
+        self._emit(msg)
+
+    def _emit(self, reason):
+        ui.log_error(f"  \\[{self._name}] {reason}")
+        ui.record_failure(
+            self._name,
+            self._name,
+            reason,
+            self._result,
+            self._screenshot,
+        )
+        with check:
+            check.fail(f"[{self._name}] {reason}")

tests/integration/test_browser_actions.py

@@ -0,0 +1,144 @@
+import base64
+import binascii
+
+import pytest
+from pytest_check import check
+
+from . import console as ui
+from .helpers import Fail, setup_screenshots_dir
+
+
+pytestmark = pytest.mark.integration
+
+setup_screenshots_dir()
+
+
+def test_navigate(browser):
+    ui.status("test_navigate → example.com")
+    result = browser.navigate(url="https://example.com")
+    ui.log(f"navigate → url={result.get('url')} title={result.get('title')}")
+
+    with check:
+        check.is_in("example.com", result.get("url", "").lower())
+        check.is_in("example", result.get("title", "").lower())
+
+
+def test_click(browser):
+    ui.status("test_click → example.com link")
+    browser.navigate(url="https://example.com")
+
+    result = browser.click(index=1)
+    ui.log(f"click → url={result.get('url')}")
+
+
+def test_go_back(browser):
+    ui.status("test_go_back → navigate to iana.org, then back")
+    browser.navigate(url="https://example.com")
+    browser.navigate(url="https://www.iana.org")
+
+    result = browser.go_back()
+    url = result.get("url", "")
+    ui.log(f"go_back → url={url}")
+
+    if "example.com" not in url.lower():
+        Fail(result).expected("url containing 'example.com'").got(url)
+
+
+def test_scroll(browser):
+    ui.status("test_scroll → example.com")
+    browser.navigate(url="https://example.com")
+
+    result = browser.scroll(direction="down", amount=3)
+    result_keys = list(result.keys()) if isinstance(result, dict) else type(result)
+    ui.log(f"scroll → {result_keys}")
+
+
+def test_extract(browser):
+    ui.status("test_extract → example.com")
+    browser.navigate(url="https://example.com")
+
+    result = browser.extract(query="What is on this page?")
+    ui.log(f"extract → {str(result)[:120]}")
+
+    if "example" not in str(result).lower():
+        Fail(result).expected("content containing 'example'").got(str(result)[:200])
+
+
+def test_find_text(browser):
+    ui.status("test_find_text → 'example'")
+    browser.navigate(url="https://example.com")
+
+    result = browser.find_text(text="example")
+    result_keys = list(result.keys()) if isinstance(result, dict) else type(result)
+    ui.log(f"find_text → {result_keys}")
+
+
+def test_evaluate(browser):
+    ui.status("test_evaluate → document.title")
+    browser.navigate(url="https://example.com")
+
+    result = browser.evaluate(code="document.title")
+    ui.log(f"evaluate → {str(result)[:120]}")
+
+    if "example" not in str(result).lower():
+        Fail(result).expected("result containing 'example'").got(str(result)[:200])
+
+
+def test_screenshot(browser):
+    ui.status("test_screenshot → example.com")
+    browser.navigate(url="https://example.com")
+
+    result = browser.screenshot()
+    screenshot = result.get("screenshot", "")
+    ui.log(f"screenshot → {len(screenshot)} bytes base64")
+
+    if len(screenshot) <= 100:
+        Fail(result).expected("> 100 bytes").got(f"{len(screenshot)} bytes")
+    try:
+        base64.b64decode(screenshot)
+    except (ValueError, binascii.Error) as e:
+        Fail(result).error(f"invalid base64: {e}")
+
+
+def test_input(browser):
+    ui.status("test_input → example.com")
+    browser.navigate(url="https://example.com")
+
+    result = browser.input(index=1, text="integration test")
+    result_keys = list(result.keys()) if isinstance(result, dict) else type(result)
+    ui.log(f"input → {result_keys}")
+
+
+def test_send_keys(browser):
+    ui.status("test_send_keys → Tab")
+    browser.navigate(url="https://example.com")
+
+    result = browser.send_keys(keys="Tab")
+    result_keys = list(result.keys()) if isinstance(result, dict) else type(result)
+    ui.log(f"send_keys → {result_keys}")
+
+
+def _tab_id(tab_info):
+    return getattr(tab_info, "target_id", "")[-4:]
+
+
+def test_switch_and_close_tab(browser):
+    ui.status("test_switch_and_close_tab → open, switch, close")
+    result = browser.navigate(url="https://example.com")
+    tabs = result.get("tabs", [])
+    original = _tab_id(tabs[0]) if tabs else None
+
+    ui.status("test_switch_and_close_tab → opening new tab")
+    result = browser.evaluate(code="window.open('https://example.com', '_blank')")
+    tabs = result.get("tabs", [])
+    ui.log(f"obtained tabs {tabs}")
+    new = _tab_id(tabs[-1]) if tabs else None
+
+    if new:
+        ui.status("test_switch_and_close_tab → switching to new tab")
+        result = browser.switch(tab_id=new)
+        ui.log(f"switch → url={result.get('url')}")
+
+    if original:
+        ui.status("test_switch_and_close_tab → switching back")
+        browser.switch(tab_id=original)