wabot/tests/unit/test_hosts.py

import http.server
import json
import os
import stat
import sys
import threading
import time

import pytest

from wabot.hosts import (
    ExternalServer,
    ManagedService,
    build_options,
    find_driver_binary,
    normalize_browser,
    service_alive,
    stop_service,
)


class TestNormalizeBrowser:
    def test_chrome_is_an_alias_for_chromium(self):
        assert normalize_browser("chrome") == "chromium"

    def test_firefox_and_chromium_pass_through(self):
        assert normalize_browser("firefox") == "firefox"
        assert normalize_browser("chromium") == "chromium"

    def test_unknown_browser_raises(self):
        with pytest.raises(ValueError, match="unsupported browser"):
            normalize_browser("safari")


class TestBuildOptions:
    def test_chromium_headless_and_user_agent(self):
        opts = build_options("chromium", headless=True, user_agent="MyBot/1.0")
        assert "--headless=new" in opts.arguments
        assert "--user-agent=MyBot/1.0" in opts.arguments

    def test_chromium_defaults_have_no_headless_or_ua(self):
        opts = build_options("chromium")
        assert not any(a.startswith(("--headless", "--user-agent")) for a in opts.arguments)

    def test_firefox_headless_and_user_agent(self):
        opts = build_options("firefox", headless=True, user_agent="MyBot/1.0")
        assert "-headless" in opts.arguments
        assert opts.preferences.get("general.useragent.override") == "MyBot/1.0"

    def test_options_type_matches_browser(self):
        assert type(build_options("firefox")).__module__.startswith("selenium.webdriver.firefox")
        assert type(build_options("chrome")).__module__.startswith("selenium.webdriver.chrome")


class _StatusHandler(http.server.BaseHTTPRequestHandler):
    ready = True

    def do_GET(self):
        if self.path == "/status":
            body = json.dumps({"value": {"ready": self.ready, "message": ""}}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(body)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, *args):  # keep test output clean
        pass


@pytest.fixture
def status_server():
    server = http.server.HTTPServer(("127.0.0.1", 0), _StatusHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield f"http://127.0.0.1:{server.server_address[1]}"
    server.shutdown()
    server.server_close()


class TestServiceAlive:
    def test_true_when_status_answers(self, status_server):
        assert service_alive(status_server) is True

    def test_true_even_when_not_ready(self, status_server):
        # geckodriver reports ready=false while its single session is in use;
        # aliveness must not require ready==true
        _StatusHandler.ready = False
        try:
            assert service_alive(status_server) is True
        finally:
            _StatusHandler.ready = True

    def test_false_when_nothing_listens(self):
        assert service_alive("http://127.0.0.1:1") is False


class TestExternalServer:
    def test_returns_url_when_alive(self, status_server):
        assert ExternalServer(status_server + "/").ensure_running() == status_server

    def test_raises_when_dead(self):
        with pytest.raises(ConnectionError, match="no WebDriver server"):
            ExternalServer("http://127.0.0.1:1").ensure_running()


class TestServiceAliveNonHttpResponder:
    def test_non_http_service_on_port_is_false_and_clean_error(self):
        # wrong-port misconfiguration: some non-HTTP service answers the socket
        import socket

        server = socket.create_server(("127.0.0.1", 0))
        port = server.getsockname()[1]

        def answer(count=2):
            for _ in range(count):
                try:
                    conn, _ = server.accept()
                except OSError:
                    return
                conn.sendall(b"SSH-2.0-OpenSSH_9.7\r\n")
                conn.close()

        thread = threading.Thread(target=answer, daemon=True)
        thread.start()
        try:
            assert service_alive(f"http://127.0.0.1:{port}", timeout=2.0) is False
            with pytest.raises(ConnectionError, match="no WebDriver server"):
                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