api-camoufox/tests/test_wrapper.py

import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from app.use_cases.camoufox.wrapper import (
    CamoufoxWrapper,
    _detect_headless_mode,
    _resolve_headless_mode,
)


def _run_async(coro):
    return asyncio.run(coro)


class TestDetectHeadlessMode:
    @patch("sys.platform", "darwin")
    def test_macos_returns_true(self):
        result = _detect_headless_mode()
        assert result is True

    @patch("sys.platform", "win32")
    def test_windows_returns_true(self):
        result = _detect_headless_mode()
        assert result is True

    @patch("shutil.which", return_value=None)
    @patch("sys.platform", "linux")
    def test_linux_without_xvfb_returns_true(self, mock_which):
        result = _detect_headless_mode()
        assert result is True

    @patch("shutil.which", return_value="/usr/bin/Xvfb")
    @patch("sys.platform", "linux")
    def test_linux_with_xvfb_returns_virtual(self, mock_which):
        result = _detect_headless_mode()
        assert result == "virtual"


class TestResolveHeadlessMode:
    @patch("app.use_cases.camoufox.wrapper.config")
    def test_env_virtual_returns_virtual(self, mock_config):
        mock_config.HEADLESS_MODE = "virtual"
        result = _resolve_headless_mode()
        assert result == "virtual"

    @patch("app.use_cases.camoufox.wrapper.config")
    def test_env_true_returns_true(self, mock_config):
        for val in ["true", "True", "1"]:
            mock_config.HEADLESS_MODE = val
            result = _resolve_headless_mode()
            assert result is True

    @patch("app.use_cases.camoufox.wrapper.config")
    def test_env_false_returns_false(self, mock_config):
        for val in ["false", "False", "0"]:
            mock_config.HEADLESS_MODE = val
            result = _resolve_headless_mode()
            assert result is False

    @patch("app.use_cases.camoufox.wrapper.config")
    @patch("app.use_cases.camoufox.wrapper._detect_headless_mode")
    def test_env_auto_uses_detection(self, mock_detect, mock_config):
        mock_config.HEADLESS_MODE = "auto"
        mock_detect.return_value = True
        result = _resolve_headless_mode()
        mock_detect.assert_called_once()
        assert result is True


class TestCamoufoxWrapper:
    def test_singleton(self):
        wrapper1 = CamoufoxWrapper()
        wrapper2 = CamoufoxWrapper()
        assert wrapper1 is wrapper2

    def test_build_proxy_config_simple(self):
        wrapper = CamoufoxWrapper()
        result = wrapper._build_proxy_config("host:8080")
        assert result["server"] == "http://host:8080"

    def test_build_proxy_config_with_scheme(self):
        wrapper = CamoufoxWrapper()
        result = wrapper._build_proxy_config("http://host:8080")
        assert result["server"] == "http://host:8080"

    def test_build_proxy_config_with_auth(self):
        wrapper = CamoufoxWrapper()
        result = wrapper._build_proxy_config(
            "host:8080", {"username": "user", "password": "pass"}
        )
        assert result["server"] == "http://host:8080"
        assert result["username"] == "user"
        assert result["password"] == "pass"

    def test_build_proxy_config_with_auth_no_password(self):
        wrapper = CamoufoxWrapper()
        result = wrapper._build_proxy_config(
            "host:8080", {"username": "user"}
        )
        assert result["server"] == "http://host:8080"
        assert result["username"] == "user"
        assert "password" not in result


