228 lines
9.5 KiB
Python
228 lines
9.5 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
|
|
|
|
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=[]), \
|
|
mock.patch(
|
|
"app.accounts.cdp.create_tab_info",
|
|
return_value={
|
|
"id": "target-new",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-new",
|
|
},
|
|
) 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-new", result["target_id"])
|
|
is_running.assert_called_once_with(9222)
|
|
launch.assert_called_once_with(account, config=cfg)
|
|
wait_ready.assert_called_once_with(9222, timeout=60)
|
|
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-new", host="127.0.0.1:9222")
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
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_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()
|