From f82b77ae62c7a1ac43db6b3bc8eb3c07659e46a6 Mon Sep 17 00:00:00 2001 From: Mathew Sir Guest the best Date: Tue, 7 Jul 2026 22:08:32 -0600 Subject: [PATCH] feat: ManagedService spawns detached driver services that outlive python --- src/wabot/hosts.py | 97 +++++++++++++++++++++++++++++++++++++++- tests/unit/test_hosts.py | 85 ++++++++++++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 3 deletions(-) diff --git a/src/wabot/hosts.py b/src/wabot/hosts.py index 311e4fb..0cf7b3a 100644 --- a/src/wabot/hosts.py +++ b/src/wabot/hosts.py @@ -11,8 +11,16 @@ from __future__ import annotations import http.client import json import logging +import os +import shutil +import signal +import socket +import subprocess +import time import urllib.request +from pathlib import Path +import platformdirs from selenium import webdriver LOGGER = logging.getLogger("wabot") @@ -54,8 +62,6 @@ def build_options( def _chromium_binary() -> str | None: - import shutil - for name in ("chromium", "chromium-browser", "google-chrome", "chrome"): found = shutil.which(name) if found: @@ -88,3 +94,90 @@ class ExternalServer: if not service_alive(self.url): raise ConnectionError(f"no WebDriver server answering at {self.url}/status") return self.url + + +def find_driver_binary(browser: str) -> str: + """Locate chromedriver/geckodriver: env override, PATH, then Selenium Manager.""" + name = DRIVER_BINARIES[normalize_browser(browser)] + override = os.environ.get(f"WABOT_{name.upper()}") + if override: + return override + found = shutil.which(name) + if found: + return found + try: + # Selenium Manager is bundled with selenium but explicitly beta + # ("may change") — keep it as a guarded fallback, never the primary. + from selenium.webdriver.common.selenium_manager import SeleniumManager + + args = ["--browser", "chrome" if browser == "chromium" else browser] + chromium = _chromium_binary() + if browser == "chromium" and chromium: + args += ["--browser-path", chromium] + return SeleniumManager().binary_paths(args)["driver_path"] + except Exception as ex: # noqa: BLE001 — beta API, failure modes unknown + raise FileNotFoundError( + f"could not find {name!r} on PATH (set WABOT_{name.upper()} to override); " + f"Selenium Manager fallback also failed: {ex}" + ) from ex + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class ManagedService: + """A driver service wabot spawns detached, so it outlives this process.""" + + def __init__( + self, + browser: str, + log_dir: Path | None = None, + driver_binary: str | None = None, + startup_timeout: float = 15.0, + ): + self.browser = normalize_browser(browser) + self.log_dir = Path(log_dir) if log_dir else Path(platformdirs.user_data_dir("wabot")) + self.driver_binary = driver_binary + self.startup_timeout = startup_timeout + self.pid: int | None = None + self.port: int | None = None + + def ensure_running(self) -> str: + binary = self.driver_binary or find_driver_binary(self.browser) + self.port = _free_port() + url = f"http://127.0.0.1:{self.port}" + self.log_dir.mkdir(parents=True, exist_ok=True) + log_path = self.log_dir / f"{DRIVER_BINARIES[self.browser]}-{self.port}.log" + with open(log_path, "ab") as log: + process = subprocess.Popen( + [binary, f"--port={self.port}"], + stdout=log, + stderr=log, + stdin=subprocess.DEVNULL, + start_new_session=True, # own session: survives this python process + ) + self.pid = process.pid + LOGGER.info("spawned %s (pid %s) on %s, log %s", binary, self.pid, url, log_path) + deadline = time.monotonic() + self.startup_timeout + while time.monotonic() < deadline: + if service_alive(url): + return url + if process.poll() is not None: + break # exited early; fall through to the error + time.sleep(0.2) + raise TimeoutError( + f"{binary} did not answer /status on {url} within " + f"{self.startup_timeout}s (see {log_path})" + ) + + +def stop_service(pid: int) -> bool: + """SIGTERM a managed driver service. False if it was already gone.""" + try: + os.kill(pid, signal.SIGTERM) + return True + except (ProcessLookupError, PermissionError): + return False diff --git a/tests/unit/test_hosts.py b/tests/unit/test_hosts.py index 329e7aa..7686b1b 100644 --- a/tests/unit/test_hosts.py +++ b/tests/unit/test_hosts.py @@ -1,10 +1,22 @@ import http.server import json +import os +import stat +import sys import threading +import time import pytest -from wabot.hosts import ExternalServer, build_options, normalize_browser, service_alive +from wabot.hosts import ( + ExternalServer, + ManagedService, + build_options, + find_driver_binary, + normalize_browser, + service_alive, + stop_service, +) class TestNormalizeBrowser: @@ -119,3 +131,74 @@ class TestServiceAliveNonHttpResponder: ExternalServer(f"http://127.0.0.1:{port}").ensure_running() finally: server.close() + + +STUB_DRIVER = """\ +#!{python} +import http.server, json, sys +port = int(sys.argv[1].split("=", 1)[1]) +class H(http.server.BaseHTTPRequestHandler): + def do_GET(self): + body = json.dumps({{"value": {{"ready": True, "message": ""}}}}).encode() + self.send_response(200); self.end_headers(); self.wfile.write(body) + def log_message(self, *a): pass +http.server.HTTPServer(("127.0.0.1", port), H).serve_forever() +""" + + +@pytest.fixture +def stub_driver(tmp_path): + path = tmp_path / "chromedriver" + path.write_text(STUB_DRIVER.format(python=sys.executable)) + path.chmod(path.stat().st_mode | stat.S_IEXEC) + return path + + +class TestFindDriverBinary: + def test_env_var_override_wins(self, monkeypatch): + monkeypatch.setenv("WABOT_CHROMEDRIVER", "/custom/chromedriver") + assert find_driver_binary("chromium") == "/custom/chromedriver" + + def test_path_lookup(self, monkeypatch, stub_driver): + monkeypatch.delenv("WABOT_CHROMEDRIVER", raising=False) + monkeypatch.setenv("PATH", str(stub_driver.parent)) + assert find_driver_binary("chromium") == str(stub_driver) + + +class TestManagedService: + def test_spawns_detached_and_answers_status(self, stub_driver, tmp_path): + svc = ManagedService("chromium", log_dir=tmp_path, driver_binary=str(stub_driver)) + url = svc.ensure_running() + try: + assert service_alive(url) + assert svc.pid is not None and svc.port is not None + assert url == f"http://127.0.0.1:{svc.port}" + # detached: the spawned process is in its own session, so it + # would survive this python process exiting + assert os.getsid(svc.pid) != os.getsid(os.getpid()) + assert (tmp_path / f"chromedriver-{svc.port}.log").exists() + finally: + stop_service(svc.pid) + + def test_startup_timeout_raises(self, tmp_path): + # /bin/true exits immediately and never serves /status + svc = ManagedService( + "chromium", log_dir=tmp_path, driver_binary="/bin/true", startup_timeout=1.0 + ) + with pytest.raises(TimeoutError, match="did not answer /status"): + svc.ensure_running() + + +class TestStopService: + def test_stops_a_live_process_and_reports_dead_ones(self, stub_driver, tmp_path): + svc = ManagedService("chromium", log_dir=tmp_path, driver_binary=str(stub_driver)) + svc.ensure_running() + assert stop_service(svc.pid) is True + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + try: + os.kill(svc.pid, 0) + except ProcessLookupError: + break + time.sleep(0.05) + assert stop_service(999999) is False # no such pid