"""브라우저 라우터 단위 테스트 모든 테스트는 mock 기반으로 동작하며 실제 네트워크 연결이 불필요합니다. """ from __future__ import annotations from dataclasses import fields from unittest.mock import AsyncMock, MagicMock, patch import pytest from tools.browser_router import ( CHROME_ENDPOINT, LIGHTPANDA_ENDPOINT, PURPOSE_MAP, BrowserChoice, BrowserEngine, check_lightpanda_health, create_browser_context, get_browser, ) # --------------------------------------------------------------------------- # 1. 모듈 임포트 및 상수 확인 # --------------------------------------------------------------------------- class TestModuleImportAndConstants: def test_import_succeeds(self) -> None: """browser_router 모듈 import 가 성공하는지 확인""" assert PURPOSE_MAP is not None def test_lightpanda_endpoint_value(self) -> None: """LIGHTPANDA_ENDPOINT 값이 'ws://127.0.0.1:9333' 인지 확인""" assert LIGHTPANDA_ENDPOINT == "ws://127.0.0.1:9333" def test_chrome_endpoint_value(self) -> None: """CHROME_ENDPOINT 값이 'ws://127.0.0.1:9222' 인지 확인""" assert CHROME_ENDPOINT == "ws://127.0.0.1:9222" def test_purpose_map_required_keys(self) -> None: """PURPOSE_MAP 에 필수 키가 모두 포함되는지 확인""" required_keys = {"crawl", "bulk", "text", "login", "screenshot", "pdf", "spa", "stealth"} assert required_keys.issubset(PURPOSE_MAP.keys()), f"누락 키: {required_keys - set(PURPOSE_MAP.keys())}" def test_purpose_map_lightpanda_keys(self) -> None: """PURPOSE_MAP 의 Lightpanda 매핑 키가 올바른지 확인""" assert PURPOSE_MAP["crawl"] == BrowserEngine.LIGHTPANDA assert PURPOSE_MAP["bulk"] == BrowserEngine.LIGHTPANDA assert PURPOSE_MAP["text"] == BrowserEngine.LIGHTPANDA def test_purpose_map_chrome_keys(self) -> None: """PURPOSE_MAP 의 Chrome 매핑 키가 올바른지 확인""" assert PURPOSE_MAP["login"] == BrowserEngine.CHROME assert PURPOSE_MAP["screenshot"] == BrowserEngine.CHROME assert PURPOSE_MAP["pdf"] == BrowserEngine.CHROME assert PURPOSE_MAP["spa"] == BrowserEngine.CHROME assert PURPOSE_MAP["stealth"] == BrowserEngine.CHROME # --------------------------------------------------------------------------- # 2. get_browser() — Lightpanda 목적 + healthy # --------------------------------------------------------------------------- class TestGetBrowserLightpandaHealthy: @pytest.mark.asyncio async def test_crawl_returns_lightpanda(self) -> None: """get_browser('crawl') 이 healthy 상태에서 LIGHTPANDA 를 반환하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)): choice = await get_browser("crawl") assert choice.engine == BrowserEngine.LIGHTPANDA @pytest.mark.asyncio async def test_bulk_returns_lightpanda(self) -> None: """get_browser('bulk') 이 healthy 상태에서 LIGHTPANDA 를 반환하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)): choice = await get_browser("bulk") assert choice.engine == BrowserEngine.LIGHTPANDA @pytest.mark.asyncio async def test_text_returns_lightpanda(self) -> None: """get_browser('text') 이 healthy 상태에서 LIGHTPANDA 를 반환하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)): choice = await get_browser("text") assert choice.engine == BrowserEngine.LIGHTPANDA @pytest.mark.asyncio async def test_lightpanda_choice_endpoint(self) -> None: """Lightpanda 선택 시 endpoint 가 LIGHTPANDA_ENDPOINT 와 일치하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)): choice = await get_browser("crawl") assert choice.endpoint == LIGHTPANDA_ENDPOINT @pytest.mark.asyncio async def test_lightpanda_choice_has_reason(self) -> None: """Lightpanda 선택 시 reason 필드가 비어있지 않은지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)): choice = await get_browser("crawl") assert choice.reason # --------------------------------------------------------------------------- # 3. get_browser() — Chrome 목적 # --------------------------------------------------------------------------- class TestGetBrowserChrome: @pytest.mark.asyncio async def test_login_returns_chrome(self) -> None: """get_browser('login') 이 CHROME 을 반환하는지 확인""" choice = await get_browser("login") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_screenshot_returns_chrome(self) -> None: """get_browser('screenshot') 이 CHROME 을 반환하는지 확인""" choice = await get_browser("screenshot") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_pdf_returns_chrome(self) -> None: """get_browser('pdf') 이 CHROME 을 반환하는지 확인""" choice = await get_browser("pdf") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_spa_returns_chrome(self) -> None: """get_browser('spa') 이 CHROME 을 반환하는지 확인""" choice = await get_browser("spa") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_stealth_returns_chrome(self) -> None: """get_browser('stealth') 이 CHROME 을 반환하는지 확인""" choice = await get_browser("stealth") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_chrome_choice_endpoint(self) -> None: """Chrome 선택 시 endpoint 가 CHROME_ENDPOINT 와 일치하는지 확인""" choice = await get_browser("login") assert choice.endpoint == CHROME_ENDPOINT @pytest.mark.asyncio async def test_chrome_choice_has_reason(self) -> None: """Chrome 선택 시 reason 필드가 비어있지 않은지 확인""" choice = await get_browser("login") assert choice.reason # --------------------------------------------------------------------------- # 4. get_browser() — Lightpanda fallback (unhealthy) # --------------------------------------------------------------------------- class TestGetBrowserLightpandaFallback: @pytest.mark.asyncio async def test_crawl_fallback_to_chrome_when_unhealthy(self) -> None: """Lightpanda 불가 시 get_browser('crawl') 이 CHROME 으로 fallback 하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)): choice = await get_browser("crawl") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_fallback_reason_contains_fallback(self) -> None: """fallback 시 reason 에 'fallback' 단어가 포함되는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)): choice = await get_browser("crawl") assert "fallback" in choice.reason.lower() @pytest.mark.asyncio async def test_fallback_endpoint_is_chrome(self) -> None: """fallback 시 endpoint 가 CHROME_ENDPOINT 인지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)): choice = await get_browser("crawl") assert choice.endpoint == CHROME_ENDPOINT @pytest.mark.asyncio async def test_bulk_fallback_to_chrome_when_unhealthy(self) -> None: """Lightpanda 불가 시 get_browser('bulk') 이 CHROME 으로 fallback 하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)): choice = await get_browser("bulk") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_text_fallback_to_chrome_when_unhealthy(self) -> None: """Lightpanda 불가 시 get_browser('text') 이 CHROME 으로 fallback 하는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)): choice = await get_browser("text") assert choice.engine == BrowserEngine.CHROME # --------------------------------------------------------------------------- # 5. get_browser() — 알 수 없는 purpose # --------------------------------------------------------------------------- class TestGetBrowserUnknownPurpose: @pytest.mark.asyncio async def test_unknown_purpose_returns_chrome(self) -> None: """알 수 없는 purpose 는 CHROME 으로 기본 선택되는지 확인""" choice = await get_browser("unknown_purpose") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_unknown_purpose_endpoint_is_chrome(self) -> None: """알 수 없는 purpose 선택 시 endpoint 가 CHROME_ENDPOINT 인지 확인""" choice = await get_browser("unknown_purpose") assert choice.endpoint == CHROME_ENDPOINT @pytest.mark.asyncio async def test_empty_purpose_returns_chrome(self) -> None: """빈 문자열 purpose 도 CHROME 으로 기본 선택되는지 확인""" choice = await get_browser("") assert choice.engine == BrowserEngine.CHROME @pytest.mark.asyncio async def test_unknown_purpose_does_not_call_health_check(self) -> None: """알 수 없는 purpose 는 health check 를 호출하지 않는지 확인""" with patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)) as mock_health: await get_browser("unknown_purpose") mock_health.assert_not_called() # --------------------------------------------------------------------------- # 6. check_lightpanda_health() # --------------------------------------------------------------------------- class TestCheckLightpandaHealth: @pytest.mark.asyncio async def test_returns_true_on_http_200(self) -> None: """HTTP 200 응답 시 True 를 반환하는지 확인 (aiohttp mock)""" mock_resp = AsyncMock() mock_resp.status = 200 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) mock_resp.__aexit__ = AsyncMock(return_value=False) mock_session = AsyncMock() mock_session.get = MagicMock(return_value=mock_resp) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_aiohttp = MagicMock() mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock(return_value=MagicMock()) with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}): result = await check_lightpanda_health() assert result is True @pytest.mark.asyncio async def test_returns_false_on_http_error(self) -> None: """HTTP 요청 예외 발생 시 False 를 반환하는지 확인""" mock_session = AsyncMock() mock_session.get = MagicMock(side_effect=Exception("Connection refused")) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_aiohttp = MagicMock() mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock(return_value=MagicMock()) with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}): result = await check_lightpanda_health() assert result is False @pytest.mark.asyncio async def test_exception_does_not_propagate(self) -> None: """예외가 호출자에게 전파되지 않는지 확인""" mock_session = AsyncMock() mock_session.get = MagicMock(side_effect=RuntimeError("Unexpected error")) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_aiohttp = MagicMock() mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock(return_value=MagicMock()) with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}): # 예외가 외부로 전파되면 테스트 실패 result = await check_lightpanda_health() assert result is False @pytest.mark.asyncio async def test_returns_false_on_non_200_status(self) -> None: """HTTP 200 이 아닌 응답 시 False 를 반환하는지 확인""" mock_resp = AsyncMock() mock_resp.status = 503 mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) mock_resp.__aexit__ = AsyncMock(return_value=False) mock_session = AsyncMock() mock_session.get = MagicMock(return_value=mock_resp) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_aiohttp = MagicMock() mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock(return_value=MagicMock()) with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}): result = await check_lightpanda_health() assert result is False @pytest.mark.asyncio async def test_accepts_custom_endpoint(self) -> None: """커스텀 endpoint 파라미터를 수용하는지 확인 (예외 없이 실행)""" mock_session = AsyncMock() mock_session.get = MagicMock(side_effect=Exception("Connection refused")) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_aiohttp = MagicMock() mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock(return_value=MagicMock()) with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}): result = await check_lightpanda_health(endpoint="ws://custom:9333", timeout=1.0) assert result is False # --------------------------------------------------------------------------- # 7. BrowserChoice dataclass # --------------------------------------------------------------------------- class TestBrowserChoiceDataclass: def test_required_fields_exist(self) -> None: """BrowserChoice 에 필수 필드(engine, endpoint, reason)가 모두 존재하는지 확인""" field_names = {f.name for f in fields(BrowserChoice)} required = {"engine", "endpoint", "reason"} assert required.issubset(field_names), f"누락 필드: {required - field_names}" def test_instantiation(self) -> None: """BrowserChoice 인스턴스를 정상적으로 생성할 수 있는지 확인""" choice = BrowserChoice( engine=BrowserEngine.LIGHTPANDA, endpoint="ws://127.0.0.1:9333", reason="test reason", ) assert choice.engine == BrowserEngine.LIGHTPANDA assert choice.endpoint == "ws://127.0.0.1:9333" assert choice.reason == "test reason" def test_engine_field_is_browser_engine_instance(self) -> None: """engine 필드가 BrowserEngine 인스턴스인지 확인""" choice = BrowserChoice( engine=BrowserEngine.CHROME, endpoint="ws://127.0.0.1:9222", reason="chrome purpose", ) assert isinstance(choice.engine, BrowserEngine) def test_engine_lightpanda_value(self) -> None: """BrowserEngine.LIGHTPANDA 값이 'lightpanda' 인지 확인""" assert BrowserEngine.LIGHTPANDA == "lightpanda" assert str(BrowserEngine.LIGHTPANDA) == "BrowserEngine.LIGHTPANDA" def test_engine_chrome_value(self) -> None: """BrowserEngine.CHROME 값이 'chrome' 인지 확인""" assert BrowserEngine.CHROME == "chrome" assert str(BrowserEngine.CHROME) == "BrowserEngine.CHROME" def test_browser_choice_with_chrome_engine(self) -> None: """CHROME 엔진으로 BrowserChoice 생성 가능한지 확인""" choice = BrowserChoice( engine=BrowserEngine.CHROME, endpoint=CHROME_ENDPOINT, reason="rendering required", ) assert isinstance(choice.engine, BrowserEngine) assert choice.engine == BrowserEngine.CHROME # --------------------------------------------------------------------------- # 8. create_browser_context() # --------------------------------------------------------------------------- class TestCreateBrowserContext: def _make_mock_pw(self) -> tuple[MagicMock, MagicMock, MagicMock]: """(mock_pw_cm, mock_chromium, mock_browser) 헬퍼""" mock_context = MagicMock() mock_browser = MagicMock() mock_browser.contexts = [mock_context] mock_browser.close = AsyncMock() mock_chromium = MagicMock() mock_chromium.connect_over_cdp = AsyncMock(return_value=mock_browser) mock_pw_instance = MagicMock() mock_pw_instance.chromium = mock_chromium mock_pw_cm = MagicMock() mock_pw_cm.__aenter__ = AsyncMock(return_value=mock_pw_instance) mock_pw_cm.__aexit__ = AsyncMock(return_value=False) return mock_pw_cm, mock_chromium, mock_browser @pytest.mark.asyncio async def test_connects_to_lightpanda_endpoint_when_healthy(self) -> None: """Lightpanda healthy 시 connect_over_cdp 가 Lightpanda endpoint 로 호출되는지 확인""" mock_pw_cm, mock_chromium, _ = self._make_mock_pw() with ( patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=True)), patch("playwright.async_api.async_playwright", return_value=mock_pw_cm), ): async with create_browser_context("crawl") as (_browser, _context): pass mock_chromium.connect_over_cdp.assert_called_once_with(LIGHTPANDA_ENDPOINT) @pytest.mark.asyncio async def test_connects_to_chrome_endpoint_for_chrome_purpose(self) -> None: """Chrome purpose 시 connect_over_cdp 가 Chrome endpoint 로 호출되는지 확인""" mock_pw_cm, mock_chromium, _ = self._make_mock_pw() with patch("playwright.async_api.async_playwright", return_value=mock_pw_cm): async with create_browser_context("screenshot") as (_browser, _context): pass mock_chromium.connect_over_cdp.assert_called_once_with(CHROME_ENDPOINT) @pytest.mark.asyncio async def test_yields_browser_and_context_tuple(self) -> None: """create_browser_context 가 (browser, context) 튜플을 yield 하는지 확인""" mock_pw_cm, _, _ = self._make_mock_pw() with patch("playwright.async_api.async_playwright", return_value=mock_pw_cm): async with create_browser_context("login") as result: assert isinstance(result, tuple) assert len(result) == 2 @pytest.mark.asyncio async def test_browser_close_called_on_exit(self) -> None: """컨텍스트 매니저 종료 시 browser.close() 가 호출되는지 확인""" mock_pw_cm, _, mock_browser = self._make_mock_pw() with patch("playwright.async_api.async_playwright", return_value=mock_pw_cm): async with create_browser_context("login"): pass mock_browser.close.assert_called_once() @pytest.mark.asyncio async def test_fallback_endpoint_used_when_lightpanda_unhealthy(self) -> None: """Lightpanda unhealthy 시 Chrome endpoint 로 fallback 연결하는지 확인""" mock_pw_cm, mock_chromium, _ = self._make_mock_pw() with ( patch("tools.browser_router.check_lightpanda_health", new=AsyncMock(return_value=False)), patch("playwright.async_api.async_playwright", return_value=mock_pw_cm), ): async with create_browser_context("crawl") as (_browser, _context): pass mock_chromium.connect_over_cdp.assert_called_once_with(CHROME_ENDPOINT)