creator-hub/browser_gateway/test_xiaohongshu.py

from __future__ import annotations

import unittest
from contextlib import contextmanager
from typing import Any, cast
from unittest.mock import Mock

from .platform import xiaohongshu as xiaohongshu_module
from .server import http as gateway_module

Gateway = gateway_module.Gateway
RequestError = gateway_module.RequestError
valid_xhs_post_url = gateway_module.valid_xhs_post_url
valid_xhs_url = gateway_module.valid_xhs_url
valid_xiaohongshu_url = gateway_module.valid_xiaohongshu_url
valid_xiaohongshu_media_url = gateway_module.valid_xiaohongshu_media_url
valid_xiaohongshu_generation = gateway_module.valid_xiaohongshu_generation
valid_xiaohongshu_page_url = gateway_module.valid_xiaohongshu_page_url
valid_xiaohongshu_source_url = gateway_module.valid_xiaohongshu_source_url


class XiaohongshuValidationTests(unittest.TestCase):
    def test_read_urls_use_explicit_host_path_and_query_allowlist(self) -> None:
        self.assertTrue(
            valid_xhs_url("https://edith.xiaohongshu.com/api/sns/web/v2/user/me")
        )
        self.assertTrue(
            valid_xiaohongshu_url(
                "https://edith.xiaohongshu.com/api/sns/web/v2/user/me"
            )
        )
        self.assertTrue(
            valid_xiaohongshu_page_url("https://www.xiaohongshu.com/user/profile/u-1")
        )
        self.assertTrue(valid_xiaohongshu_source_url("https://xhslink.com/a/abc"))
        self.assertTrue(
            valid_xhs_url(
                "https://edith.xiaohongshu.com/api/sns/web/v1/user_posted?user_id=u-1&cursor=&num=30&xsec_source=pc_user"
            )
        )
        self.assertFalse(
            valid_xhs_url(
                "https://edith.xiaohongshu.com/api/sns/web/v1/user_posted?user_id=u-1&num=10"
            )
        )
        self.assertFalse(
            valid_xhs_url("https://edith.xiaohongshu.com.evil/api/sns/web/v2/user/me")
        )
        self.assertTrue(
            valid_xhs_post_url("https://so.xiaohongshu.com/api/sns/web/v2/search/notes")
        )
        self.assertTrue(
            valid_xhs_post_url("https://edith.xiaohongshu.com/api/sns/web/v1/feed")
        )
        self.assertTrue(
            valid_xiaohongshu_media_url(
                "https://www.xiaohongshu.com/explore/n-1?xsec_source=pc_search"
            )
        )
        self.assertFalse(
            valid_xiaohongshu_media_url(
                "https://www.xiaohongshu.com/explore/n-1#fragment"
            )
        )

    def test_generation_shape_matches_existing_browser_fence(self) -> None:
        self.assertTrue(
            valid_xiaohongshu_generation(
                {
                    "binding_version": 1,
                    "runtime_id": "a" * 64,
                    "network_id": "native-" + "b" * 32,
                    "network_exit_id": "",
                }
            )
        )
        self.assertFalse(
            valid_xiaohongshu_generation(
                {"binding_version": 1, "runtime_id": "runtime", "network_id": "network"}
            )
        )

    def test_facade_exports_shared_browser(self) -> None:
        self.assertIs(
            xiaohongshu_module.XiaohongshuBrowser,
            gateway_module.XiaohongshuBrowser,
        )


class XiaohongshuBrowserTests(unittest.TestCase):
    def test_resolve_allows_anonymous_blank_runtime(self) -> None:
        browser = xiaohongshu_module.XiaohongshuBrowser()
        cdp = Mock()
        cdp.evaluate.side_effect = [
            "null",
            True,
            "https://www.xiaohongshu.com/explore/n-1",
        ]
        cdp.command.return_value = {"frameId": "frame-1"}
        cdp.wait_event.return_value = {
            "params": {"frame": {"url": "https://www.xiaohongshu.com/explore/n-1"}}
        }

        @contextmanager
        def bound(alias: str):
            del alias
            yield cdp

        cast(Any, browser).connection = bound
        self.assertEqual(
            browser.resolve("safe", "https://xhslink.com/a/abc"),
            "https://www.xiaohongshu.com/explore/n-1",
        )


class XiaohongshuRouteTests(unittest.TestCase):
    def test_read_only_routes_dispatch_without_action_or_event_routes(self) -> None:
        handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler)
        gateway = Mock()
        gateway.get_xiaohongshu.return_value = {"status": 200}
        gateway.post_xiaohongshu.return_value = {"status": 200}
        gateway.get_xiaohongshu_media.return_value = {"status": 200}
        gateway.xiaohongshu_identity.return_value = {"uid": "u-1"}
        gateway.resolve_xiaohongshu.return_value = {
            "url": "https://www.xiaohongshu.com/explore/n-1"
        }
        server = Mock()
        server.gateway = gateway
        cast(Any, handler).server = server
        cast(Any, handler).server_as_gateway = lambda: server

        self.assertEqual(
            handler._route("POST", "/v1/browsers/account-a/xiaohongshu/get", {}, {}),
            {"status": 200},
        )
        self.assertEqual(
            handler._route("POST", "/v1/browsers/account-a/xiaohongshu/post", {}, {}),
            {"status": 200},
        )
        self.assertEqual(
            handler._route("POST", "/v1/browsers/account-a/xiaohongshu/media", {}, {}),
            {"status": 200},
        )
        self.assertEqual(
            handler._route(
                "POST", "/v1/browsers/account-a/xiaohongshu/identity", {}, {}
            ),
            {"uid": "u-1"},
        )
        self.assertEqual(
            handler._route(
                "POST", "/v1/browsers/account-a/xiaohongshu/resolve", {}, {}
            ),
            {"url": "https://www.xiaohongshu.com/explore/n-1"},
        )
        with self.assertRaises(RequestError):
            handler._route("POST", "/v1/browsers/account-a/xiaohongshu/action", {}, {})
        gateway.get_xiaohongshu.assert_called_once_with("account-a", {})
        gateway.post_xiaohongshu.assert_called_once_with("account-a", {})
        gateway.get_xiaohongshu_media.assert_called_once_with("account-a", {})
        gateway.xiaohongshu_identity.assert_called_once_with("account-a", {})
        gateway.resolve_xiaohongshu.assert_called_once_with("account-a", {})