class TestCamoufoxWrapperScreenshot:
    def test_get_screenshot_quality(self):
        wrapper = CamoufoxWrapper()
        assert wrapper._get_screenshot_quality() == 60

    def test_apply_screenshot_delay_zero(self):
        wrapper = CamoufoxWrapper()
        with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
            _run_async(wrapper._apply_screenshot_delay({"screenshotDelay": 0}))
            mock_sleep.assert_not_awaited()

    def test_apply_screenshot_delay_positive(self):
        wrapper = CamoufoxWrapper()
        with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
            _run_async(wrapper._apply_screenshot_delay({"screenshotDelay": 500}))
            mock_sleep.assert_awaited_once_with(0.5)

    def test_click_element_no_selector(self):
        wrapper = CamoufoxWrapper()
        page = MagicMock()
        _run_async(wrapper._click_element(page, {}))
        page.wait_for_selector.assert_not_called()
        page.click.assert_not_called()

    def test_click_element_success(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        _run_async(wrapper._click_element(page, {"clickSelector": "button#btn", "selectorTimeout": 2000}))
        page.wait_for_selector.assert_awaited_once_with("button#btn", timeout=2000)
        page.click.assert_awaited_once_with("button#btn")

    def test_click_element_failure_logs_warning(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        page.wait_for_selector.side_effect = Exception("timeout")
        with patch("app.use_cases.camoufox.wrapper.log.logger.warning") as mock_warn:
            _run_async(wrapper._click_element(page, {"clickSelector": "button#btn"}))
            mock_warn.assert_called_once()

    def test_take_screenshot_full_page(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        page.screenshot.return_value = b"fullpage"
        result = _run_async(wrapper._take_screenshot(page, {}))
        page.screenshot.assert_awaited_once_with(full_page=True, type="jpeg", quality=60)
        assert result == b"fullpage"

    def test_take_screenshot_by_selector(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        element = AsyncMock()
        element.screenshot.return_value = b"element"
        page.query_selector.return_value = element
        result = _run_async(
            wrapper._take_screenshot(page, {"screenshotSelector": "#product", "selectorTimeout": 2000})
        )
        page.wait_for_selector.assert_awaited_once_with("#product", timeout=2000)
        page.query_selector.assert_awaited_once_with("#product")
        element.screenshot.assert_awaited_once_with(type="jpeg", quality=60)
        assert result == b"element"

    def test_take_screenshot_selector_fallback_to_full_page(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        page.query_selector.return_value = None
        page.screenshot.return_value = b"fallback"
        with patch("app.use_cases.camoufox.wrapper.log.logger.warning") as mock_warn:
            result = _run_async(
                wrapper._take_screenshot(page, {"screenshotSelector": "#missing"})
            )
            mock_warn.assert_called_once()
            page.screenshot.assert_awaited_once_with(full_page=True, type="jpeg", quality=60)
            assert result == b"fallback"

    def test_take_screenshot_with_delay(self):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        page.screenshot.return_value = b"delayed"
        with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
            result = _run_async(wrapper._take_screenshot(page, {"screenshotDelay": 300}))
            mock_sleep.assert_awaited_once_with(0.3)
            assert result == b"delayed"


class TestCamoufoxWrapperVisitScreenshotFlow:
    @patch("app.use_cases.camoufox.wrapper._resolve_headless_mode", return_value=True)
    @patch("app.use_cases.camoufox.wrapper.AsyncCamoufox")
    @patch("app.use_cases.camoufox.wrapper.base64.b64encode", return_value=b"base64data")
    def test_visit_triggers_screenshot_with_screenshot_selector(
        self, mock_b64encode, mock_browser_cls, mock_headless
    ):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        element = AsyncMock()
        element.screenshot.return_value = b"element-shot"
        page.query_selector.return_value = element
        page.content.return_value = "<html></html>"
        browser = AsyncMock()
        browser.new_page.return_value = page
        mock_browser_cls.return_value.__aenter__ = AsyncMock(return_value=browser)
        mock_browser_cls.return_value.__aexit__ = AsyncMock(return_value=False)

        result = _run_async(
            wrapper.visit("https://example.com", {"screenshotSelector": "#product"})
        )

        page.wait_for_selector.assert_awaited_once_with("#product", timeout=1000)
        element.screenshot.assert_awaited_once()
        assert result["screenshot"] == "base64data"

    @patch("app.use_cases.camoufox.wrapper._resolve_headless_mode", return_value=True)
    @patch("app.use_cases.camoufox.wrapper.AsyncCamoufox")
    def test_visit_triggers_click_then_screenshot(
        self, mock_browser_cls, mock_headless
    ):
        wrapper = CamoufoxWrapper()
        page = AsyncMock()
        page.content.return_value = "<html></html>"
        page.screenshot.return_value = b"shot"
        browser = AsyncMock()
        browser.new_page.return_value = page
        mock_browser_cls.return_value.__aenter__ = AsyncMock(return_value=browser)
        mock_browser_cls.return_value.__aexit__ = AsyncMock(return_value=False)

        _run_async(wrapper.visit("https://example.com", {"clickSelector": "button"}))

        page.wait_for_selector.assert_any_await("button", timeout=1000)
        page.click.assert_awaited_once_with("button")
        page.screenshot.assert_awaited_once()