cmshoppe/tests/test_cdp.py

import os
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(__file__))

from _helpers import REPO_ROOT

if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from app import cdp


class FakeBrowserCDP:
    sent = []
    fail_first_create = False

    def __init__(self, ws_url):
        self.ws_url = ws_url
        self.closed = False

    def send(self, method, params=None):
        self.sent.append((method, params or {}))
        if method == "Target.createTarget":
            if self.fail_first_create:
                type(self).fail_first_create = False
                raise RuntimeError("Target.createTarget: Invalid parameters")
            return {"targetId": "target-new"}
        return {}

    def close(self):
        self.closed = True


class CdpTests(unittest.TestCase):
    def test_wait_target_closed_returns_when_target_disappears(self):
        targets = [
            [{"id": "target-closing", "type": "page"}],
            [],
        ]
        with mock.patch("app.cdp.http_get", side_effect=targets) as http_get, mock.patch(
            "app.cdp.time.sleep"
        ) as sleep:
            closed = cdp.wait_target_closed(
                "target-closing",
                host="127.0.0.1:9222",
                timeout=1.0,
                poll_interval=0.1,
            )

        self.assertTrue(closed)
        sleep.assert_called_once_with(0.1)
        self.assertLessEqual(http_get.call_args_list[0].kwargs["timeout"], 1.0)

    def test_wait_target_closed_stops_at_timeout(self):
        with mock.patch(
            "app.cdp.http_get",
            return_value=[{"id": "target-closing", "type": "page"}],
        ) as http_get, mock.patch("app.cdp.time.sleep") as sleep:
            closed = cdp.wait_target_closed(
                "target-closing",
                host="127.0.0.1:9222",
                timeout=0,
            )

        self.assertFalse(closed)
        sleep.assert_not_called()
        self.assertEqual(0.05, http_get.call_args.kwargs["timeout"])

    def test_close_tab_and_wait_only_waits_after_close_request_succeeds(self):
        with mock.patch("app.cdp.close_tab", return_value=True) as close_tab, mock.patch(
            "app.cdp.wait_target_closed",
            return_value=True,
        ) as wait_target_closed:
            closed = cdp.close_tab_and_wait(
                "target-closing",
                host="127.0.0.1:9222",
                timeout=1.5,
            )

        self.assertTrue(closed)
        close_tab.assert_called_once_with("target-closing", host="127.0.0.1:9222")
        wait_target_closed.assert_called_once_with(
            "target-closing",
            host="127.0.0.1:9222",
            timeout=1.5,
            poll_interval=0.1,
        )

    def test_create_tab_info_can_request_background_target(self):
        FakeBrowserCDP.sent = []
        FakeBrowserCDP.fail_first_create = False
        responses = [
            {"webSocketDebuggerUrl": "ws-browser"},
            [
                {
                    "id": "target-new",
                    "type": "page",
                    "url": "https://seller.shopee.tw/portal/product/1",
                    "webSocketDebuggerUrl": "ws-page",
                }
            ],
        ]

        with mock.patch("app.cdp.http_get", side_effect=responses), \
                mock.patch("app.cdp.CDP", FakeBrowserCDP), \
                mock.patch("app.cdp.time.sleep"):
            tab = cdp.create_tab_info(
                "https://seller.shopee.tw/portal/product/1",
                host="127.0.0.1:9222",
                background=True,
            )

        self.assertEqual("ws-page", tab["webSocketDebuggerUrl"])
        self.assertEqual(
            [
                (
                    "Target.createTarget",
                    {
                        "url": "https://seller.shopee.tw/portal/product/1",
                        "background": True,
                    },
                )
            ],
            FakeBrowserCDP.sent,
        )

    def test_create_tab_info_keeps_foreground_default_params(self):
        FakeBrowserCDP.sent = []
        FakeBrowserCDP.fail_first_create = False
        responses = [
            {"webSocketDebuggerUrl": "ws-browser"},
            [
                {
                    "id": "target-new",
                    "type": "page",
                    "url": "https://seller.shopee.tw/portal/product/1",
                    "webSocketDebuggerUrl": "ws-page",
                }
            ],
        ]

        with mock.patch("app.cdp.http_get", side_effect=responses), \
                mock.patch("app.cdp.CDP", FakeBrowserCDP), \
                mock.patch("app.cdp.time.sleep"):
            cdp.create_tab_info(
                "https://seller.shopee.tw/portal/product/1",
                host="127.0.0.1:9222",
            )

        self.assertEqual(
            [
                (
                    "Target.createTarget",
                    {"url": "https://seller.shopee.tw/portal/product/1"},
                )
            ],
            FakeBrowserCDP.sent,
        )

    def test_create_tab_info_falls_back_when_background_is_rejected(self):
        FakeBrowserCDP.sent = []
        FakeBrowserCDP.fail_first_create = True
        responses = [
            {"webSocketDebuggerUrl": "ws-browser"},
            [
                {
                    "id": "target-new",
                    "type": "page",
                    "url": "https://seller.shopee.tw/portal/product/1",
                    "webSocketDebuggerUrl": "ws-page",
                }
            ],
        ]

        with mock.patch("app.cdp.http_get", side_effect=responses), \
                mock.patch("app.cdp.CDP", FakeBrowserCDP), \
                mock.patch("app.cdp.time.sleep"):
            cdp.create_tab_info(
                "https://seller.shopee.tw/portal/product/1",
                host="127.0.0.1:9222",
                background=True,
            )

        self.assertEqual(
            [
                (
                    "Target.createTarget",
                    {
                        "url": "https://seller.shopee.tw/portal/product/1",
                        "background": True,
                    },
                ),
                (
                    "Target.createTarget",
                    {"url": "https://seller.shopee.tw/portal/product/1"},
                ),
            ],
            FakeBrowserCDP.sent,
        )


if __name__ == "__main__":
    unittest.main()