chengma
b05a856796
新增 shopee_update 配置段，默认关闭真实提交和封面更新，限制测试商品ID、单次最大更新条数，并支持成功后关闭本轮自动新开编辑页。 Tab⑤ 设置页新增 Shopee 更新安全表单；Tab③ 开始更新前先读取安全配置并阻断未授权真实提交、超量、非测试商品和未允许的封面更新。 ApplyWorker 将 close_success_tab 传入 editor.apply_task；editor 仅在提交成功且页面为本轮自动新开时关闭 tab，失败和复用页面保留。 同步架构、API、路由、任务看板、当前状态与 progress 文档；补充 GUI 与 editor 单测覆盖安全设置保存、共享配置读取、安全拦截和 tab 关闭策略。
299 lines
10 KiB
Python
299 lines
10 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from _helpers import REPO_ROOT
|
|
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from app import editor
|
|
|
|
|
|
class FakeCDP:
|
|
instances = []
|
|
|
|
def __init__(self, ws, url="", cookies=None):
|
|
self.ws = ws
|
|
self.url = url
|
|
self.cookies = cookies or []
|
|
self.closed = False
|
|
FakeCDP.instances.append(self)
|
|
|
|
def val(self, expr):
|
|
if expr == "location.href":
|
|
return self.url
|
|
return None
|
|
|
|
def send(self, method, params=None):
|
|
if method == "Network.getAllCookies":
|
|
return {"cookies": self.cookies}
|
|
return {}
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class FakeProductCDP:
|
|
def __init__(self, ws):
|
|
self.ws = ws
|
|
self.closed = False
|
|
self.sent = []
|
|
|
|
def send(self, method, params=None):
|
|
self.sent.append((method, params or {}))
|
|
return {}
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def cookie(name, domain=".shopee.tw"):
|
|
return {"name": name, "domain": domain}
|
|
|
|
|
|
class EditorLoginTests(unittest.TestCase):
|
|
def setUp(self):
|
|
FakeCDP.instances = []
|
|
|
|
def patch_cdp(self, url="", cookies=None):
|
|
def factory(ws):
|
|
return FakeCDP(ws, url=url, cookies=cookies)
|
|
|
|
return mock.patch("app.editor.CDP", side_effect=factory)
|
|
|
|
def test_login_status_false_for_login_page_url(self):
|
|
with mock.patch(
|
|
"app.editor.http_get",
|
|
return_value=[
|
|
{
|
|
"type": "page",
|
|
"url": "https://seller.shopee.tw/account/signin",
|
|
"webSocketDebuggerUrl": "ws-login",
|
|
}
|
|
],
|
|
):
|
|
status = editor.login_status({"debug_port": 9222}, timeout=0)
|
|
|
|
self.assertFalse(status["logged_in"])
|
|
self.assertEqual("LOGIN_PAGE", status["reason"])
|
|
self.assertEqual([], FakeCDP.instances)
|
|
|
|
def test_login_status_true_with_session_cookie(self):
|
|
with mock.patch(
|
|
"app.editor.http_get",
|
|
return_value=[
|
|
{
|
|
"type": "page",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-portal",
|
|
}
|
|
],
|
|
), self.patch_cdp(
|
|
url="https://seller.shopee.tw/portal/",
|
|
cookies=[cookie("SPC_ST"), cookie("csrftoken")],
|
|
):
|
|
status = editor.login_status({"debug_port": 9222}, timeout=1)
|
|
|
|
self.assertTrue(status["logged_in"])
|
|
self.assertIsNone(status["reason"])
|
|
self.assertIn("SPC_ST", status["cookie_names"])
|
|
self.assertTrue(FakeCDP.instances[0].closed)
|
|
|
|
def test_login_status_false_without_session_cookie(self):
|
|
with mock.patch(
|
|
"app.editor.http_get",
|
|
return_value=[
|
|
{
|
|
"type": "page",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"webSocketDebuggerUrl": "ws-portal",
|
|
}
|
|
],
|
|
), self.patch_cdp(
|
|
url="https://seller.shopee.tw/portal/",
|
|
cookies=[cookie("csrftoken")],
|
|
):
|
|
status = editor.login_status({"debug_port": 9222}, timeout=0)
|
|
|
|
self.assertFalse(status["logged_in"])
|
|
self.assertEqual("NO_SESSION_COOKIE", status["reason"])
|
|
|
|
def test_login_status_opens_seller_home_when_no_shopee_page(self):
|
|
with mock.patch("app.editor.http_get", return_value=[]), mock.patch(
|
|
"app.editor.create_tab", return_value="ws-new"
|
|
) as create_tab, self.patch_cdp(
|
|
url="https://seller.shopee.tw/",
|
|
cookies=[cookie("SPC_U")],
|
|
):
|
|
status = editor.login_status(
|
|
{"debug_port": 9222, "region_host": "seller.shopee.tw"},
|
|
timeout=1,
|
|
)
|
|
|
|
self.assertTrue(status["logged_in"])
|
|
create_tab.assert_called_once_with(
|
|
"https://seller.shopee.tw/",
|
|
host="127.0.0.1:9222",
|
|
)
|
|
|
|
def test_is_logged_in_returns_boolean(self):
|
|
with mock.patch(
|
|
"app.editor.login_status",
|
|
return_value={"logged_in": True},
|
|
):
|
|
self.assertTrue(editor.is_logged_in({"debug_port": 9222}))
|
|
|
|
def test_open_product_marks_reused_existing_tab(self):
|
|
fake = FakeProductCDP("ws-existing")
|
|
with mock.patch(
|
|
"app.editor.find_product_tab",
|
|
return_value={"id": "target-existing", "webSocketDebuggerUrl": "ws-existing"},
|
|
) as find_product_tab, mock.patch(
|
|
"app.editor.create_tab_info",
|
|
) as create_tab_info, mock.patch(
|
|
"app.editor.CDP",
|
|
return_value=fake,
|
|
), mock.patch("app.editor._ensure_page_domains"), mock.patch("app.editor._wait_ready"):
|
|
cdp = editor.open_product({"debug_port": 9222}, "51100639510")
|
|
|
|
self.assertIs(fake, cdp)
|
|
self.assertEqual("target-existing", cdp.target_id)
|
|
self.assertFalse(cdp.created_by_app)
|
|
self.assertEqual("127.0.0.1:9222", cdp.cdp_host)
|
|
find_product_tab.assert_called_once_with("51100639510", host="127.0.0.1:9222")
|
|
create_tab_info.assert_not_called()
|
|
self.assertIn(
|
|
(
|
|
"Page.navigate",
|
|
{
|
|
"url": (
|
|
"https://seller.shopee.tw/portal/product/51100639510"
|
|
"?pageEntry=product_list&ignore-html-cache=1"
|
|
)
|
|
},
|
|
),
|
|
fake.sent,
|
|
)
|
|
|
|
def test_open_product_marks_auto_created_tab(self):
|
|
fake = FakeProductCDP("ws-new")
|
|
with mock.patch("app.editor.find_product_tab", return_value=None), mock.patch(
|
|
"app.editor.create_tab_info",
|
|
return_value={"id": "target-new", "webSocketDebuggerUrl": "ws-new"},
|
|
) as create_tab_info, mock.patch(
|
|
"app.editor.CDP",
|
|
return_value=fake,
|
|
), mock.patch("app.editor._ensure_page_domains"), mock.patch("app.editor._wait_ready"):
|
|
cdp = editor.open_product({"debug_port": 9223}, "51100639511")
|
|
|
|
self.assertEqual("target-new", cdp.target_id)
|
|
self.assertTrue(cdp.created_by_app)
|
|
self.assertEqual("127.0.0.1:9223", cdp.cdp_host)
|
|
create_tab_info.assert_called_once()
|
|
|
|
def test_collect_closes_only_auto_created_product_tab(self):
|
|
cdp = FakeProductCDP("ws-new")
|
|
cdp.target_id = "target-new"
|
|
cdp.created_by_app = True
|
|
cdp.cdp_host = "127.0.0.1:9222"
|
|
|
|
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
|
|
"app.editor.read_title",
|
|
return_value="旧标题",
|
|
), mock.patch(
|
|
"app.editor.read_cover_src",
|
|
return_value="https://down-ws-sg.vod.susercontent.com/cover.jpg",
|
|
), mock.patch(
|
|
"app.editor.download_cover",
|
|
return_value="images/main/51100639510_old.jpg",
|
|
), mock.patch("app.editor.close_tab", return_value=True) as close_tab:
|
|
result = editor.collect(
|
|
{"debug_port": 9222},
|
|
{"item_id": "51100639510", "old_cover_path": "old.jpg"},
|
|
)
|
|
|
|
self.assertEqual("旧标题", result["old_title"])
|
|
self.assertTrue(cdp.closed)
|
|
close_tab.assert_called_once_with("target-new", host="127.0.0.1:9222")
|
|
|
|
def test_collect_keeps_reused_product_tab_open(self):
|
|
cdp = FakeProductCDP("ws-existing")
|
|
cdp.target_id = "target-existing"
|
|
cdp.created_by_app = False
|
|
cdp.cdp_host = "127.0.0.1:9222"
|
|
|
|
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
|
|
"app.editor.read_title",
|
|
return_value="旧标题",
|
|
), mock.patch(
|
|
"app.editor.read_cover_src",
|
|
return_value="https://down-ws-sg.vod.susercontent.com/cover.jpg",
|
|
), mock.patch(
|
|
"app.editor.download_cover",
|
|
return_value="images/main/51100639510_old.jpg",
|
|
), mock.patch("app.editor.close_tab") as close_tab:
|
|
editor.collect(
|
|
{"debug_port": 9222},
|
|
{"item_id": "51100639510", "old_cover_path": "old.jpg"},
|
|
)
|
|
|
|
self.assertTrue(cdp.closed)
|
|
close_tab.assert_not_called()
|
|
|
|
def test_apply_task_closes_auto_created_product_tab_after_success_when_enabled(self):
|
|
cdp = FakeProductCDP("ws-new")
|
|
cdp.target_id = "target-new"
|
|
cdp.created_by_app = True
|
|
cdp.cdp_host = "127.0.0.1:9222"
|
|
|
|
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
|
|
"app.editor.change_title",
|
|
return_value={"ok": True, "reason": None},
|
|
) as change_title, mock.patch("app.editor.replace_cover") as replace_cover, mock.patch(
|
|
"app.editor.click_update",
|
|
return_value={"clicked": True, "reason": None},
|
|
), mock.patch("app.editor.close_tab", return_value=True) as close_tab:
|
|
result = editor.apply_task(
|
|
{"debug_port": 9222},
|
|
{"item_id": "51100639510", "new_title": "新标题"},
|
|
close_success_tab=True,
|
|
)
|
|
|
|
self.assertTrue(result["committed"])
|
|
change_title.assert_called_once_with(cdp, "新标题")
|
|
replace_cover.assert_not_called()
|
|
self.assertTrue(cdp.closed)
|
|
close_tab.assert_called_once_with("target-new", host="127.0.0.1:9222")
|
|
|
|
def test_apply_task_keeps_product_tab_after_failed_update(self):
|
|
cdp = FakeProductCDP("ws-new")
|
|
cdp.target_id = "target-new"
|
|
cdp.created_by_app = True
|
|
cdp.cdp_host = "127.0.0.1:9222"
|
|
|
|
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
|
|
"app.editor.change_title",
|
|
return_value={"ok": True, "reason": None},
|
|
), mock.patch(
|
|
"app.editor.click_update",
|
|
return_value={"clicked": False, "reason": "UPDATE_DISABLED"},
|
|
), mock.patch("app.editor.close_tab") as close_tab:
|
|
result = editor.apply_task(
|
|
{"debug_port": 9222},
|
|
{"item_id": "51100639510", "new_title": "新标题"},
|
|
close_success_tab=True,
|
|
)
|
|
|
|
self.assertFalse(result["committed"])
|
|
self.assertEqual("UPDATE_DISABLED", result["error"])
|
|
self.assertTrue(cdp.closed)
|
|
close_tab.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|