wabot/tests/unit/test_api.py

from unittest.mock import MagicMock

import pytest

import wabot
from wabot.sessions import SessionRecord, SessionStore


@pytest.fixture
def store(tmp_path):
    return SessionStore(path=tmp_path / "sessions.json")


@pytest.fixture
def fake_selenium(monkeypatch):
    """Patch every constructor that would talk to a real browser."""
    fakes = MagicMock(name="fakes")
    fakes.Remote.return_value.session_id = "new-session-id"
    monkeypatch.setattr(wabot, "_new_remote", fakes.Remote)
    monkeypatch.setattr(wabot, "_new_local", fakes.Local)
    monkeypatch.setattr(wabot, "attach", fakes.attach)
    monkeypatch.setattr(wabot, "service_alive", fakes.service_alive)
    fakes.managed = MagicMock(name="ManagedService_instance")
    fakes.managed.ensure_running.return_value = "http://127.0.0.1:7777"
    fakes.managed.pid, fakes.managed.port = 4321, 7777
    monkeypatch.setattr(wabot, "ManagedService", MagicMock(return_value=fakes.managed))
    return fakes


class TestEphemeral:
    def test_no_session_no_host_uses_local_driver(self, fake_selenium, store):
        bot = wabot.browser(store=store)
        fake_selenium.Local.assert_called_once()
        assert bot.driver is fake_selenium.Local.return_value
        assert store.names() == []  # nothing persisted

    def test_no_session_with_host_uses_remote_without_saving(self, fake_selenium, store):
        wabot.browser(host="http://grid:4444", store=store)
        fake_selenium.Remote.assert_called_once()
        assert fake_selenium.Remote.call_args.args[0] == "http://grid:4444"
        assert store.names() == []


class TestPersistentCreate:
    def test_managed_service_created_and_recorded(self, fake_selenium, store):
        bot = wabot.browser(session="s1", browser="chromium", store=store)
        record = store.get("s1")
        assert record == SessionRecord(
            name="s1",
            executor_url="http://127.0.0.1:7777",
            session_id="new-session-id",
            browser="chromium",
            created_at=record.created_at,
            service_pid=4321,
            service_port=7777,
        )
        assert bot.session_name == "s1"

    def test_external_host_records_no_pid(self, fake_selenium, store, monkeypatch):
        external = MagicMock()
        external.ensure_running.return_value = "http://grid:4444"
        del external.pid  # ExternalServer has no pid/port attributes
        del external.port
        monkeypatch.setattr(wabot, "ExternalServer", MagicMock(return_value=external))
        wabot.browser(session="s1", host="http://grid:4444", store=store)
        record = store.get("s1")
        assert record.executor_url == "http://grid:4444"
        assert record.service_pid is None and record.service_port is None


class TestReattach:
    def make_record(self, store):
        from datetime import datetime, timezone

        store.save(
            SessionRecord(
                name="s1",
                executor_url="http://127.0.0.1:7777",
                session_id="old-session",
                browser="chromium",
                created_at=datetime.now(timezone.utc).isoformat(),
            )
        )

    def test_live_record_is_reattached(self, fake_selenium, store):
        self.make_record(store)
        fake_selenium.service_alive.return_value = True
        bot = wabot.browser(session="s1", store=store)
        fake_selenium.attach.assert_called_once_with(
            "http://127.0.0.1:7777", "old-session", "chromium"
        )
        assert bot.driver is fake_selenium.attach.return_value
        fake_selenium.Remote.assert_not_called()

    def test_dead_record_falls_through_to_fresh_creation(self, fake_selenium, store):
        self.make_record(store)
        fake_selenium.service_alive.return_value = True
        fake_selenium.attach.return_value = None  # session gone
        wabot.browser(session="s1", store=store)
        fake_selenium.Remote.assert_called_once()
        assert store.get("s1").session_id == "new-session-id"

    def test_dead_server_skips_attach_entirely(self, fake_selenium, store):
        self.make_record(store)
        fake_selenium.service_alive.return_value = False
        wabot.browser(session="s1", store=store)
        fake_selenium.attach.assert_not_called()
        fake_selenium.Remote.assert_called_once()


