Files
crawler-plugin/app/tests/test_patrol_delete.py

1220 lines
48 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_omits_status_fields_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": [
{"quantity": "12", "deleteQuantity": "3"},
{"status": "缺少的信息", "quantity": "1", "deleteQuantity": "1", "processStatus": "已完成"},
],
}
]
def test_replace_complete_draft_rows_uses_quick_view_rows():
country_section = {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"status": "商品信息草稿", "quantity": "", "deleteQuantity": "0", "processStatus": ""},
],
}
complete_draft_section = {
"country": "西班牙",
"rows": [
{"type": "completeDraft", "quantity": "1", "deleteQuantity": "1"},
{"type": "completeDraft", "quantity": "0", "deleteQuantity": "0"},
],
}
merged = PatrolDeleteTask._replace_complete_draft_rows(country_section, complete_draft_section)
assert merged == {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"type": "completeDraft", "quantity": "1", "deleteQuantity": "1"},
{"type": "completeDraft", "quantity": "0", "deleteQuantity": "0"},
],
}
assert PatrolDeleteTask._normalize_country_sections_for_result([merged]) == [
{
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"quantity": "1", "deleteQuantity": "1"},
{"quantity": "0", "deleteQuantity": "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, delete_conditions=None):
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_process_task_passes_delete_conditions(monkeypatch):
from config import runing_shop, runing_task
runing_task.clear()
runing_shop.clear()
seen_conditions = []
class Driver:
def close_store(self):
return None
def fake_open_shop(self, **kwargs):
return Driver()
def fake_process_country(self, driver, task_id, shop_name, country_name, delete_conditions=None):
seen_conditions.append(delete_conditions)
return {
"success": True,
"countrySection": {
"country": country_name,
"rows": [{"status": "全部", "quantity": "", "deleteQuantity": "0", "processStatus": "无可删商品"}],
},
"cartRatio": {"country": country_name, "ratio": ""},
"deletedCount": 0,
"reopenRequired": False,
}
monkeypatch.setattr(PatrolDeleteTask, "open_shop", fake_open_shop)
monkeypatch.setattr(PatrolDeleteTask, "process_country", fake_process_country)
monkeypatch.setattr(PatrolDeleteTask, "post_result", lambda self, **kwargs: None)
PatrolDeleteTask().process_task(
{
"type": "patrol-delete-run",
"data": {
"taskId": 7,
"items": [{"shopName": "郭亚芳", "companyName": "rongchuang123"}],
"country_sections": [{"country": "德国", "rows": [{"status": "全部"}]}],
"delete_conditions": [{"id": 1, "conditionText": "商品信息草稿、补全草稿"}],
},
}
)
assert seen_conditions == [["商品信息草稿", "补全草稿"]]
def test_delete_complete_draft_country_section_reports_delete_quantity(monkeypatch):
class Driver:
def delete_complete_drafts_from_non_whitelisted_tags(self, shop_name, country_name, **kwargs):
return {
"draftRows": [
{"tag": "未提交的草稿", "quantity": 0},
{"tag": "已提交:提供缺少的信息", "quantity": 2},
],
"deleteCounts": {"未提交的草稿": 3},
"totalDeleted": 3,
}
result = PatrolDeleteTask()._delete_complete_draft_country_section_safe(
Driver(),
"郭亚芳",
"德国",
["未提交的草稿"],
)
assert result["deletedCount"] == 3
assert result["countrySection"] == {
"country": "德国",
"rows": [
{"type": "completeDraft", "status": "未提交的草稿", "quantity": "0", "deleteQuantity": "3"},
{"type": "completeDraft", "status": "已提交:提供缺少的信息", "quantity": "2", "deleteQuantity": "0"},
],
}
assert PatrolDeleteTask._normalize_country_sections_for_result([result["countrySection"]]) == [
{
"country": "德国",
"rows": [
{"quantity": "0", "deleteQuantity": "3"},
{"quantity": "2", "deleteQuantity": "0"},
],
}
]
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(
"郭亚芳",
"德国",
delete_conditions=["商品信息草稿"],
)
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_delete_all_listings_from_non_whitelisted_statuses_respects_allowed_statuses(monkeypatch):
driver = InventoryManage({})
state = {"selected": [], "deleted": 0}
def fake_read_listing_status_rows_for_delete(shop_name, country, **kwargs):
return [
{"status": "商品信息草稿", "quantity": "2", "deleteQuantity": "", "processStatus": ""},
{"status": "缺少的信息", "quantity": "3", "deleteQuantity": "", "processStatus": ""},
{"status": "全部", "quantity": "9", "deleteQuantity": "", "processStatus": ""},
]
def fake_select_listing_status(status, **kwargs):
state["selected"].append(status)
def fake_delete_selected_filtered_inventory_rows(status, **kwargs):
state["deleted"] += 1
return {
"deletedCount": 1,
"target": {"sku": status, "rawText": status},
"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(
"郭亚芳",
"德国",
allowed_statuses=["缺少的信息"],
)
assert state["selected"] == ["缺少的信息"]
assert state["deleted"] == 3
assert result["deleteCounts"] == {"缺少的信息": 3}
def test_delete_all_listings_from_non_whitelisted_statuses_empty_allowed_statuses_deletes_nothing(monkeypatch):
driver = InventoryManage({})
state = {"selected": []}
monkeypatch.setattr(
driver,
"_read_listing_status_rows_for_delete",
lambda shop_name, country, **kwargs: [{"status": "商品信息草稿", "quantity": "2"}],
)
monkeypatch.setattr(driver, "select_listing_status", lambda status, **kwargs: state["selected"].append(status))
result = driver.delete_all_listings_from_non_whitelisted_statuses(
"郭亚芳",
"德国",
allowed_statuses=[],
)
assert state["selected"] == []
assert result["deleteCandidates"] == []
assert result["totalDeleted"] == 0
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_build_deletable_status_rows_filters_by_delete_conditions():
rows = [
{"status": "商品信息草稿", "quantity": "2"},
{"status": "缺少的信息", "quantity": "5"},
{"status": "全部", "quantity": "9"},
]
assert InventoryManage._build_deletable_status_rows(rows, ["商品信息草稿"]) == [
{"status": "商品信息草稿", "canonicalStatus": "商品信息草稿", "quantity": 2},
]
assert InventoryManage._build_deletable_status_rows(rows, []) == []
def test_build_deletable_complete_draft_tag_rows_respects_complete_draft_condition():
rows = [
{"tag": "未提交的草稿", "quantity": 1},
{"tag": "已提交:提供缺少的信息", "quantity": 2},
]
assert InventoryManage._build_deletable_complete_draft_tag_rows(rows, ["已提交:提供缺少的信息"]) == [
{"tag": "已提交:提供缺少的信息", "canonicalTag": "已提交:提供缺少的信息", "quantity": 2},
]
assert InventoryManage._build_deletable_complete_draft_tag_rows(rows, ["补全草稿"]) == [
{"tag": "未提交的草稿", "canonicalTag": "未提交的草稿", "quantity": 1},
{"tag": "已提交:提供缺少的信息", "canonicalTag": "已提交:提供缺少的信息", "quantity": 2},
]
assert InventoryManage._build_deletable_complete_draft_tag_rows(rows, []) == []
def test_parse_delete_condition_texts_accepts_frontend_payload_shapes():
assert PatrolDeleteTask._parse_delete_condition_texts(
[
{"id": 1, "conditionText": "商品信息草稿、缺少的信息"},
{"id": 2, "condition_text": "补全草稿"},
"详情页面已删除",
]
) == ["商品信息草稿", "缺少的信息", "补全草稿", "详情页面已删除"]
def test_resolve_delete_condition_texts_defaults_to_empty_list_when_missing():
assert PatrolDeleteTask._resolve_delete_condition_texts({}) == []
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_extract_delete_success_count_accepts_success_message():
assert InventoryManage._extract_delete_success_count("所做更改需要 15 分钟才会显示在商品详情页面上。 3 个商品已删除。") == 3
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_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%"
)
}
]
)