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_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": [ {"status": "商品信息草稿", "quantity": "12", "deleteQuantity": "3", "processStatus": "已完成"}, {"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", "processStatus": "已完成"}, {"type": "completeDraft", "quantity": "0", "deleteQuantity": "0", "processStatus": ""}, ], } 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", "processStatus": "已完成"}, {"type": "completeDraft", "quantity": "0", "deleteQuantity": "0", "processStatus": ""}, ], } assert PatrolDeleteTask._normalize_country_sections_for_result([merged]) == [ { "country": "西班牙", "rows": [ {"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"}, {"status": "商品信息草稿", "quantity": "1", "deleteQuantity": "1", "processStatus": "已完成"}, {"status": "商品信息草稿", "quantity": "0", "deleteQuantity": "0", "processStatus": ""}, ], } ] 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": "郭亚芳", "shop_done": False, "heartbeat_only": True, }, { "task_id": 1, "shop_name": "郭亚芳", "shop_done": False, "heartbeat_only": True, }, { "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", "processStatus": "已完成", }, { "type": "completeDraft", "status": "已提交：提供缺少的信息", "quantity": "2", "deleteQuantity": "0", "processStatus": "", }, ], } assert PatrolDeleteTask._normalize_country_sections_for_result([result["countrySection"]]) == [ { "country": "德国", "rows": [ {"status": "未提交的草稿", "quantity": "0", "deleteQuantity": "3", "processStatus": "已完成"}, {"status": "已提交：提供缺少的信息", "quantity": "2", "deleteQuantity": "0", "processStatus": ""}, ], } ] 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%" ) } ] )