Files
crawler-plugin/app/tests/test_patrol_delete.py
2026-04-30 21:40:35 +08:00

1107 lines
45 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from amazon.patrol_delete import InventoryManage, PatrolDeleteTask
def test_patrol_delete_task_post_result_builds_patrol_delete_payload(monkeypatch):
captured = {}
class Response:
status_code = 200
text = '{"success":true}'
def json(self):
return {"success": True}
def fake_post(url, **kwargs):
captured["url"] = url
captured["kwargs"] = kwargs
return Response()
monkeypatch.setattr("config.DELETE_BRAND_API_BASE", "http://java.example")
monkeypatch.setattr("amazon.patrol_delete.requests.post", fake_post)
country_sections = [
{
"country": "德国",
"rows": [{"status": "正常", "quantity": "12", "deleteQuantity": "3", "processStatus": "处理中"}],
}
]
cart_ratios = [{"country": "德国", "ratio": "25%"}]
PatrolDeleteTask().post_result(
task_id=3089,
shop_name="郭亚芳",
country_sections=country_sections,
cart_ratios=cart_ratios,
shop_done=True,
)
assert captured["url"] == "http://java.example/api/patrol-delete/tasks/3089/result"
assert captured["kwargs"]["headers"] == {"Content-Type": "application/json"}
assert captured["kwargs"]["timeout"] == 30
assert captured["kwargs"]["verify"] is False
assert captured["kwargs"]["json"]["shops"] == [
{
"shopName": "郭亚芳",
"shopDone": True,
"submissionId": captured["kwargs"]["json"]["shops"][0]["submissionId"],
"chunkIndex": 1,
"chunkTotal": 1,
"countrySections": country_sections,
"cartRatios": cart_ratios,
}
]
assert captured["kwargs"]["json"]["shops"][0]["submissionId"].startswith("patrol-delete:3089:郭亚芳:")
def test_patrol_delete_task_post_result_rejects_empty_success_payload(monkeypatch):
calls = []
def fake_post(url, **kwargs):
calls.append((url, kwargs))
raise AssertionError("empty success payload must not be posted")
monkeypatch.setattr("amazon.patrol_delete.requests.post", fake_post)
try:
PatrolDeleteTask().post_result(task_id=3089, shop_name="郭亚芳", country_sections=[], cart_ratios=[])
except RuntimeError as exc:
assert "缺少国家状态数据" in str(exc)
else:
raise AssertionError("expected empty success payload to fail")
assert calls == []
def test_patrol_delete_task_post_result_retries_business_failure(monkeypatch):
calls = []
class Response:
status_code = 200
text = '{"success":false,"message":"bad payload"}'
def json(self):
return {"success": False, "message": "bad payload"}
def fake_post(url, **kwargs):
calls.append((url, kwargs))
return Response()
monkeypatch.setattr("config.DELETE_BRAND_API_BASE", "http://java.example")
monkeypatch.setattr("amazon.patrol_delete.requests.post", fake_post)
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: None)
try:
PatrolDeleteTask().post_result(
task_id=3089,
shop_name="郭亚芳",
country_sections=[{"country": "德国", "rows": [{"status": "全部", "quantity": "1"}]}],
cart_ratios=[{"country": "德国", "ratio": ""}],
)
except RuntimeError as exc:
assert "最终失败" in str(exc)
else:
raise AssertionError("expected Java business failure to fail")
assert len(calls) == 3
def test_normalize_country_sections_keeps_only_status_and_quantity_for_complete_draft_rows():
rows = PatrolDeleteTask._normalize_country_sections_for_result(
[
{
"country": "意大利",
"rows": [
{
"status": "商品信息草稿",
"quantity": "12",
"deleteQuantity": "3",
"processStatus": "已完成",
},
{
"status": "缺少的信息",
"quantity": "1",
"deleteQuantity": "1",
"processStatus": "已完成",
},
],
}
]
)
assert rows == [
{
"country": "意大利",
"rows": [
{"status": "商品信息草稿", "quantity": "12"},
{"status": "缺少的信息", "quantity": "1", "deleteQuantity": "1", "processStatus": "已完成"},
],
}
]
def test_replace_complete_draft_rows_uses_quick_view_rows():
country_section = {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "无需删除", "processStatus": "无可删商品"},
{"status": "商品信息草稿", "quantity": "", "deleteQuantity": "0", "processStatus": ""},
],
}
complete_draft_section = {
"country": "西班牙",
"rows": [
{"type": "completeDraft", "status": "未提交的草稿", "quantity": "1"},
{"type": "completeDraft", "status": "已提交:提供缺少的信息", "quantity": "0"},
],
}
merged = PatrolDeleteTask._replace_complete_draft_rows(country_section, complete_draft_section)
assert merged == {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "无需删除", "processStatus": "无可删商品"},
{"type": "completeDraft", "status": "未提交的草稿", "quantity": "1"},
{"type": "completeDraft", "status": "已提交:提供缺少的信息", "quantity": "0"},
],
}
assert PatrolDeleteTask._normalize_country_sections_for_result([merged]) == [
{
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "无需删除", "processStatus": "无可删商品"},
{"status": "未提交的草稿", "quantity": "1"},
{"status": "已提交:提供缺少的信息", "quantity": "0"},
],
}
]
def test_patrol_delete_task_process_task_aggregates_country_results(monkeypatch):
from config import runing_shop, runing_task
runing_task.clear()
runing_shop.clear()
calls = []
class Driver:
def close_store(self):
return None
def fake_post_result(self, **kwargs):
calls.append(kwargs)
def fake_open_shop(self, **kwargs):
return Driver()
def fake_process_country(self, driver, task_id, shop_name, country_name):
if country_name == "德国":
return {
"success": True,
"countrySection": {
"country": "德国",
"rows": [
{
"status": "商品信息草稿",
"quantity": "0",
"deleteQuantity": "2",
"processStatus": "已完成",
}
],
},
"cartRatio": {"country": "德国", "ratio": "25%"},
"deletedCount": 2,
"reopenRequired": False,
}
return {
"success": False,
"countrySection": {
"country": "英国",
"rows": [
{
"status": "全部",
"quantity": "",
"deleteQuantity": "",
"processStatus": "处理失败: 切换国家失败",
}
],
},
"cartRatio": {"country": "英国", "ratio": ""},
"deletedCount": 0,
"reopenRequired": True,
}
monkeypatch.setattr(PatrolDeleteTask, "open_shop", fake_open_shop)
monkeypatch.setattr(PatrolDeleteTask, "process_country", fake_process_country)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
country_sections = [
{"country": "德国", "rows": [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}]},
{"country": "英国", "rows": [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}]},
]
cart_ratios = [{"country": "德国", "ratio": ""}, {"country": "英国", "ratio": ""}]
PatrolDeleteTask().process_task(
{
"type": "patrol-delete-run",
"data": {
"taskId": 1,
"items": [{"shopName": "郭亚芳", "companyName": "rongchuang123"}],
"country_sections": country_sections,
"cart_ratios": cart_ratios,
},
}
)
assert runing_task[1]["status"] == "completed"
assert runing_task[1]["processed_shops"] == 1
assert runing_task[1]["processed_countries"] == 2
assert runing_task[1]["success_count"] == 1
assert runing_task[1]["failed_count"] == 0
assert runing_task[1]["failed_countries"] == 1
assert runing_task[1]["deleted_listings_count"] == 2
assert "郭亚芳" not in runing_shop
assert calls == [
{
"task_id": 1,
"shop_name": "郭亚芳",
"country_sections": [
{
"country": "德国",
"rows": [
{
"status": "商品信息草稿",
"quantity": "0",
"deleteQuantity": "2",
"processStatus": "已完成",
}
],
},
{
"country": "英国",
"rows": [
{
"status": "全部",
"quantity": "",
"deleteQuantity": "",
"processStatus": "处理失败: 切换国家失败",
}
],
},
],
"cart_ratios": [{"country": "德国", "ratio": "25%"}, {"country": "英国", "ratio": ""}],
"shop_done": True,
}
]
def test_patrol_delete_task_process_task_skips_missing_required_data(monkeypatch):
calls = []
def fake_post_result(self, **kwargs):
calls.append(kwargs)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
PatrolDeleteTask().process_task({"type": "patrol-delete-run", "data": {"items": [{"shopName": "郭亚芳"}]}})
PatrolDeleteTask().process_task({"type": "patrol-delete-run", "data": {"taskId": 1, "items": []}})
assert calls == []
def test_patrol_delete_task_reports_shop_error(monkeypatch):
from config import runing_shop, runing_task
runing_task.clear()
runing_shop.clear()
calls = []
def fake_post_result(self, **kwargs):
calls.append(kwargs)
def fake_open_shop(self, **kwargs):
return None
monkeypatch.setattr(PatrolDeleteTask, "open_shop", fake_open_shop)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
PatrolDeleteTask().process_task(
{
"type": "patrol-delete-run",
"data": {
"taskId": 2,
"items": [{"shopName": "郭亚芳", "companyName": "rongchuang123"}],
"country_sections": [{"country": "德国", "rows": [{"status": "全部"}]}],
},
}
)
assert runing_task[2]["status"] == "completed"
assert runing_task[2]["processed_shops"] == 1
assert runing_task[2]["success_count"] == 0
assert runing_task[2]["failed_count"] == 1
assert "郭亚芳" not in runing_shop
assert calls == [{"task_id": 2, "shop_name": "郭亚芳", "error": "店铺 郭亚芳 打开失败", "shop_done": True}]
def test_delete_all_listings_from_non_whitelisted_statuses_deletes_each_row_in_status(monkeypatch):
driver = InventoryManage({})
status_rows = [
[
{"status": "商品信息草稿", "quantity": "2", "deleteQuantity": "", "processStatus": ""},
{"status": "全部", "quantity": "9", "deleteQuantity": "", "processStatus": ""},
],
[
{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""},
],
]
state = {"index": 0, "selected": [], "deleted": 0}
def fake_read_listing_status_rows_for_delete(shop_name, country, **kwargs):
current = status_rows[min(state["index"], len(status_rows) - 1)]
state["index"] += 1
return current
def fake_select_listing_status(status, **kwargs):
state["selected"].append(status)
def fake_delete_selected_filtered_inventory_rows(status, **kwargs):
state["deleted"] += 2
return {
"deletedCount": 2,
"target": {"sku": "bulk-selected", "rawText": "bulk-selected"},
"successMessage": f"{status} 删除成功",
}
monkeypatch.setattr(driver, "_read_listing_status_rows_for_delete", fake_read_listing_status_rows_for_delete)
monkeypatch.setattr(driver, "select_listing_status", fake_select_listing_status)
monkeypatch.setattr(driver, "_delete_selected_filtered_inventory_rows", fake_delete_selected_filtered_inventory_rows)
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: None)
result = driver.delete_all_listings_from_non_whitelisted_statuses("郭亚芳", "德国")
assert state["selected"] == ["商品信息草稿"]
assert state["deleted"] == 2
assert result["deleteCounts"] == {"商品信息草稿": 2}
assert result["totalDeleted"] == 2
assert result["stopReason"] == "processed-candidates"
assert result["statusRows"] == [{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""}]
assert [item["target"]["sku"] for item in result["results"]] == ["bulk-selected"]
def test_select_listing_status_uses_dropdown_option_not_visible_status(monkeypatch):
driver = InventoryManage({})
calls = []
monkeypatch.setattr(driver, "open_listing_status_dropdown", lambda: calls.append("open") or {"ok": True})
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: None)
monkeypatch.setattr(driver, "_run_js", lambda script, *args: calls.append((script, args)) or {"ok": True})
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: None)
result = driver.select_listing_status("商品信息草稿")
assert result["mode"] == "dropdown-option"
assert calls[0] == "open"
assert calls[-1][1] == ({"status": "商品信息草稿"},)
def test_get_listing_status_dropdown_options_closes_dropdown_after_read(monkeypatch):
driver = InventoryManage({})
calls = []
monkeypatch.setattr(driver, "open_listing_status_dropdown", lambda: calls.append("open") or {"ok": True})
monkeypatch.setattr(
driver,
"_wait_listing_status_option_rows",
lambda: calls.append("read") or [{"raw_text": "详情页面已删除 (1)", "value": "DetailPageRemoved"}],
)
monkeypatch.setattr(driver, "_close_listing_status_dropdown", lambda: calls.append("close"))
rows = driver.get_listing_status_dropdown_options("郭亚芳", "德国")
assert calls == ["open", "read", "close"]
assert rows[0]["status"] == "详情页面已删除"
def test_open_listing_status_dropdown_waits_until_component_enabled(monkeypatch):
driver = InventoryManage({})
calls = []
class Wait:
def doc_loaded(self, raise_err=False):
calls.append("doc_loaded")
class Tab:
wait = Wait()
responses = [
{"ok": False, "reason": "listing status dropdown disabled/loading"},
{"ok": True, "selected": {"text": "全部在售不可售商品信息草稿"}},
]
driver.tab = Tab()
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: calls.append(("loader", timeout)))
monkeypatch.setattr(driver, "_run_js", lambda script: calls.append("js") or responses.pop(0))
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: calls.append(("sleep", seconds)))
result = driver.open_listing_status_dropdown()
assert result["ok"] is True
assert calls.count("js") == 2
def test_extract_country_cart_ratio_prefers_ratio2daysago_only():
ratio = PatrolDeleteTask._extract_country_cart_ratio(
[
{"country": "德国", "ratio2DaysAgo": "25%", "ratio30DaysAgo": "19%"},
{"country": "英国", "ratio30DaysAgo": "40%"},
],
"德国",
)
missing_ratio2 = PatrolDeleteTask._extract_country_cart_ratio(
[{"country": "英国", "ratio30DaysAgo": "40%"}],
"英国",
)
assert ratio == {"country": "德国", "ratio": "25%"}
assert missing_ratio2 == {"country": "英国", "ratio": ""}
def test_parse_status_option_rows_keeps_value_and_raw_text():
rows = InventoryManage._parse_status_option_rows(
[
{"raw_text": "在售 (1,234)", "value": "Active"},
{"raw_text": "禁止显示", "value": "SearchSuppressed"},
],
"郭亚芳",
"德国",
)
assert rows == [
{
"shop_name": "郭亚芳",
"country": "德国",
"status": "在售",
"quantity": 1234,
"value": "Active",
"raw_text": "在售 (1,234)",
},
{
"shop_name": "郭亚芳",
"country": "德国",
"status": "禁止显示",
"quantity": None,
"value": "SearchSuppressed",
"raw_text": "禁止显示",
},
]
def test_parse_status_text_accepts_chinese_parentheses():
assert InventoryManage._parse_status_text("不可售42") == ("不可售", 42)
def test_parse_status_text_preserves_status_without_quantity():
assert InventoryManage._parse_status_text("非在售") == ("非在售", None)
def test_parse_listing_status_section_rows_extracts_rendered_status_tabs():
rows = InventoryManage._parse_listing_status_section_rows(
[
{
"raw_text": (
"全部(6182) 在售(4097) 在搜索结果中禁止显示(2) 不可售(2083) "
"详情页面已删除(2) 定价问题(1) 需要批准(2080) 商品信息草稿 缺少的信息(1)"
)
}
]
)
assert rows == [
{"status": "全部", "quantity": "6182", "raw_text": "全部(6182)"},
{"status": "在售", "quantity": "4097", "raw_text": "在售(4097)"},
{"status": "在搜索结果中禁止显示", "quantity": "2", "raw_text": "在搜索结果中禁止显示(2)"},
{"status": "不可售", "quantity": "2083", "raw_text": "不可售(2083)"},
{"status": "详情页面已删除", "quantity": "2", "raw_text": "详情页面已删除(2)"},
{"status": "定价问题", "quantity": "1", "raw_text": "定价问题(1)"},
{"status": "需要批准", "quantity": "2080", "raw_text": "需要批准(2080)"},
{"status": "商品信息草稿", "quantity": "", "raw_text": "商品信息草稿"},
{"status": "缺少的信息", "quantity": "1", "raw_text": "缺少的信息(1)"},
]
def test_parse_listing_status_section_rows_supports_order_page_deleted_alias():
rows = InventoryManage._parse_listing_status_section_rows(
[
{
"raw_text": "不可售(2083) 订单页面已删除(2) 定价问题(1) 需要批准(2080)"
}
]
)
assert rows == [
{"status": "不可售", "quantity": "2083", "raw_text": "不可售(2083)"},
{"status": "订单页面已删除", "quantity": "2", "raw_text": "订单页面已删除(2)"},
{"status": "定价问题", "quantity": "1", "raw_text": "定价问题(1)"},
{"status": "需要批准", "quantity": "2080", "raw_text": "需要批准(2080)"},
]
def test_build_country_section_rows_adds_required_placeholder_fields():
rows = InventoryManage._build_country_section_rows(
[
{"status": "正常", "quantity": "12"},
{"status": "需要批准", "quantity": "2080"},
]
)
assert rows == [
{"status": "正常", "quantity": "12", "deleteQuantity": "", "processStatus": ""},
{"status": "需要批准", "quantity": "2080", "deleteQuantity": "", "processStatus": ""},
]
def test_parse_listing_status_section_rows_supports_latest_status_labels():
rows = InventoryManage._parse_listing_status_section_rows(
[
{
"raw_text": (
"全部(10) 在售(4) 配送问题(1) 缺少报价(2) 已停售(3) "
"在搜索结果中禁止显示(0) 需要批准(0)"
)
}
]
)
assert rows == [
{"status": "全部", "quantity": "10", "raw_text": "全部(10)"},
{"status": "在售", "quantity": "4", "raw_text": "在售(4)"},
{"status": "配送问题", "quantity": "1", "raw_text": "配送问题(1)"},
{"status": "缺少报价", "quantity": "2", "raw_text": "缺少报价(2)"},
{"status": "已停售", "quantity": "3", "raw_text": "已停售(3)"},
{"status": "在搜索结果中禁止显示", "quantity": "0", "raw_text": "在搜索结果中禁止显示(0)"},
{"status": "需要批准", "quantity": "0", "raw_text": "需要批准(0)"},
]
def test_build_deletable_status_rows_skips_whitelisted_statuses():
rows = [
{"status": "全部", "quantity": "12"},
{"status": "定价问题", "quantity": "2"},
{"status": "需要批准", "quantity": "3"},
{"status": "在搜索结果中禁止显示", "quantity": "1"},
{"status": "配送问题", "quantity": "5"},
{"status": "缺少报价", "quantity": "4"},
{"status": "已停售", "quantity": "6"},
{"status": "不可售", "quantity": "8"},
{"status": "在售", "quantity": "7"},
]
assert InventoryManage._build_deletable_status_rows(rows) == []
def test_build_deletable_status_rows_keeps_non_whitelisted_statuses():
rows = [
{"status": "商品信息草稿", "quantity": "2"},
{"status": "自定义异常状态", "quantity": "5"},
{"status": "全部", "quantity": "9"},
]
assert InventoryManage._build_deletable_status_rows(rows) == [
{"status": "商品信息草稿", "canonicalStatus": "商品信息草稿", "quantity": 2},
{"status": "自定义异常状态", "canonicalStatus": "自定义异常状态", "quantity": 5},
]
def test_build_deletable_status_rows_skips_group_labels_without_quantity():
rows = [
{"status": "商品信息草稿", "quantity": ""},
{"status": "详情页面已删除", "quantity": None},
{"status": "自定义异常状态", "quantity": "0"},
{"status": "缺少的信息", "quantity": "1"},
{"status": "全部", "quantity": ""},
]
assert InventoryManage._build_deletable_status_rows(rows) == [
{"status": "缺少的信息", "canonicalStatus": "缺少的信息", "quantity": 1},
]
def test_build_deletable_status_rows_compatibility_map_prevents_whitelist_misdelete():
rows = [
{"status": "搜尋結果中禁止顯示", "quantity": "4"},
{"status": "在搜索中禁止显示结果", "quantity": "1"},
{"status": "配送問題", "quantity": "2"},
{"status": "缺少報價", "quantity": "3"},
]
assert InventoryManage._build_deletable_status_rows(rows) == []
def test_build_deletable_status_rows_keeps_legacy_non_whitelist_statuses_deletable():
rows = [
{"status": "详情页面已删除", "quantity": "1"},
{"status": "缺少的信息", "quantity": "2"},
{"status": "订单页面已删除", "quantity": "3"},
]
assert InventoryManage._build_deletable_status_rows(rows) == [
{"status": "详情页面已删除", "canonicalStatus": "详情页面已删除", "quantity": 1},
{"status": "缺少的信息", "canonicalStatus": "缺少的信息", "quantity": 2},
{"status": "订单页面已删除", "canonicalStatus": "详情页面已删除", "quantity": 3},
]
def test_is_delete_success_message_accepts_both_copy_variants():
assert InventoryManage._is_delete_success_message("1 个商品已经删除。所作更改需要15分钟才会显示在商品详情页面上")
assert InventoryManage._is_delete_success_message("所做更改需要 15 分钟才会显示在商品详情页面上。 1 个商品已删除。")
def test_is_delete_success_message_accepts_deleted_count_without_delay_notice():
assert InventoryManage._is_delete_success_message("1 个商品已删除。")
assert InventoryManage._is_delete_success_message("已删除 2 个商品。")
assert not InventoryManage._is_delete_success_message("0 个商品已删除。")
def test_is_delete_success_message_accepts_deleted_without_exact_count():
assert InventoryManage._is_delete_success_message("一个或多个商品已删除。")
assert InventoryManage._is_delete_success_message("一个或多个商品已经删除")
assert InventoryManage._extract_delete_success_count("一个或多个商品已删除。") == 0
def test_extract_delete_success_count_accepts_success_message():
assert InventoryManage._extract_delete_success_count("所做更改需要 15 分钟才会显示在商品详情页面上。 3 个商品已删除。") == 3
def test_short_error_simplifies_delete_success_message_timeout():
message = (
"状态[详情页面已删除]删除后未读取到成功提示: "
"['所做更改需要 15 分钟才会显示在商品详情页面上。 一个或多个商品已删除', "
"'莆田市城厢区从奇偶贸易...']"
)
assert PatrolDeleteTask._short_error(message) == "删除后未读取到成功提示"
def test_short_error_simplifies_common_patrol_delete_failures():
examples = {
"检测到任务 12 的暂停请求,停止处理": "任务已中断",
"用户取消任务": "任务已中断",
"与页面的连接已断开: target frame detached": "浏览器连接中断",
"切换国家 德国 失败: element not found": "切换国家失败",
"未找到可用商品状态下拉框: {'ok': False, 'text': '...'}": "未找到商品状态下拉框",
"商品状态下拉框已打开但未读取到选项: {'ok': True}": "未读取到商品状态下拉选项",
"未找到商品状态选项[详情页面已删除]: {'rows': ['...']}": "未找到商品状态选项",
"状态[详情页面已删除]全选商品失败: {'ok': False, 'reason': '...'}": "全选商品失败",
"状态[详情页面已删除]点击批量删除失败: {'menuOpened': False, 'html': '...'}": "点击批量删除失败",
"状态[详情页面已删除]未出现删除确认弹窗": "未出现删除确认弹窗",
"当前页面对象不支持执行 JavaScript无法探测 Shadow DOM": "当前页面无法执行 JavaScript",
}
for message, expected in examples.items():
assert PatrolDeleteTask._short_error(message) == expected
def test_build_country_section_result_uses_simplified_delete_timeout_failure_reason():
section = PatrolDeleteTask._build_country_section_result(
country_name="英国",
status_rows=[{"status": "详情页面已删除", "quantity": "1"}],
delete_counts={},
default_process_status=f"处理失败: {PatrolDeleteTask._short_error('状态[详情页面已删除]删除后未读取到成功提示: []')}",
)
assert section["rows"][0]["processStatus"] == "处理失败: 删除后未读取到成功提示"
def test_build_country_section_result_sets_zero_delete_quantity_when_no_deletable_products():
section = PatrolDeleteTask._build_country_section_result(
country_name="德国",
status_rows=[{"status": "全部", "quantity": "12"}],
delete_counts={},
default_process_status="无可删商品",
)
assert section == {
"country": "德国",
"rows": [{"status": "全部", "quantity": "12", "deleteQuantity": "无需删除", "processStatus": "无可删商品"}],
}
def test_build_country_section_result_sets_zero_delete_quantity_for_empty_no_deletable_result():
section = PatrolDeleteTask._build_country_section_result(
country_name="德国",
status_rows=[],
delete_counts={},
default_process_status="无可删商品",
)
assert section == {
"country": "德国",
"rows": [{"status": "全部", "quantity": "", "deleteQuantity": "无需删除", "processStatus": "无可删商品"}],
}
def test_build_country_section_result_marks_all_whitelisted_statuses_as_no_delete_needed():
whitelist_statuses = [
"全部",
"定价问题",
"需要批准",
"在搜索结果中禁止显示",
"配送问题",
"缺少报价",
"已停售",
"不可售",
"在售",
]
section = PatrolDeleteTask._build_country_section_result(
country_name="爱尔兰",
status_rows=[{"status": status, "quantity": "1"} for status in whitelist_statuses]
+ [{"status": "缺少的信息", "quantity": "2"}],
delete_counts={"缺少的信息": 2},
)
rows_by_status = {row["status"]: row for row in section["rows"]}
for status in whitelist_statuses:
assert rows_by_status[status]["deleteQuantity"] == "无需删除"
assert rows_by_status["缺少的信息"]["deleteQuantity"] == "2"
def test_parse_label_quantity_text_accepts_complete_draft_sample():
assert InventoryManage._parse_label_quantity_text("未提交的草稿 (0)") == ("未提交的草稿", 0)
def test_parse_label_quantity_text_preserves_long_colon_label():
assert InventoryManage._parse_label_quantity_text("已提交:提供缺少的信息 (1)") == (
"已提交:提供缺少的信息",
1,
)
def test_parse_label_quantity_text_accepts_chinese_parentheses():
assert InventoryManage._parse_label_quantity_text("已提交查看限制0") == ("已提交:查看限制", 0)
def test_parse_label_quantity_text_accepts_dynamic_text_and_thousands():
assert InventoryManage._parse_label_quantity_text("Some dynamic label (1,234)") == (
"Some dynamic label",
1234,
)
def test_parse_label_quantity_text_rejects_empty_missing_quantity_and_numeric_label():
assert InventoryManage._parse_label_quantity_text("") is None
assert InventoryManage._parse_label_quantity_text("没有数量") is None
assert InventoryManage._parse_label_quantity_text("123 (4)") is None
assert InventoryManage._parse_label_quantity_text("未提交的草稿 (0) 已提交:提供缺少的信息 (1)") is None
def test_parse_quick_view_tag_rows_supports_raw_and_split_dom_rows():
rows = InventoryManage._parse_quick_view_tag_rows(
[
{"raw_text": "未提交的草稿 (0)"},
{"raw_text": "未提交的草稿 (0)"},
{"label_text": "已提交:提供缺少的信息", "quantity_text": "1"},
{"raw_text": "没有数量"},
],
"郭亚芳",
"德国",
)
assert rows == [
{
"shop_name": "郭亚芳",
"country": "德国",
"tag": "未提交的草稿",
"quantity": 0,
"raw_text": "未提交的草稿 (0)",
},
{
"shop_name": "郭亚芳",
"country": "德国",
"tag": "已提交:提供缺少的信息",
"quantity": 1,
"raw_text": "已提交:提供缺少的信息 (1)",
},
]
def test_parse_quick_view_tag_rows_filters_navigation_and_pagination_noise():
rows = InventoryManage._parse_quick_view_tag_rows(
[
{"raw_text": "补全草稿 (1)"},
{"raw_text": "未提交的草稿 (0)"},
{"raw_text": "已提交:提供缺少的信息 (1)"},
{"raw_text": "已提交:查看限制 (0)"},
{"raw_text": "商品信息草稿 至 0 / 0 (0)"},
{"raw_text": "页面 / 0 (0)"},
],
"郭亚芳",
"德国",
)
assert rows == [
{
"shop_name": "郭亚芳",
"country": "德国",
"tag": "未提交的草稿",
"quantity": 0,
"raw_text": "未提交的草稿 (0)",
},
{
"shop_name": "郭亚芳",
"country": "德国",
"tag": "已提交:提供缺少的信息",
"quantity": 1,
"raw_text": "已提交:提供缺少的信息 (1)",
},
{
"shop_name": "郭亚芳",
"country": "德国",
"tag": "已提交:查看限制",
"quantity": 0,
"raw_text": "已提交:查看限制 (0)",
},
]
def test_parse_cart_ratio_rows_extracts_chinese_day_ratios():
rows = InventoryManage._parse_cart_ratio_rows(
[{"raw_text": "推荐报价百分比 2 天前 25% 30 天前 30%"}],
"德国",
)
assert rows == {"country": "德国", "ratio2DaysAgo": "25%", "ratio30DaysAgo": "30%"}
def test_parse_cart_ratio_rows_extracts_english_day_ratios():
rows = InventoryManage._parse_cart_ratio_rows(
[{"raw_text": "Featured Offer Percentage 2 days ago 25.5% 30 days ago 30%"}],
"英国",
)
assert rows == {"country": "英国", "ratio2DaysAgo": "25.5%", "ratio30DaysAgo": "30%"}
def test_parse_cart_ratio_rows_supports_percent_before_day_label():
rows = InventoryManage._parse_cart_ratio_rows(
[{"raw_text": "25% 2 天前 30% 30 天前"}],
"法国",
)
assert rows == {"country": "法国", "ratio2DaysAgo": "25%", "ratio30DaysAgo": "30%"}
def test_parse_cart_ratio_rows_returns_empty_for_missing_day():
rows = InventoryManage._parse_cart_ratio_rows(
[{"raw_text": "推荐报价百分比 2 天前 25%"}],
"意大利",
)
assert rows == {"country": "意大利", "ratio2DaysAgo": "25%", "ratio30DaysAgo": ""}
def test_parse_all_cart_ratio_rows_extracts_all_target_countries_without_switching():
rows = InventoryManage._parse_all_cart_ratio_rows(
[
{
"raw_text": (
"推荐报价百分比 商城 2 天前 30 天前 "
"欧洲 60% 64% "
"英国 50% 61% "
"德国 57% 65% "
"法国 81% 77% "
"西班牙 70% 70% "
"意大利 70% 62%"
)
}
]
)
assert rows == [
{"country": "欧洲", "ratio2DaysAgo": "60%", "ratio30DaysAgo": "64%"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%"},
{"country": "法国", "ratio2DaysAgo": "81%", "ratio30DaysAgo": "77%"},
{"country": "西班牙", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "70%"},
{"country": "意大利", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "62%"},
]
def test_parse_all_cart_ratio_rows_prefers_direct_rendered_country_rows():
rows = InventoryManage._parse_all_cart_ratio_rows(
[
{"country": "欧洲", "ratio2DaysAgo": "60 %", "ratio30DaysAgo": "64%", "source": "rendered-country-row"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%", "source": "rendered-country-row"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%", "source": "rendered-country-row"},
]
)
assert rows == [
{"country": "欧洲", "ratio2DaysAgo": "60%", "ratio30DaysAgo": "64%"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%"},
{"country": "法国", "ratio2DaysAgo": "", "ratio30DaysAgo": ""},
{"country": "西班牙", "ratio2DaysAgo": "", "ratio30DaysAgo": ""},
{"country": "意大利", "ratio2DaysAgo": "", "ratio30DaysAgo": ""},
]
def test_parse_all_cart_ratio_rows_supports_date_headers_without_day_labels():
rows = InventoryManage._parse_all_cart_ratio_rows(
[
{
"raw_text": (
"推荐报价百分比 商城 24 Apr 27 Mar "
"欧洲 60% 64% "
"英国 50% 61% "
"德国 57% 65% "
"法国 81% 77% "
"西班牙 70% 70% "
"意大利 70% 62%"
)
}
]
)
assert rows == [
{"country": "欧洲", "ratio2DaysAgo": "60%", "ratio30DaysAgo": "64%"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%"},
{"country": "法国", "ratio2DaysAgo": "81%", "ratio30DaysAgo": "77%"},
{"country": "西班牙", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "70%"},
{"country": "意大利", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "62%"},
]
def test_parse_all_cart_ratio_rows_supports_rendered_geometry_rows():
rows = InventoryManage._parse_all_cart_ratio_rows(
[
{"attrs": {"title": "欧洲"}, "rect_left": 1315.0, "rect_top": 346.5, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "60%", "rect_left": 1510.0, "rect_top": 357.6, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "64%", "rect_left": 1693.3, "rect_top": 357.6, "rect_width": 32.0, "rect_height": 20.0},
{"attrs": {"title": "英国"}, "rect_left": 1315.0, "rect_top": 390.6, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "50%", "rect_left": 1510.1, "rect_top": 401.7, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "61%", "rect_left": 1693.3, "rect_top": 401.6, "rect_width": 32.0, "rect_height": 20.0},
{"attrs": {"title": "德国"}, "rect_left": 1315.0, "rect_top": 433.8, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "57%", "rect_left": 1510.1, "rect_top": 444.9, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "65%", "rect_left": 1693.3, "rect_top": 444.9, "rect_width": 32.0, "rect_height": 20.0},
{"attrs": {"title": "法国"}, "rect_left": 1315.0, "rect_top": 477.0, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "81%", "rect_left": 1510.1, "rect_top": 488.1, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "77%", "rect_left": 1693.3, "rect_top": 488.1, "rect_width": 32.0, "rect_height": 20.0},
{"attrs": {"title": "西班牙"}, "rect_left": 1315.0, "rect_top": 520.2, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "70%", "rect_left": 1510.1, "rect_top": 531.3, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "70%", "rect_left": 1693.3, "rect_top": 531.3, "rect_width": 32.0, "rect_height": 20.0},
{"attrs": {"title": "意大利"}, "rect_left": 1315.0, "rect_top": 563.4, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "70%", "rect_left": 1510.1, "rect_top": 574.5, "rect_width": 32.0, "rect_height": 20.0},
{"label_text": "62%", "rect_left": 1693.3, "rect_top": 574.5, "rect_width": 32.0, "rect_height": 20.0},
]
)
assert rows == [
{"country": "欧洲", "ratio2DaysAgo": "60%", "ratio30DaysAgo": "64%"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%"},
{"country": "法国", "ratio2DaysAgo": "81%", "ratio30DaysAgo": "77%"},
{"country": "西班牙", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "70%"},
{"country": "意大利", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "62%"},
]
def test_parse_all_cart_ratio_rows_dedupes_nested_rendered_percentage_nodes():
rows = InventoryManage._parse_all_cart_ratio_rows(
[
{"attrs": {"title": "欧洲"}, "rect_left": 1315.0, "rect_top": 346.5, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "60%", "rect_left": 1510.1, "rect_top": 357.6, "rect_width": 160.0, "rect_height": 20.0},
{"label_text": "60%", "rect_left": 1510.1, "rect_top": 357.6, "rect_width": 29.6, "rect_height": 20.0},
{"raw_text": "64%", "rect_left": 1681.7, "rect_top": 346.5, "rect_width": 183.2, "rect_height": 43.2},
{"label_text": "64%", "rect_left": 1693.3, "rect_top": 357.6, "rect_width": 160.0, "rect_height": 20.0},
{"attrs": {"title": "英国"}, "rect_left": 1315.0, "rect_top": 390.6, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "50%", "rect_left": 1510.1, "rect_top": 401.7, "rect_width": 160.0, "rect_height": 18.2},
{"label_text": "50%", "rect_left": 1510.1, "rect_top": 401.7, "rect_width": 27.6, "rect_height": 18.2},
{"label_text": "61%", "rect_left": 1693.3, "rect_top": 401.6, "rect_width": 27.6, "rect_height": 18.2},
{"attrs": {"title": "德国"}, "rect_left": 1315.0, "rect_top": 433.8, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "57%", "rect_left": 1510.1, "rect_top": 444.9, "rect_width": 160.0, "rect_height": 18.2},
{"label_text": "57%", "rect_left": 1510.1, "rect_top": 444.9, "rect_width": 27.6, "rect_height": 18.2},
{"label_text": "65%", "rect_left": 1693.3, "rect_top": 444.9, "rect_width": 27.6, "rect_height": 18.2},
{"attrs": {"title": "法国"}, "rect_left": 1315.0, "rect_top": 477.0, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "81%", "rect_left": 1510.1, "rect_top": 488.1, "rect_width": 160.0, "rect_height": 18.2},
{"label_text": "81%", "rect_left": 1510.1, "rect_top": 488.1, "rect_width": 27.6, "rect_height": 18.2},
{"label_text": "77%", "rect_left": 1693.3, "rect_top": 488.1, "rect_width": 27.6, "rect_height": 18.2},
{"attrs": {"title": "西班牙"}, "rect_left": 1315.0, "rect_top": 520.2, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "70%", "rect_left": 1510.1, "rect_top": 531.3, "rect_width": 160.0, "rect_height": 18.2},
{"label_text": "70%", "rect_left": 1510.1, "rect_top": 531.3, "rect_width": 27.6, "rect_height": 18.2},
{"label_text": "70%", "rect_left": 1693.3, "rect_top": 531.3, "rect_width": 27.6, "rect_height": 18.2},
{"attrs": {"title": "意大利"}, "rect_left": 1315.0, "rect_top": 563.4, "rect_width": 550.0, "rect_height": 43.0},
{"raw_text": "70%", "rect_left": 1510.1, "rect_top": 574.5, "rect_width": 160.0, "rect_height": 18.2},
{"label_text": "70%", "rect_left": 1510.1, "rect_top": 574.5, "rect_width": 27.6, "rect_height": 18.2},
{"raw_text": "62%", "rect_left": 1681.7, "rect_top": 563.4, "rect_width": 183.2, "rect_height": 43.2},
{"label_text": "62%", "rect_left": 1693.3, "rect_top": 574.5, "rect_width": 27.6, "rect_height": 18.2},
]
)
assert rows == [
{"country": "欧洲", "ratio2DaysAgo": "60%", "ratio30DaysAgo": "64%"},
{"country": "英国", "ratio2DaysAgo": "50%", "ratio30DaysAgo": "61%"},
{"country": "德国", "ratio2DaysAgo": "57%", "ratio30DaysAgo": "65%"},
{"country": "法国", "ratio2DaysAgo": "81%", "ratio30DaysAgo": "77%"},
{"country": "西班牙", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "70%"},
{"country": "意大利", "ratio2DaysAgo": "70%", "ratio30DaysAgo": "62%"},
]
def test_has_any_cart_ratio_detects_country_table_rows():
assert InventoryManage._has_any_cart_ratio(
[
{
"raw_text": (
"推荐报价百分比 商城 2 天前 30 天前 "
"欧洲 60% 64% 英国 50% 61% 德国 57% 65%"
)
}
]
)
def test_has_any_cart_ratio_detects_date_header_country_table_rows():
assert InventoryManage._has_any_cart_ratio(
[
{
"raw_text": (
"推荐报价百分比 商城 24 Apr 27 Mar "
"欧洲 60% 64% 英国 50% 61% 德国 57% 65%"
)
}
]
)
def test_has_complete_cart_ratio_rows_requires_all_target_countries():
assert not InventoryManage._has_complete_cart_ratio_rows(
[
{
"raw_text": (
"推荐报价百分比 商城 2 天前 30 天前 "
"欧洲 60% 64% 英国 50% 61% 德国 57% 65%"
)
}
]
)
def test_has_complete_cart_ratio_rows_detects_full_country_table():
assert InventoryManage._has_complete_cart_ratio_rows(
[
{
"raw_text": (
"推荐报价百分比 商城 2 天前 30 天前 "
"欧洲 60% 64% "
"英国 50% 61% "
"德国 57% 65% "
"法国 81% 77% "
"西班牙 70% 70% "
"意大利 70% 62%"
)
}
]
)