rogee
e50bed9ad8
douyin-release-gate / verify (push) Failing after 18m54s
Add Douyin notification polling, event details, and manual multi-account private messaging. Refine environment memory settings, account operations, login collection recovery, and message UI; update tests and documentation.
188 lines
8.9 KiB
Python
188 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from importlib import import_module
|
|
from unittest.mock import Mock, patch
|
|
|
|
from .platform.douyin import CHAT_PAGE_NAME, CHAT_PAGE_URL, DouyinBrowser, DouyinError, ORIGIN
|
|
|
|
|
|
douyin_module = import_module(f"{__package__}.platform.douyin")
|
|
|
|
|
|
class BrowserTargetSelectionTests(unittest.TestCase):
|
|
@staticmethod
|
|
def page(target_id: str, url: str = ORIGIN) -> dict:
|
|
return {
|
|
"id": target_id,
|
|
"type": "page",
|
|
"url": url,
|
|
"webSocketDebuggerUrl": f"ws://127.0.0.1:9222/devtools/page/{target_id}",
|
|
}
|
|
|
|
@contextmanager
|
|
def targets(self, pages: list, browser: DouyinBrowser | None = None):
|
|
browser = browser or DouyinBrowser(lambda _: "http://browser:9222")
|
|
response = Mock(status=200)
|
|
response.read.return_value = json.dumps(pages).encode()
|
|
http = Mock()
|
|
http.getresponse.return_value = response
|
|
cdp = Mock()
|
|
with (
|
|
patch.object(douyin_module.http.client, "HTTPConnection", return_value=http),
|
|
patch.object(douyin_module.websocket, "create_connection") as connect,
|
|
patch.object(douyin_module, "CDPConnection", return_value=cdp),
|
|
):
|
|
yield browser, connect, cdp
|
|
|
|
def assert_selected(self, connect: Mock, target_id: str) -> None:
|
|
connect.assert_called_once_with(
|
|
f"ws://browser:9222/devtools/page/{target_id}",
|
|
timeout=douyin_module.CONTROL_TIMEOUT,
|
|
suppress_origin=True,
|
|
enable_multithread=True,
|
|
)
|
|
|
|
def test_multiple_douyin_pages_prefer_self_profile_in_any_order(self) -> None:
|
|
pages = [
|
|
self.page("A", ORIGIN + "/video/123"),
|
|
self.page("Z", ORIGIN + "/user/self?showTab=post"),
|
|
self.page("B", ORIGIN + "/"),
|
|
]
|
|
for ordered in (pages, list(reversed(pages))):
|
|
with self.subTest(order=[page["id"] for page in ordered]):
|
|
with self.targets(ordered) as (browser, connect, cdp):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "Z")
|
|
cdp.command.assert_not_called()
|
|
cdp.notify.assert_not_called()
|
|
|
|
def test_multiple_profile_pages_use_stable_target_id(self) -> None:
|
|
pages = [
|
|
self.page("Z", ORIGIN + "/user/self"),
|
|
self.page("A", ORIGIN + "/user/self?showTab=like"),
|
|
]
|
|
with self.targets(pages) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "A")
|
|
|
|
def test_multiple_douyin_pages_without_profile_use_stable_target_id(self) -> None:
|
|
pages = [self.page("Z", ORIGIN + "/video/123"), self.page("A")]
|
|
for ordered in (pages, list(reversed(pages))):
|
|
with self.subTest(order=[page["id"] for page in ordered]):
|
|
with self.targets(ordered) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "A")
|
|
|
|
def test_unrelated_pages_and_non_page_targets_are_not_selected(self) -> None:
|
|
pages = [
|
|
self.page("A", "https://example.com/"),
|
|
{"id": "worker", "type": "service_worker", "url": ORIGIN},
|
|
None,
|
|
self.page("Z", ORIGIN + "/video/123"),
|
|
self.page("B", ORIGIN + "/user/self"),
|
|
]
|
|
with self.targets(pages) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "B")
|
|
|
|
def test_dedicated_chat_target_is_not_used_for_regular_operations(self) -> None:
|
|
browser = DouyinBrowser(lambda _: "http://browser:9222")
|
|
browser._chat_targets["safe"] = "A"
|
|
pages = [self.page("A", CHAT_PAGE_URL), self.page("Z"), self.page("B")]
|
|
with self.targets(pages, browser) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "B")
|
|
self.assertEqual(browser._chat_targets["safe"], "A")
|
|
|
|
def test_dedicated_chat_page_is_identified_after_gateway_restart(self) -> None:
|
|
pages = [self.page("chat", CHAT_PAGE_URL + "#" + CHAT_PAGE_NAME), self.page("B"), self.page("A")]
|
|
with self.targets(pages) as (browser, connect, cdp):
|
|
cdp.evaluate.return_value = CHAT_PAGE_NAME
|
|
browser._connect("safe")
|
|
self.assertEqual(connect.call_count, 2)
|
|
self.assertEqual(connect.call_args.args[0], "ws://browser:9222/devtools/page/A")
|
|
cdp.close.assert_called_once()
|
|
cdp.command.assert_not_called()
|
|
cdp.notify.assert_not_called()
|
|
|
|
def test_only_dedicated_chat_page_after_restart_is_not_reused(self) -> None:
|
|
with self.targets([self.page("chat", CHAT_PAGE_URL + "#" + CHAT_PAGE_NAME)]) as (browser, connect, cdp):
|
|
cdp.evaluate.return_value = CHAT_PAGE_NAME
|
|
with self.assertRaisesRegex(DouyinError, "page target is unavailable"):
|
|
browser._connect("safe")
|
|
# Only the read-only page-role probe connects; no account request follows.
|
|
connect.assert_called_once()
|
|
cdp.close.assert_called_once()
|
|
cdp.command.assert_not_called()
|
|
cdp.notify.assert_not_called()
|
|
|
|
def test_only_dedicated_chat_page_is_not_reused(self) -> None:
|
|
browser = DouyinBrowser(lambda _: "http://browser:9222")
|
|
browser._chat_targets["safe"] = "chat"
|
|
with self.targets([self.page("chat", CHAT_PAGE_URL)], browser) as (browser, connect, _):
|
|
with self.assertRaisesRegex(DouyinError, "page target is unavailable"):
|
|
browser._connect("safe")
|
|
connect.assert_not_called()
|
|
|
|
def test_chat_page_with_blank_regular_page_preserves_initial_login(self) -> None:
|
|
browser = DouyinBrowser(lambda _: "http://browser:9222")
|
|
browser._chat_targets["safe"] = "chat"
|
|
pages = [self.page("chat", CHAT_PAGE_URL), self.page("blank", "about:blank")]
|
|
with self.targets(pages, browser) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "blank")
|
|
|
|
def test_explicit_target_remains_pinned_even_for_private_chat(self) -> None:
|
|
browser = DouyinBrowser(lambda _: "http://browser:9222", target_id="chat")
|
|
browser._chat_targets["safe"] = "chat"
|
|
pages = [self.page("main", ORIGIN + "/user/self"), self.page("chat", CHAT_PAGE_URL)]
|
|
with self.targets(pages, browser) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "chat")
|
|
|
|
def test_missing_explicit_target_does_not_choose_another_page(self) -> None:
|
|
browser = DouyinBrowser(lambda _: "http://browser:9222", target_id="missing")
|
|
with self.targets([self.page("A"), self.page("B")], browser) as (browser, connect, _):
|
|
with self.assertRaisesRegex(DouyinError, "configured browser page target is unavailable"):
|
|
browser._connect("safe")
|
|
connect.assert_not_called()
|
|
|
|
def test_multiple_unrelated_pages_are_not_navigated_arbitrarily(self) -> None:
|
|
pages = [self.page("A", "https://example.com/"), self.page("B", "about:blank")]
|
|
with self.targets(pages) as (browser, connect, _):
|
|
with self.assertRaises(DouyinError):
|
|
browser._connect("safe")
|
|
connect.assert_not_called()
|
|
|
|
def test_single_blank_page_still_supports_initial_login(self) -> None:
|
|
with self.targets([self.page("blank", "about:blank")]) as (browser, connect, _):
|
|
browser._connect("safe")
|
|
self.assert_selected(connect, "blank")
|
|
|
|
def test_multi_page_identity_verifies_bound_uid_and_closes_connection(self) -> None:
|
|
payload = {"status_code": 0, "user": {"uid": "1004291301", "sec_uid": "safe_sec_uid"}}
|
|
pages = [self.page("A", ORIGIN + "/video/123"), self.page("B", ORIGIN + "/user/self")]
|
|
with self.targets(pages) as (browser, connect, cdp):
|
|
cdp.evaluate.side_effect = [ORIGIN, {"status": 200, "body": json.dumps(payload)}]
|
|
identity = browser.identity("safe", "1004291301")
|
|
self.assertEqual(identity["uid"], "1004291301")
|
|
self.assert_selected(connect, "B")
|
|
cdp.close.assert_called_once()
|
|
cdp.notify.assert_not_called()
|
|
|
|
def test_multi_page_identity_still_rejects_wrong_uid(self) -> None:
|
|
payload = {"status_code": 0, "user": {"uid": "987654321", "sec_uid": "safe_sec_uid"}}
|
|
with self.targets([self.page("A"), self.page("B")]) as (browser, connect, cdp):
|
|
cdp.evaluate.side_effect = [ORIGIN, {"status": 200, "body": json.dumps(payload)}]
|
|
with self.assertRaisesRegex(DouyinError, "does not match the expected account"):
|
|
browser.identity("safe", "1004291301")
|
|
self.assert_selected(connect, "A")
|
|
cdp.close.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|