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()