class TestHousekeeping:
    def test_sessions_lists_names(self, store):
        assert wabot.sessions(store=store) == []

    def test_destroy_quits_kills_and_removes(self, fake_selenium, store, monkeypatch):
        from datetime import datetime, timezone

        stop = MagicMock()
        monkeypatch.setattr(wabot, "stop_service", stop)
        store.save(
            SessionRecord(
                name="s1",
                executor_url="http://127.0.0.1:7777",
                session_id="old-session",
                browser="chromium",
                created_at=datetime.now(timezone.utc).isoformat(),
                service_pid=4321,
            )
        )
        fake_selenium.service_alive.return_value = True
        assert wabot.destroy("s1", store=store) is True
        fake_selenium.attach.return_value.quit.assert_called_once_with()
        stop.assert_called_once_with(4321)
        assert store.get("s1") is None

    def test_destroy_missing_returns_false(self, store):
        assert wabot.destroy("nope", store=store) is False


class TestResourceCleanup:
    def test_managed_service_stopped_when_driver_creation_fails(
        self, fake_selenium, store, monkeypatch
    ):
        from selenium.common.exceptions import WebDriverException

        stopped = []
        monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
        fake_selenium.Remote.side_effect = WebDriverException("startup failed")
        with pytest.raises(WebDriverException):
            wabot.browser(session="s1", browser="chromium", store=store)
        assert stopped == [4321]  # spawned managed service was stopped
        assert store.get("s1") is None  # nothing persisted

    def test_dead_managed_session_stops_old_service_before_recreating(
        self, fake_selenium, store, monkeypatch
    ):
        from datetime import datetime, timezone

        stopped = []
        monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
        store.save(
            wabot.SessionRecord(
                name="s1",
                executor_url="http://127.0.0.1:7777",
                session_id="old",
                browser="chromium",
                created_at=datetime.now(timezone.utc).isoformat(),
                service_pid=9999,
                service_port=7777,
            )
        )
        fake_selenium.service_alive.return_value = True
        fake_selenium.attach.return_value = None  # session dead on a live server
        wabot.browser(session="s1", store=store)
        assert 9999 in stopped  # old managed service stopped, not leaked
        fake_selenium.Remote.assert_called_once()  # fresh browser created
        assert store.get("s1").session_id == "new-session-id"


class TestDestroyEdges:
    def _save(self, store, **kw):
        from datetime import datetime, timezone

        defaults = dict(
            name="s1",
            executor_url="http://127.0.0.1:7777",
            session_id="old",
            browser="chromium",
            created_at=datetime.now(timezone.utc).isoformat(),
        )
        defaults.update(kw)
        store.save(wabot.SessionRecord(**defaults))

    def test_destroy_external_session_does_not_stop_service(
        self, fake_selenium, store, monkeypatch
    ):
        stopped = []
        monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
        self._save(store, name="ext", executor_url="http://grid:4444", service_pid=None)
        fake_selenium.service_alive.return_value = True
        assert wabot.destroy("ext", store=store) is True
        assert stopped == []  # external server: not ours to stop
        assert store.get("ext") is None

    def test_destroy_dead_session_still_stops_and_removes(self, fake_selenium, store, monkeypatch):
        stopped = []
        monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
        self._save(store, service_pid=4321)
        fake_selenium.service_alive.return_value = True
        fake_selenium.attach.return_value = None  # dead session
        assert wabot.destroy("s1", store=store) is True
        assert stopped == [4321]  # service stopped despite dead session
        assert store.get("s1") is None


class TestPlumbing:
    def test_headless_and_user_agent_reach_build_options(self, fake_selenium, store, monkeypatch):
        captured = {}
        real_build = wabot.build_options

        def spy(browser, **kwargs):
            captured.update(kwargs)
            return real_build(browser, **kwargs)

        monkeypatch.setattr(wabot, "build_options", spy)
        wabot.browser(browser="chromium", headless=True, user_agent="Bot/1.0", store=store)
        assert captured == {"headless": True, "user_agent": "Bot/1.0"}

    def test_pacing_reaches_browser(self, fake_selenium, store):
        pacing = wabot.NoPacing()
        bot = wabot.browser(browser="chromium", pacing=pacing, store=store)
        assert bot.pacing is pacing


class TestPublicSurface:
    def test_all_exports_exist(self):
        for name in wabot.__all__:
            assert hasattr(wabot, name), name