420 lines
16 KiB
Python
420 lines
16 KiB
Python
import os
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from _helpers import TempDirMixin
|
|
|
|
from app import accounts
|
|
from app import config as account_config
|
|
|
|
|
|
class AccountsTests(TempDirMixin, unittest.TestCase):
|
|
def make_config(self, temp_dir):
|
|
return {
|
|
"chrome_path": "chrome.exe",
|
|
"user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"),
|
|
"db_path": os.path.join(temp_dir, "cmshopee.db"),
|
|
"debug_port_range": [9222, 9224],
|
|
}
|
|
|
|
def test_create_account_stores_record_and_creates_user_data_dir(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
account = accounts.create_account(
|
|
"主店",
|
|
"alias",
|
|
"https://seller.shopee.tw/portal/",
|
|
9222,
|
|
password="secret",
|
|
note="备注",
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertEqual("主店", account.account_name)
|
|
self.assertEqual("alias", account.alias)
|
|
self.assertEqual("seller.shopee.tw", account.region_host)
|
|
self.assertEqual("alias_cdb6fdbe", account.slug)
|
|
self.assertTrue(os.path.isdir(account.user_data_dir))
|
|
self.assertEqual("******", accounts.mask_password(account.password))
|
|
self.assertEqual([account], accounts.list_accounts(config=cfg))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_update_account_can_change_alias_and_rebuild_directory(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
|
|
updated = accounts.update_account(
|
|
"alias",
|
|
"新主店",
|
|
"new alias",
|
|
"seller.shopee.tw",
|
|
9333,
|
|
password="new-secret",
|
|
note="新备注",
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertEqual("new alias", updated.alias)
|
|
self.assertEqual(account_config.make_slug("new alias"), updated.slug)
|
|
self.assertEqual(9333, updated.debug_port)
|
|
self.assertTrue(os.path.isdir(updated.user_data_dir))
|
|
with self.assertRaises(accounts.AccountError):
|
|
accounts.get_account("alias", config=cfg)
|
|
self.assertEqual("new-secret", accounts.get_account("new alias", config=cfg).password)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_next_debug_port_skips_existing_accounts(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("店1", "a1", debug_port=9222, config=cfg)
|
|
accounts.create_account("店2", "a2", debug_port=9223, config=cfg)
|
|
|
|
self.assertEqual(9224, accounts.next_debug_port(config=cfg))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_launch_for_login_starts_chrome_when_port_not_running(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
account = accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
|
|
class FakeProcess:
|
|
pid = 1234
|
|
|
|
initial_tab = {
|
|
"type": "page",
|
|
"id": "target-initial",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-initial",
|
|
}
|
|
|
|
with mock.patch("app.accounts.chrome.is_running", return_value=False) as is_running, \
|
|
mock.patch("app.accounts.chrome.launch_chrome", return_value=FakeProcess()) as launch, \
|
|
mock.patch("app.accounts.chrome.wait_debug_ready", return_value=True) as wait_ready, \
|
|
mock.patch("app.accounts.cdp.http_get", return_value=[initial_tab]), \
|
|
mock.patch("app.accounts.cdp.create_tab_info") as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
|
|
result = accounts.launch_for_login("alias", config=cfg)
|
|
|
|
self.assertEqual("launched", result["action"])
|
|
self.assertTrue(result["launched"])
|
|
self.assertFalse(result["reused"])
|
|
self.assertEqual(1234, result["pid"])
|
|
self.assertEqual("target-initial", result["target_id"])
|
|
self.assertFalse(result["created_tab"])
|
|
self.assertTrue(result["startup_target_reused"])
|
|
is_running.assert_called_once_with(9222)
|
|
launch.assert_called_once_with(
|
|
account,
|
|
config=cfg,
|
|
initial_url="https://seller.shopee.tw/portal/",
|
|
)
|
|
wait_ready.assert_called_once_with(9222, timeout=60)
|
|
create_tab_info.assert_not_called()
|
|
activate_tab.assert_called_once_with(
|
|
"target-initial",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_new_chrome_waits_for_delayed_initial_login_target(self):
|
|
account = {
|
|
"alias": "alias",
|
|
"region_host": "seller.shopee.tw",
|
|
"debug_port": 9222,
|
|
}
|
|
blank_tab = {
|
|
"type": "page",
|
|
"id": "target-blank",
|
|
"url": "about:blank",
|
|
"webSocketDebuggerUrl": "ws-blank",
|
|
}
|
|
portal_tab = {
|
|
"type": "page",
|
|
"id": "target-portal",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-portal",
|
|
}
|
|
probe_count = 0
|
|
|
|
def page_targets(*_args, **_kwargs):
|
|
nonlocal probe_count
|
|
probe_count += 1
|
|
return [blank_tab] if probe_count == 1 else [portal_tab]
|
|
|
|
with mock.patch("app.accounts.cdp.http_get", side_effect=page_targets), \
|
|
mock.patch("app.accounts.cdp.create_tab_info") as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
|
|
result = accounts._open_or_activate_login_tab(
|
|
account,
|
|
newly_launched=True,
|
|
startup_wait_timeout=0.3,
|
|
)
|
|
|
|
self.assertGreaterEqual(probe_count, 2)
|
|
self.assertEqual("target-portal", result["target_id"])
|
|
self.assertTrue(result["startup_target_reused"])
|
|
self.assertFalse(result["created_tab"])
|
|
create_tab_info.assert_not_called()
|
|
activate_tab.assert_called_once_with(
|
|
"target-portal",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
|
|
def test_new_chrome_navigates_its_only_blank_page(self):
|
|
account = {
|
|
"alias": "alias",
|
|
"region_host": "seller.shopee.tw",
|
|
"debug_port": 9222,
|
|
}
|
|
blank_tab = {
|
|
"type": "page",
|
|
"id": "target-blank",
|
|
"url": "about:blank",
|
|
"webSocketDebuggerUrl": "ws-blank",
|
|
}
|
|
client = mock.Mock()
|
|
client.send.return_value = {}
|
|
|
|
with mock.patch("app.accounts.cdp.http_get", return_value=[blank_tab]), \
|
|
mock.patch("app.accounts.cdp.CDP", return_value=client) as cdp_client, \
|
|
mock.patch("app.accounts.cdp.create_tab_info") as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
|
|
result = accounts._open_or_activate_login_tab(
|
|
account,
|
|
newly_launched=True,
|
|
startup_wait_timeout=0,
|
|
)
|
|
|
|
cdp_client.assert_called_once_with("ws-blank")
|
|
client.send.assert_called_once_with(
|
|
"Page.navigate",
|
|
{"url": "https://seller.shopee.tw/portal/"},
|
|
)
|
|
client.close.assert_called_once_with()
|
|
self.assertFalse(result["created_tab"])
|
|
self.assertTrue(result["startup_target_reused"])
|
|
self.assertTrue(result["startup_page_navigated"])
|
|
create_tab_info.assert_not_called()
|
|
activate_tab.assert_called_once_with(
|
|
"target-blank",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
|
|
def test_new_chrome_treats_intro_as_reusable_startup_page(self):
|
|
intro_tab = {
|
|
"type": "page",
|
|
"id": "target-intro",
|
|
"url": "chrome://intro/",
|
|
"webSocketDebuggerUrl": "ws-intro",
|
|
}
|
|
|
|
self.assertEqual(intro_tab, accounts._safe_startup_page([intro_tab]))
|
|
|
|
def test_new_chrome_without_page_creates_one_fallback_target(self):
|
|
account = {
|
|
"alias": "alias",
|
|
"region_host": "seller.shopee.tw",
|
|
"debug_port": 9222,
|
|
}
|
|
created_tab = {
|
|
"type": "page",
|
|
"id": "target-fallback",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-fallback",
|
|
}
|
|
|
|
with mock.patch("app.accounts.cdp.http_get", return_value=[]), \
|
|
mock.patch(
|
|
"app.accounts.cdp.create_tab_info",
|
|
return_value=created_tab,
|
|
) as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
|
|
result = accounts._open_or_activate_login_tab(
|
|
account,
|
|
newly_launched=True,
|
|
startup_wait_timeout=0,
|
|
)
|
|
|
|
self.assertTrue(result["created_tab"])
|
|
self.assertEqual("startup_page_missing", result["fallback_reason"])
|
|
create_tab_info.assert_called_once_with(
|
|
"https://seller.shopee.tw/portal/",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
activate_tab.assert_called_once_with(
|
|
"target-fallback",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
|
|
def test_running_chrome_preserves_normal_page_and_opens_login_tab(self):
|
|
account = {
|
|
"alias": "alias",
|
|
"region_host": "seller.shopee.tw",
|
|
"debug_port": 9222,
|
|
}
|
|
normal_tab = {
|
|
"type": "page",
|
|
"id": "target-user",
|
|
"url": "https://example.com/",
|
|
"webSocketDebuggerUrl": "ws-user",
|
|
}
|
|
created_tab = {
|
|
"type": "page",
|
|
"id": "target-login",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-login",
|
|
}
|
|
|
|
with mock.patch("app.accounts.cdp.http_get", return_value=[normal_tab]), \
|
|
mock.patch("app.accounts.cdp.CDP") as cdp_client, \
|
|
mock.patch(
|
|
"app.accounts.cdp.create_tab_info",
|
|
return_value=created_tab,
|
|
) as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab"):
|
|
result = accounts._open_or_activate_login_tab(
|
|
account,
|
|
newly_launched=False,
|
|
)
|
|
|
|
self.assertTrue(result["created_tab"])
|
|
self.assertEqual("existing_chrome_no_login_tab", result["fallback_reason"])
|
|
create_tab_info.assert_called_once()
|
|
cdp_client.assert_not_called()
|
|
|
|
def test_launch_for_login_reuses_running_chrome_without_starting_process(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
existing_tab = {
|
|
"type": "page",
|
|
"id": "target-existing",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-existing",
|
|
}
|
|
|
|
with mock.patch("app.accounts.chrome.is_running", return_value=True) as is_running, \
|
|
mock.patch("app.accounts.chrome.launch_chrome") as launch, \
|
|
mock.patch("app.accounts.chrome.wait_debug_ready") as wait_ready, \
|
|
mock.patch("app.accounts.cdp.http_get", return_value=[existing_tab]) as http_get, \
|
|
mock.patch("app.accounts.cdp.create_tab_info") as create_tab_info, \
|
|
mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
|
|
result = accounts.launch_for_login("alias", config=cfg)
|
|
|
|
self.assertEqual("reused", result["action"])
|
|
self.assertTrue(result["reused"])
|
|
self.assertFalse(result["launched"])
|
|
self.assertIsNone(result["pid"])
|
|
self.assertFalse(result["created_tab"])
|
|
self.assertEqual("target-existing", result["target_id"])
|
|
is_running.assert_called_once_with(9222)
|
|
http_get.assert_called_once_with("/json", host="127.0.0.1:9222")
|
|
activate_tab.assert_called_once_with("target-existing", host="127.0.0.1:9222")
|
|
launch.assert_not_called()
|
|
wait_ready.assert_not_called()
|
|
create_tab_info.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_create_shortcut_uses_account_record(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
account = accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
shortcut_path = os.path.join(temp_dir, "Desktop", "主店.lnk")
|
|
|
|
with mock.patch(
|
|
"app.accounts.chrome.create_shortcut",
|
|
return_value=shortcut_path,
|
|
) as create_shortcut:
|
|
result = accounts.create_shortcut(
|
|
"alias",
|
|
shortcut_path=shortcut_path,
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertEqual(shortcut_path, result)
|
|
create_shortcut.assert_called_once_with(
|
|
account,
|
|
shortcut_path=shortcut_path,
|
|
desktop_dir=None,
|
|
config=cfg,
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_detect_login_updates_last_login_at_when_logged_in(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
|
|
with mock.patch(
|
|
"app.accounts.editor.login_status",
|
|
return_value={"logged_in": True, "reason": None},
|
|
) as login_status:
|
|
status = accounts.detect_login("alias", timeout=1, config=cfg)
|
|
|
|
self.assertTrue(status["logged_in"])
|
|
login_status.assert_called_once()
|
|
self.assertIsNotNone(accounts.get_account("alias", config=cfg).last_login_at)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_login_status_text_marks_target_failure_as_temporarily_uncertain(self):
|
|
self.assertEqual(
|
|
"登录状态暂不可确认",
|
|
accounts.login_status_text(
|
|
{
|
|
"logged_in": False,
|
|
"reason": "LOGIN_CHECK_TARGET_UNAVAILABLE",
|
|
}
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"登录状态暂不可确认",
|
|
accounts.login_status_text(
|
|
{
|
|
"logged_in": False,
|
|
"reason": "LOGIN_CHECK_FAILED: Target closed",
|
|
}
|
|
),
|
|
)
|
|
|
|
def test_duplicate_alias_raises_clear_error(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("店1", "alias", debug_port=9222, config=cfg)
|
|
|
|
with self.assertRaises(accounts.AccountError):
|
|
accounts.create_account("店2", "alias", debug_port=9223, config=cfg)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_duplicate_debug_port_raises_clear_error(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("店1", "a1", debug_port=9222, config=cfg)
|
|
accounts.create_account("店2", "a2", debug_port=9223, config=cfg)
|
|
|
|
with self.assertRaises(accounts.AccountError):
|
|
accounts.create_account("店3", "a3", debug_port=9222, config=cfg)
|
|
with self.assertRaises(accounts.AccountError):
|
|
accounts.update_account(
|
|
"a2",
|
|
"店2",
|
|
"a2",
|
|
"seller.shopee.tw",
|
|
9222,
|
|
config=cfg,
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|