"""Tests for marketing_intelligence.browser.base.""" import asyncio import pytest from marketing_intelligence.browser.base import ( PageSnapshot, create_adapter, create_adapter_from_session, ) from marketing_intelligence.browser.drissionpage import DrissionPageAdapter from marketing_intelligence.browser.nodriver import NoDriverAdapter from marketing_intelligence.browser.playwright import PlaywrightAdapter from marketing_intelligence.evasion.behavior import BehaviorEmulator from marketing_intelligence.evasion.profiles import get_profile # ─────────────────────────── PageSnapshot ────────────────────────────────── class TestPageSnapshot: def test_required_fields(self) -> None: s = PageSnapshot(url="https://example.com", html="") assert s.url == "https://example.com" assert s.html == "" def test_optional_fields_default_none(self) -> None: s = PageSnapshot(url="https://example.com", html="") assert s.title is None assert s.final_url is None def test_response_time_defaults_zero(self) -> None: s = PageSnapshot(url="https://example.com", html="") assert s.response_time_ms == 0 def test_all_fields_set(self) -> None: s = PageSnapshot( url="https://example.com", html="", title="Title", final_url="https://example.com/final", response_time_ms=1234, ) assert s.title == "Title" assert s.final_url == "https://example.com/final" assert s.response_time_ms == 1234 # ─────────────────────────── create_adapter ──────────────────────────────── class TestCreateAdapter: def test_camoufox_returns_playwright_adapter(self) -> None: assert isinstance(create_adapter("camoufox"), PlaywrightAdapter) def test_chromium_returns_playwright_adapter(self) -> None: assert isinstance(create_adapter("chromium"), PlaywrightAdapter) def test_webkit_returns_playwright_adapter(self) -> None: assert isinstance(create_adapter("webkit"), PlaywrightAdapter) def test_nodriver_returns_nodriver_adapter(self) -> None: assert isinstance(create_adapter("nodriver"), NoDriverAdapter) def test_drissionpage_returns_drissionpage_adapter(self) -> None: assert isinstance(create_adapter("drissionpage"), DrissionPageAdapter) def test_invalid_engine_raises_value_error(self) -> None: with pytest.raises(ValueError, match="Unknown engine"): create_adapter("firefox") def test_proxy_passed_to_adapter(self) -> None: proxy = {"server": "http://proxy:8080"} adapter = create_adapter("camoufox", proxy=proxy) assert adapter.proxy == proxy def test_profile_name_applied(self) -> None: adapter = create_adapter("camoufox", profile_name="mobile_ios") assert adapter.profile.name == "mobile_ios" def test_human_behavior_false(self) -> None: adapter = create_adapter("camoufox", human_behavior=False) assert adapter.behavior.human is False def test_delay_factor_applied(self) -> None: adapter = create_adapter("camoufox", delay_factor=0.5) assert adapter.behavior.delay_factor == 0.5 def test_camoufox_engine_stored_on_playwright(self) -> None: adapter = create_adapter("camoufox") assert isinstance(adapter, PlaywrightAdapter) assert adapter.engine == "camoufox" def test_chromium_engine_stored_on_playwright(self) -> None: adapter = create_adapter("chromium") assert isinstance(adapter, PlaywrightAdapter) assert adapter.engine == "chromium" # ─────────────────────────── create_adapter_from_session ─────────────────── class TestCreateAdapterFromSession: class _FakeSession: engine = "nodriver" profile_name = "mobile_ios" human_behavior = False delay_factor = 2.0 scroll_speed = "slow" def test_uses_session_engine(self) -> None: adapter = create_adapter_from_session(self._FakeSession()) assert isinstance(adapter, NoDriverAdapter) def test_uses_session_profile(self) -> None: adapter = create_adapter_from_session(self._FakeSession()) assert adapter.profile.name == "mobile_ios" def test_uses_session_human_behavior(self) -> None: adapter = create_adapter_from_session(self._FakeSession()) assert adapter.behavior.human is False def test_uses_session_delay_factor(self) -> None: adapter = create_adapter_from_session(self._FakeSession()) assert adapter.behavior.delay_factor == 2.0 def test_proxy_override(self) -> None: proxy = {"server": "http://proxy:8080"} adapter = create_adapter_from_session(self._FakeSession(), proxy=proxy) assert adapter.proxy == proxy def test_missing_attrs_use_defaults(self) -> None: adapter = create_adapter_from_session(object()) assert isinstance(adapter, PlaywrightAdapter) # ─────────────────────────── PlaywrightAdapter ───────────────────────────── class TestPlaywrightAdapterInit: def _adapter(self, engine: str) -> PlaywrightAdapter: return PlaywrightAdapter( engine=engine, profile=get_profile("desktop"), behavior=BehaviorEmulator(), proxy=None, ) def test_valid_engines_accepted(self) -> None: for engine in ("camoufox", "chromium", "webkit"): assert self._adapter(engine).engine == engine def test_invalid_engine_raises(self) -> None: with pytest.raises(ValueError, match="engine must be one of"): self._adapter("firefox") # ─────────────────────────── fetch_interactive not supported ─────────────── class TestFetchInteractiveNotSupported: async def _dummy_cb(self, page: object, html: str, url: str, ms: int) -> None: pass def test_nodriver_raises_not_implemented(self) -> None: adapter = NoDriverAdapter( profile=get_profile("desktop"), behavior=BehaviorEmulator(), proxy=None, ) with pytest.raises(NotImplementedError): asyncio.run( adapter.fetch_interactive("https://example.com", self._dummy_cb) ) def test_drissionpage_raises_not_implemented(self) -> None: adapter = DrissionPageAdapter( profile=get_profile("desktop"), behavior=BehaviorEmulator(), proxy=None, ) with pytest.raises(NotImplementedError): asyncio.run( adapter.fetch_interactive("https://example.com", self._dummy_cb) ) # ─────────────────────────── PersistentBrowserPool ───────────────────────── class TestPersistentBrowserPool: def test_close_nonexistent_session_is_safe(self) -> None: from marketing_intelligence.browser.pool import PersistentBrowserPool pool = PersistentBrowserPool() asyncio.run(pool.close("nonexistent")) def test_close_all_on_empty_pool_is_safe(self) -> None: from marketing_intelligence.browser.pool import PersistentBrowserPool pool = PersistentBrowserPool() asyncio.run(pool.close_all()) def test_get_page_returns_cached_page(self) -> None: from marketing_intelligence.browser.pool import PersistentBrowserPool pool = PersistentBrowserPool() mock_page = object() pool._entries["test-session"] = (None, mock_page) result = asyncio.run(pool.get_page("test-session", None)) assert result is mock_page