chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(accounts): keep first login launch in one window

Browse Source
This commit is contained in:
chengma
2026-07-14 17:22:52 +08:00
parent a4ac3f8ee5
commit 699625ea0e
10 changed files with 608 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_accounts.py
+186 -14
View File
@@ -84,18 +84,18 @@ class AccountsTests(TempDirMixin, unittest.TestCase):
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=[]), \
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.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)
@@ -103,18 +103,190 @@ class AccountsTests(TempDirMixin, unittest.TestCase):
self.assertTrue(result["launched"])
self.assertFalse(result["reused"])
self.assertEqual(1234, result["pid"])
self.assertEqual("target-new", result["target_id"])
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)
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_called_once_with(
"https://seller.shopee.tw/portal/",
create_tab_info.assert_not_called()
activate_tab.assert_called_once_with(
"target-initial",
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_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)
tests/test_chrome.py
+51 -1
View File
@@ -236,6 +236,8 @@ class ChromeTests(TempDirMixin, unittest.TestCase):
self.assertIn("--remote-debugging-port=9222", args)
self.assertIn("--remote-allow-origins=*", args)
self.assertIn(f"--user-data-dir={user_data_dir}", args)
self.assertNotIn("--no-first-run", args)
self.assertNotIn("--no-default-browser-check", args)
self.assertTrue(os.path.isdir(user_data_dir))
self.assert_removed(temp_dir)
@@ -255,6 +257,45 @@ class ChromeTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_build_launch_args_appends_valid_initial_url_once(self):
with self.make_temp_dir() as temp_dir:
cfg = {"chrome_path": "chrome.exe", "user_data_root": temp_dir}
account = {"alias": "alias", "debug_port": 9222}
initial_url = "https://seller.shopee.tw/portal/"
args = chrome.build_launch_args(
account,
config=cfg,
initial_url=initial_url,
)
self.assertEqual(initial_url, args[-1])
self.assertEqual(1, args.count(initial_url))
self.assertEqual(1, args.count("--no-first-run"))
self.assertEqual(1, args.count("--no-default-browser-check"))
self.assert_removed(temp_dir)
def test_build_launch_args_rejects_untrusted_initial_url(self):
account = {"alias": "alias", "debug_port": 9222}
cfg = {"chrome_path": "chrome.exe"}
for initial_url in (
"ftp://seller.shopee.tw/portal/",
"https://example.com/portal/",
"https://evilshopee.tw/portal/",
"https://user:secret@seller.shopee.tw/portal/",
"https:///portal/",
):
with self.subTest(initial_url=initial_url), self.assertRaises(
chrome.ChromeLaunchError
):
chrome.build_launch_args(
account,
config=cfg,
initial_url=initial_url,
)
def test_build_launch_args_rejects_invalid_account(self):
with self.assertRaises(chrome.ChromeLaunchError):
chrome.build_launch_args(
@@ -268,8 +309,13 @@ class ChromeTests(TempDirMixin, unittest.TestCase):
cfg = {"chrome_path": "chrome.exe", "user_data_root": temp_dir}
process = object()
initial_url = "https://seller.shopee.tw/portal/"
with mock.patch("app.chrome.subprocess.Popen", return_value=process) as popen:
result = chrome.launch_chrome(account, config=cfg)
result = chrome.launch_chrome(
account,
config=cfg,
initial_url=initial_url,
)
self.assertIs(process, result)
popen.assert_called_once()
@@ -277,6 +323,7 @@ class ChromeTests(TempDirMixin, unittest.TestCase):
self.assertEqual("chrome.exe", args[0])
self.assertIn("--remote-debugging-port=9222", args)
self.assertIn("--remote-allow-origins=*", args)
self.assertEqual(initial_url, args[-1])
self.assert_removed(temp_dir)
@@ -309,6 +356,9 @@ class ChromeTests(TempDirMixin, unittest.TestCase):
self.assertIn("--remote-allow-origins=*", script)
self.assertIn("--user-data-dir=", script)
self.assertIn("profile with space", script)
self.assertNotIn("seller.shopee", script)
self.assertNotIn("--no-first-run", script)
self.assertNotIn("--no-default-browser-check", script)
self.assert_removed(temp_dir)
tests/test_gui.py
+89 -2
View File
@@ -7231,7 +7231,9 @@ class GuiTests(TempDirMixin, unittest.TestCase):
with mock.patch("app.gui.AccountDialog", FakeDialog), mock.patch(
"app.gui.QMessageBox.warning"
) as warning:
) as warning, mock.patch(
"app.gui.accounts.launch_for_login"
) as launch_for_login:
tab.add_account()
warning.assert_called_once()
@@ -7247,6 +7249,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertNotIn("plain-password", visible_values)
self.assertIn("账号已新增", statuses[-1])
self.assertNotIn("plain-password", statuses[-1])
launch_for_login.assert_not_called()
self.assert_removed(temp_dir)
@@ -9045,6 +9048,10 @@ class GuiTests(TempDirMixin, unittest.TestCase):
"pid": 1234,
"target_id": "target-new",
"url": "https://seller.shopee.tw/portal/",
"created_tab": False,
"startup_target_reused": True,
"startup_page_navigated": False,
"page_target_count": 1,
},
) as launch_for_login:
tab.launch_login()
@@ -9061,6 +9068,10 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIn("step=launch_chrome result=launched", messages)
self.assertIn("pid=1234", messages)
self.assertIn("target_id=target-new", messages)
self.assertIn("created_tab=0", messages)
self.assertIn("startup_target_reused=1", messages)
self.assertTrue(tab.table.isEnabled())
self.assertTrue(tab.add_button.isEnabled())
self.assert_removed(temp_dir)
@@ -9083,10 +9094,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
},
) as launch_for_login:
tab.launch_login()
tab.table.selectRow(0)
tab.launch_login()
launch_for_login.assert_called_once_with(account, config=cfg)
self.assertEqual("已启动", tab.login_statuses["alias-a"])
self.assertIn("已复用现有窗口", statuses[-1])
self.assertTrue(any("已复用现有窗口" in message for message in statuses))
run_log = db.list_run_logs(limit=1, run_type="chrome_launch", path=cfg["db_path"])[0]
self.assertEqual("done", run_log.status)
self.assertEqual(1, run_log.success_count)
@@ -9096,7 +9109,81 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIn("step=launch_chrome result=reused", messages)
self.assertIn("target_id=target-existing", messages)
self.assertIn("url=https://seller.shopee.tw/portal/", messages)
self.assertIn("result=ignored", messages)
self.assert_removed(temp_dir)
def test_accounts_tab_ignores_reentrant_launch_and_restores_controls(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
statuses = []
tab = AccountsTab(config=cfg, status_callback=statuses.append)
self.addCleanup(tab.close)
tab.table.selectRow(0)
def launch_once(selected_account, config=None):
self.assertEqual(account, selected_account)
self.assertEqual(cfg, config)
self.assertFalse(tab.table.isEnabled())
self.assertFalse(tab.add_button.isEnabled())
self.assertFalse(tab.launch_button.isEnabled())
tab.launch_login()
return {
"action": "launched",
"pid": 1234,
"target_id": "target-initial",
"url": "https://seller.shopee.tw/portal/",
"created_tab": False,
"startup_target_reused": True,
"page_target_count": 1,
}
with mock.patch(
"app.gui.accounts.launch_for_login",
side_effect=launch_once,
) as launch_for_login:
tab.launch_login()
launch_for_login.assert_called_once_with(account, config=cfg)
self.assertTrue(tab.table.isEnabled())
self.assertTrue(tab.add_button.isEnabled())
run_logs = db.list_run_logs(
limit=10,
run_type="chrome_launch",
path=cfg["db_path"],
)
self.assertEqual(1, len(run_logs))
events = db.list_run_log_events(run_logs[0].id, path=cfg["db_path"])
messages = "\n".join(event.message for event in events)
self.assertIn("result=ignored", messages)
self.assertIn("reason=launch_in_progress", messages)
self.assertTrue(any("正在启动" in message for message in statuses))
self.assert_removed(temp_dir)
def test_accounts_tab_launch_failure_restores_controls(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
tab = AccountsTab(config=cfg)
self.addCleanup(tab.close)
tab.table.selectRow(0)
with mock.patch(
"app.gui.accounts.launch_for_login",
side_effect=RuntimeError("启动失败"),
), mock.patch("app.gui.QMessageBox.warning"):
tab.launch_login()
self.assertTrue(tab.table.isEnabled())
self.assertTrue(tab.add_button.isEnabled())
self.assertTrue(tab.launch_button.isEnabled())
run_log = db.list_run_logs(
limit=1,
run_type="chrome_launch",
path=cfg["db_path"],
)[0]
self.assertEqual("failed", run_log.status)
if __name__ == "__main__":
unittest.main()