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 TestPublicSurface:
    def test_all_exports_exist(self):
        for name in wabot.__all__:
            assert hasattr(wabot, name), name