chengma
82d390c25a
实现 replace_cover 满9张时先校验 old_cover_path 本地备份，备份缺失返回 OLD_COVER_BACKUP_MISSING 且不删除线上图片。 备份存在时点击第一张删除按钮并处理可见确认框，删除后继续上传新封面并拖到第一位；apply_task 会把任务 old_cover_path 传入 replace_cover。 补充 editor 单测覆盖备份缺失阻断、删除确认后上传拖首位、删除按钮缺失和 old_cover_path 传递；同步需求、架构、编码规则、任务看板、API、routes、current-state 与 progress。 已在测试商品 29671243750 上验证 9图删除和满9张替换的不提交流程；未点击更新保存线上。
455 lines
16 KiB
Python
455 lines
16 KiB
Python
import os
|
|
import json
|
|
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 cover_rects(count, prefix="old", start=0):
|
|
return [
|
|
{
|
|
"i": index,
|
|
"x": 100 + index * 10,
|
|
"y": 100,
|
|
"left": 60 + index * 10,
|
|
"top": 60,
|
|
"w": 80,
|
|
"h": 80,
|
|
"src": f"https://susercontent.com/{prefix}-{start + index}.jpg",
|
|
}
|
|
for index in range(count)
|
|
]
|
|
|
|
|
|
class FakeCoverCDP:
|
|
def __init__(self, count=9, delete_click=True, confirm_click=True):
|
|
self.initial = cover_rects(count)
|
|
self.after_delete = cover_rects(max(0, count - 1), prefix="kept", start=1)
|
|
self.new_rect = {
|
|
"i": max(0, count - 1),
|
|
"x": 300,
|
|
"y": 100,
|
|
"left": 260,
|
|
"top": 60,
|
|
"w": 80,
|
|
"h": 80,
|
|
"src": "https://susercontent.com/new-cover.jpg",
|
|
}
|
|
self.delete_click = delete_click
|
|
self.confirm_click = confirm_click
|
|
self.delete_clicked = False
|
|
self.confirm_clicked = False
|
|
self.uploaded = False
|
|
self.dragged = False
|
|
self.sent = []
|
|
self.drag_calls = []
|
|
|
|
def _rects(self):
|
|
if self.uploaded and self.dragged:
|
|
return [self.new_rect] + self.after_delete
|
|
if self.uploaded:
|
|
return self.after_delete + [self.new_rect]
|
|
if self.confirm_clicked:
|
|
return self.after_delete
|
|
return self.initial
|
|
|
|
def val(self, expr):
|
|
if expr == editor.JS_RECTS:
|
|
return json.dumps(self._rects())
|
|
if expr == editor.JS_CLICK_FIRST_DELETE:
|
|
if not self.delete_click:
|
|
return json.dumps({"clicked": False, "reason": "NO_DELETE_BUTTON"})
|
|
self.delete_clicked = True
|
|
return json.dumps({"clicked": True, "reason": None})
|
|
if expr == editor.JS_CLICK_DELETE_CONFIRM:
|
|
if not self.confirm_click:
|
|
return json.dumps({"clicked": False, "reason": "NO_CONFIRM_BUTTON"})
|
|
self.confirm_clicked = True
|
|
return json.dumps({"clicked": True, "reason": None, "text": "刪除"})
|
|
return True
|
|
|
|
def object_id(self, expr):
|
|
return "upload-input"
|
|
|
|
def send(self, method, params=None):
|
|
self.sent.append((method, params or {}))
|
|
if method == "DOM.setFileInputFiles":
|
|
self.uploaded = True
|
|
return {}
|
|
|
|
def drag(self, *args, **kwargs):
|
|
self.dragged = True
|
|
self.drag_calls.append((args, kwargs))
|
|
|
|
|
|
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()
|
|
|
|
def test_apply_task_passes_old_cover_backup_to_replace_cover(self):
|
|
cdp = FakeProductCDP("ws-new")
|
|
|
|
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
|
|
"app.editor.replace_cover",
|
|
return_value={"ok": True, "reason": None},
|
|
) as replace_cover, mock.patch(
|
|
"app.editor.click_update",
|
|
return_value={"clicked": True, "reason": None},
|
|
):
|
|
result = editor.apply_task(
|
|
{"debug_port": 9222},
|
|
{
|
|
"item_id": "51100639510",
|
|
"new_cover_path": "new.jpg",
|
|
"old_cover_path": "old.jpg",
|
|
},
|
|
)
|
|
|
|
self.assertTrue(result["committed"])
|
|
replace_cover.assert_called_once_with(
|
|
cdp,
|
|
"new.jpg",
|
|
old_cover_path="old.jpg",
|
|
)
|
|
|
|
def test_replace_cover_full_slots_requires_old_cover_backup(self):
|
|
cdp = FakeCoverCDP(count=9)
|
|
missing_backup = os.path.join(os.path.dirname(__file__), "missing-old-cover.jpg")
|
|
|
|
with mock.patch("app.editor.time.sleep"):
|
|
result = editor.replace_cover(
|
|
cdp,
|
|
__file__,
|
|
old_cover_path=missing_backup,
|
|
)
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual("OLD_COVER_BACKUP_MISSING", result["reason"])
|
|
self.assertFalse(cdp.delete_clicked)
|
|
self.assertFalse(cdp.uploaded)
|
|
|
|
def test_replace_cover_full_slots_deletes_first_then_uploads_and_drags(self):
|
|
cdp = FakeCoverCDP(count=9)
|
|
|
|
with mock.patch("app.editor.time.sleep"):
|
|
result = editor.replace_cover(
|
|
cdp,
|
|
__file__,
|
|
old_cover_path=__file__,
|
|
timeout=1,
|
|
)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(cdp.delete_clicked)
|
|
self.assertTrue(cdp.confirm_clicked)
|
|
self.assertTrue(cdp.uploaded)
|
|
self.assertTrue(cdp.dragged)
|
|
self.assertEqual(9, result["count_before"])
|
|
self.assertEqual(9, result["count_after"])
|
|
self.assertEqual(8, result["delete"]["count_after"])
|
|
self.assertEqual("https://susercontent.com/new-cover.jpg", result["new_src"])
|
|
|
|
def test_replace_cover_full_slots_reports_missing_delete_button(self):
|
|
cdp = FakeCoverCDP(count=9, delete_click=False)
|
|
|
|
with mock.patch("app.editor.time.sleep"):
|
|
result = editor.replace_cover(
|
|
cdp,
|
|
__file__,
|
|
old_cover_path=__file__,
|
|
timeout=1,
|
|
)
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual("NO_DELETE_BUTTON", result["reason"])
|
|
self.assertFalse(cdp.uploaded)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|