dalongxia/remote_sync/test_dewu_1688_auto_fulfillment.py

3502 lines
132 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.

#!/usr/bin/env python3
from __future__ import annotations
import importlib.util
import json
import os
import queue
import sqlite3
import subprocess
import tempfile
import threading
import time
from pathlib import Path
import unittest
from unittest import mock
ROOT = Path(__file__).resolve().parent
ENGINE_PATH = (
ROOT
/ "dewu-1688-auto-fulfillment"
/ "scripts"
/ "fulfillment_engine.py"
)
CHAT_API_PATH = ROOT / "chat_api.py"
def load_module(path: Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
class FulfillmentEngineUnitTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.module = load_module(ENGINE_PATH, "dewu_1688_engine_test")
def make_engine(self, run_erp):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
engine = self.module.FulfillmentEngine(
db_path=str(Path(temp.name) / "state.sqlite3"),
run_erp=run_erp,
template_url=lambda: "https://cdn.example.com/dewu-template.xlsx",
)
return engine
@staticmethod
def identity():
return {
"user_id": "u1",
"user_name": "陆南亭",
"_dewu_1688_authorized": True,
}
@staticmethod
def order_message():
return "\n".join(
[
"得物1688自动发货",
"订单号:110208666749084488",
"收件人:赵",
"手机号17819054454",
"收货地址:云南省昭通市威信县扎西镇红城国际2期16栋2801",
]
)
def test_message_parser_and_validation(self):
rows = self.module.parse_orders_from_message(self.order_message())
valid, errors = self.module.validate_orders(rows)
self.assertFalse(errors)
self.assertEqual(valid[0]["order_id"], "110208666749084488")
self.assertEqual(valid[0]["recipient"], "赵")
self.assertEqual(valid[0]["phone"], "17819054454")
self.assertIn("红城国际2期", valid[0]["address"])
def test_bare_trigger_returns_entry_guidance(self):
engine = self.make_engine(lambda action, payload: {})
answer = engine.handle(
"得物1688自动发货",
{},
self.identity(),
"c-entry",
"u1",
"s-entry",
)
self.assertIn("请直接上传", answer)
self.assertIn("下载导入模板", answer)
self.assertIn("无需二次确认", answer)
self.assertNotIn("17819054454", answer)
self.assertNotIn("红城国际", answer)
def test_csv_alias_headers_and_duplicate_rejection(self):
data = (
"线上订单号,收货人,联系电话,详细地址,备注\n"
"110208666749084488,赵,17819054454,"
"云南省昭通市威信县扎西镇红城国际2期16栋2801,测试\n"
).encode("utf-8")
rows = self.module.parse_spreadsheet_bytes(data, "orders.csv")
valid, errors = self.module.validate_orders(rows)
self.assertFalse(errors)
self.assertEqual(valid[0]["remark"], "测试")
_, duplicate_errors = self.module.validate_orders(rows + rows)
self.assertTrue(any("订单号重复" in item for item in duplicate_errors))
def test_dewu_export_headers_privacy_phone_and_split_address(self):
data = (
"订单号,订单类型,收件人姓名,收件人手机号,收件人省,收件人市,"
"收件人区,收件人街道,收件人详细地址,订单备注\n"
"110208666356855463,品牌直发,蹦蹦[6429],18466443253-6429,"
"山西省,朔州市,朔城区,北旺庄街道,长宁东街11号山西工学院[6429],直发\n"
).encode("utf-8")
rows = self.module.parse_spreadsheet_bytes(data, "得物订单导出.csv")
valid, errors = self.module.validate_orders(rows)
self.assertFalse(errors)
self.assertEqual(valid[0]["recipient"], "蹦蹦[6429]")
self.assertEqual(valid[0]["phone"], "18466443253-6429")
self.assertEqual(
valid[0]["address"],
"山西省朔州市朔城区北旺庄街道长宁东街11号山西工学院[6429]",
)
self.assertEqual(valid[0]["remark"], "直发")
def test_chat_api_markdown_table_batch(self):
message = "\n".join(
[
"【技能路由:dewu-1688-auto-fulfillment】",
"得物1688自动发货",
"```markdown",
"| 订单号 | 订单类型 | 收件人姓名 | 收件人手机号 | 收件人省 | 收件人市 | 收件人区 | 收件人街道 | 收件人详细地址 | 订单备注 |",
"| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |",
"| 110208666356855463 | 品牌直发 | 蹦蹦[6429] | 18466443253-6429 | 山西省 | 朔州市 | 朔城区 | 北旺庄街道 | 长宁东街11号山西工学院[6429] | 直发 |",
"| 110208670256244987 | 品牌直发 | 李[9068] | 18466443438-9068 | 四川省 | 泸州市 | 江阳区 | 大山坪街道 | 新区转盘军干所[9068] | 备注含\\|竖线 |",
"```",
]
)
rows = self.module.parse_orders_from_markdown_table(message)
valid, errors = self.module.validate_orders(rows)
self.assertFalse(errors)
self.assertEqual(len(valid), 2)
self.assertEqual(valid[0]["phone"], "18466443253-6429")
self.assertEqual(
valid[1]["address"],
"四川省泸州市江阳区大山坪街道新区转盘军干所[9068]",
)
self.assertEqual(valid[1]["remark"], "备注含|竖线")
first_message_table = message.replace(
"【技能路由:dewu-1688-auto-fulfillment】\n得物1688自动发货\n",
"请分析上传的表格内容\n",
)
self.assertTrue(self.module.is_request(first_message_table))
engine = self.make_engine(lambda action, payload: {})
loaded, load_errors = engine._load_orders(message, {})
self.assertFalse(load_errors)
self.assertEqual([row["order_id"] for row in loaded], [
"110208666356855463",
"110208670256244987",
])
def test_order_table_takes_priority_over_template_text_in_skill_context(self):
run_erp = mock.Mock(
return_value={
"status": "ok",
"orders": [
{
"order_id": "110209221705869901",
"state": "blocked",
"message": "isolated test stop",
}
],
}
)
engine = self.make_engine(run_erp)
message = "\n".join(
[
"【技能路由:dewu-1688-auto-fulfillment】",
"技能说明:用户可以发送‘下载得物1688发货模板’。",
"| 订单号 | 收件人姓名 | 收件人手机号 | 收件人省 | 收件人市 | 收件人区 | 收件人街道 | 收件人详细地址 |",
"| --- | --- | --- | --- | --- | --- | --- | --- |",
"| 110209221705869901 | 张三 | 13800000000 | 江苏省 | 常州市 | 新北区 | 三井街道 | 测试路1号 |",
]
)
answer = engine.handle(
message,
{},
self.identity(),
"c-template-context",
"u1",
"s-template-context",
)
self.assertEqual(run_erp.call_count, 1)
self.assertIn("110209221705869901", answer)
self.assertNotIn("请直接上传", answer)
def test_unrelated_markdown_table_does_not_trigger_fulfillment(self):
message = "\n".join(
[
"请分析这个表",
"| SKU | 销量 | 库存 |",
"| --- | --- | --- |",
"| SKU-001 | 10 | 20 |",
]
)
self.assertFalse(self.module.is_request(message))
def test_batch_accepts_500_rows_and_rejects_501(self):
rows = [
{
"order_id": str(100000000000 + index),
"recipient": f"测试{index}",
"phone": "17819054454",
"address": "江苏省常州市测试区测试路1号",
}
for index in range(500)
]
valid, errors = self.module.validate_orders(rows)
self.assertEqual(len(valid), 500)
self.assertFalse(errors)
_, too_many = self.module.validate_orders(
rows
+ [
{
"order_id": "999999999999",
"recipient": "测试",
"phone": "17819054454",
"address": "江苏省常州市测试区测试路1号",
}
]
)
self.assertIn("单批最多处理 500 条", too_many[0])
def test_attachment_collector_accepts_xlsx_and_ignores_images(self):
payload = {
"attachments": [
{
"filename": "得物订单.xlsx",
"url": "https://cdn.example.com/orders",
"mime_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
},
{
"filename": "product.png",
"url": "https://cdn.example.com/product.png",
"mime_type": "image/png",
},
]
}
result = self.module.collect_spreadsheet_urls(payload)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["filename"], "得物订单.xlsx")
def test_sms_code_resumes_same_batch_without_persisting_code(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if len(calls) == 1:
return {"status": "needs_sms"}
if action == "apply":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900001",
"online_1688_order_no": "5126531473532161999",
}
],
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常1688一件代发",
}
],
}
engine = self.make_engine(run_erp)
first = engine.handle(
self.order_message(),
{},
self.identity(),
"c1",
"u1",
"s1",
)
self.assertIn("需要短信验证", first)
second = engine.handle(
"588513",
{},
self.identity(),
"c1",
"u1",
"s1",
)
self.assertIn("900001", second)
self.assertEqual(calls[1][1]["otp"], "588513")
self.assertEqual([item[0] for item in calls], ["inspect", "inspect", "apply"])
with sqlite3.connect(engine.db_path) as connection:
raw = json.dumps(connection.execute(
"SELECT * FROM dewu_1688_batches"
).fetchall(), ensure_ascii=False)
self.assertNotIn("588513", raw)
def test_missing_sms_can_request_resend(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
return {
"status": "needs_sms",
"message": "ERP 已重新发送登录验证码。",
}
engine = self.make_engine(run_erp)
first = engine.handle(
self.order_message(),
{},
self.identity(),
"c-sms",
"u1",
"s-sms",
)
self.assertIn("需要短信验证", first)
second = engine.handle(
"没有收到验证码,重新发送验证码",
{},
self.identity(),
"c-sms",
"u1",
"s-sms",
)
self.assertIn("已重新发送", second)
self.assertTrue(calls[1][1]["resend_sms"])
def test_purchase_without_number_is_blocked(self):
run_erp = mock.Mock(
return_value={
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "already_processed",
"shop_name": "得物-全米",
"order_status": "1688一件代发 / 已生成采购",
"purchase_order_no": "",
}
],
}
)
engine = self.make_engine(run_erp)
answer = engine.handle(
self.order_message(),
{},
self.identity(),
"c2",
"u1",
"s2",
)
self.assertIn("禁止重复", answer)
latest = engine.latest_batch(engine.context_key("c2", "u1", "s2"))
self.assertEqual(latest["stage"], "blocked")
self.assertEqual(run_erp.call_count, 1)
def test_repeated_completed_orders_return_completed_without_reexecution(self):
calls = []
def run_erp(action, payload):
calls.append(action)
if action == "inspect":
state = "ready" if calls.count("inspect") == 1 else "already_processed"
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": state,
"shop_name": "得物-全米",
"order_status": "1688一件代发 / 已生成采购",
"purchase_order_no": "800488" if state == "already_processed" else "",
}
],
"browser_code": "ERP-01",
"browser_reset": "order_list",
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "800488",
"online_1688_order_no": "5127606396343016332",
"message": "完成",
}
],
"browser_code": "ERP-01",
"browser_reset": "order_list",
}
engine = self.make_engine(run_erp)
first = engine.handle(
self.order_message(),
{},
self.identity(),
"c-repeat-first",
"u1",
"s-repeat-first",
)
self.assertIn("5127606396343016332", first)
second = engine.handle(
self.order_message(),
{},
self.identity(),
"c-repeat-second",
"u1",
"s-repeat-second",
)
latest = engine.latest_batch(
engine.context_key("c-repeat-second", "u1", "s-repeat-second")
)
self.assertEqual(latest["stage"], "completed")
self.assertEqual(
latest["result"]["completion_source"],
"existing_verified_history",
)
self.assertIn("5127606396343016332", second)
self.assertIn("本批次已完成 1 条", second)
self.assertEqual(calls, ["inspect", "apply", "inspect"])
def test_ready_batch_applies_automatically(self):
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常1688一件代发",
}
],
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900001",
"online_1688_order_no": "5126531473532161999",
}
],
}
run_erp_mock = mock.Mock(side_effect=run_erp)
engine = self.make_engine(run_erp_mock)
preview = engine.handle(
self.order_message(),
{},
self.identity(),
"c3",
"u1",
"s3",
)
latest = engine.latest_batch(engine.context_key("c3", "u1", "s3"))
batch_id = latest["batch_id"]
self.assertIn(batch_id, preview)
self.assertIn("900001", preview)
self.assertIn("5126531473532161999", preview)
self.assertIn("| 110208666749084488 | 900001 | 5126531473532161999 | 完成 |", preview)
self.assertNotIn("待回写", preview)
self.assertEqual(run_erp_mock.call_count, 2)
def test_completed_result_without_online_number_is_rejected(self):
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"shop_name": "得物-全米",
"order_status": "1688一件代发 / 已生成采购",
"purchase_order_no": "788586",
}
],
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "788586",
"online_1688_order_no": "",
"message": "完成",
}
],
}
run_erp_mock = mock.Mock(side_effect=run_erp)
engine = self.make_engine(run_erp_mock)
first = engine.handle(
self.order_message(),
{},
self.identity(),
"c-history-first",
"u1",
"s-history-first",
)
self.assertIn("得物1688自动发货未完成", first)
self.assertIn("1688线上单号复核未通过", first)
latest = engine.latest_batch(
engine.context_key("c-history-first", "u1", "s-history-first")
)
self.assertEqual(latest["stage"], "blocked")
self.assertEqual(
latest["result"]["orders"][0]["state"],
"stopped",
)
self.assertEqual(
[call.args[0] for call in run_erp_mock.call_args_list],
["inspect", "apply", "inspect", "apply"],
)
def test_apply_interruption_rechecks_and_finishes_before_reply(self):
calls = []
inspect_count = 0
apply_count = 0
def run_erp(action, payload):
nonlocal inspect_count, apply_count
calls.append((action, dict(payload)))
if action == "inspect":
inspect_count += 1
if inspect_count == 1:
return {
"status": "ok",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
}
],
}
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"purchase_order_no": "900030",
}
],
}
apply_count += 1
if apply_count == 1:
return {
"status": "stopped",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "stopped",
"purchase_order_no": "900030",
"message": "浏览器执行中断",
}
],
}
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900030",
"online_1688_order_no": "5126531473532162030",
}
],
}
engine = self.make_engine(run_erp)
answer = engine.handle(
self.order_message(),
{},
self.identity(),
"c-interrupted-apply",
"u1",
"s-interrupted-apply",
)
latest = engine.latest_batch(
engine.context_key(
"c-interrupted-apply",
"u1",
"s-interrupted-apply",
)
)
self.assertEqual(
[item[0] for item in calls],
["inspect", "apply", "inspect", "apply"],
)
self.assertEqual(latest["stage"], "completed")
self.assertIn("本批次已完成 1 条", answer)
self.assertIn("5126531473532162030", answer)
progress_codes = {
event["step_code"]
for event in engine.progress_events(
latest["batch_id"],
limit=100,
)
}
self.assertIn("batch.recovery.queued", progress_codes)
self.assertIn("batch.recovery.started", progress_codes)
self.assertIn("batch.recovery.completed", progress_codes)
def test_continue_returns_latest_completed_result_without_reexecution(self):
run_erp = mock.Mock()
engine = self.make_engine(run_erp)
context_key = engine.context_key(
"c-continue-completed",
"u1",
"s-continue-completed",
)
batch = engine._create_batch(
context_key,
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
engine._update_batch(
batch["batch_id"],
stage="completed",
result={
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900040",
"online_1688_order_no": "5126531473532162040",
}
],
},
)
answer = engine.handle(
"继续",
{},
self.identity(),
"c-continue-completed",
"u1",
"s-continue-completed",
)
self.assertIn(batch["batch_id"], answer)
self.assertIn("本批次已完成 1 条", answer)
self.assertIn("5126531473532162040", answer)
self.assertNotIn("当前批次尚未结束", answer)
run_erp.assert_not_called()
def test_new_upload_after_completed_batch_starts_new_batch_with_active_route(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
return {
"status": "ok",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常1688一件代发",
}
],
}
self.assertEqual(action, "apply")
return {
"status": "ok",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900042",
"online_1688_order_no": "5126531473532162042",
}
],
}
engine = self.make_engine(run_erp)
context_key = engine.context_key(
"c-new-after-completed",
"u1",
"s-new-after-completed",
)
previous = engine._create_batch(
context_key,
"u1",
[
{
"order_id": "110208600000000001",
"recipient": "旧订单",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇旧地址",
}
],
)
engine._update_batch(
previous["batch_id"],
stage="completed",
result={"status": "partial", "orders": []},
)
identity = {
**self.identity(),
"_dewu_1688_active_context": True,
}
answer = engine.handle(
self.order_message(),
{},
identity,
"c-new-after-completed",
"u1",
"s-new-after-completed",
)
latest = engine.latest_batch(context_key)
self.assertNotEqual(previous["batch_id"], latest["batch_id"])
self.assertEqual(latest["stage"], "completed")
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
self.assertIn("本批次已完成 1 条", answer)
self.assertNotIn("当前批次尚未结束", answer)
def test_continue_rechecks_blocked_batch_and_returns_latest_result(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
return {
"status": "ok",
"browser_code": "ERP-03",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"purchase_order_no": "900041",
}
],
}
self.assertEqual(action, "apply")
return {
"status": "ok",
"browser_code": "ERP-03",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900041",
"online_1688_order_no": "5126531473532162041",
}
],
}
engine = self.make_engine(run_erp)
context_key = engine.context_key(
"c-continue-blocked",
"u1",
"s-continue-blocked",
)
batch = engine._create_batch(
context_key,
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
engine._update_batch(
batch["batch_id"],
stage="blocked",
result={
"status": "stopped",
"orders": [
{
"order_id": "110208666749084488",
"state": "stopped",
"purchase_order_no": "900041",
"message": "浏览器执行中断",
}
],
},
)
answer = engine.handle(
"继续",
{},
self.identity(),
"c-continue-blocked",
"u1",
"s-continue-blocked",
)
latest = engine.latest_batch(context_key)
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
self.assertEqual(latest["stage"], "completed")
self.assertIn("本批次已完成 1 条", answer)
self.assertIn("5126531473532162041", answer)
queued_messages = [
event["message"]
for event in engine.progress_events(
batch["batch_id"],
limit=100,
)
if event["step_code"] == "batch.recovery.queued"
]
self.assertIn(
"收到继续指令,正在从已确认的最后一步恢复处理",
queued_messages,
)
def test_strict_submission_ack_completes_before_online_number_writeback(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"shop_name": "得物-全米",
"order_status": "1688一件代发 / 已生成采购",
"purchase_order_no": "788586",
}
],
}
if action == "reconcile":
return {
"status": "ok",
"order_id": "110208666749084488",
"purchase_order_no": "788586",
"verified": True,
"online_1688_order_no": "5126531473532162444",
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "788586",
"online_1688_order_no": "",
"submission_acknowledged": True,
"submission_purchase_order_no": "788586",
"completion_evidence": (
"erp_backend_and_success_dialog"
),
"message": "完成",
}
],
}
engine = self.make_engine(run_erp)
with mock.patch.object(
engine,
"_schedule_ack_reconciliation",
) as schedule:
answer = engine.handle(
self.order_message(),
{},
self.identity(),
"c-strict-ack",
"u1",
"s-strict-ack",
)
latest = engine.latest_batch(
engine.context_key("c-strict-ack", "u1", "s-strict-ack")
)
self.assertEqual(latest["stage"], "completed")
self.assertIn("ERP已确认下单成功", answer)
self.assertIn("| 完成 |", answer)
self.assertNotIn("未完成", answer)
schedule.assert_called_once_with(latest["batch_id"])
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
engine._run_ack_reconciliation(latest["batch_id"])
reconciled = engine.batch_by_id(latest["batch_id"])
self.assertEqual(
reconciled["result"]["orders"][0]["online_1688_order_no"],
"5126531473532162444",
)
self.assertEqual(
[item[0] for item in calls],
["inspect", "apply", "reconcile"],
)
history = engine._completed_order_history(
["110208666749084488"]
)
self.assertEqual(
history["110208666749084488"]["online_1688_order_no"],
"5126531473532162444",
)
def test_background_reconciliation_stops_after_persisted_attempt_limit(self):
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
raise RuntimeError("temporary verification failure")
engine = self.make_engine(run_erp)
batch = engine._create_batch(
"session:reconcile-limit",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
result = {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "788586",
"online_1688_order_no": "",
"submission_acknowledged": True,
"submission_purchase_order_no": "788586",
"completion_evidence": "erp_backend_and_success_dialog",
"message": "完成",
}
],
}
engine._record_apply_result(batch["batch_id"], result)
engine._update_batch(batch["batch_id"], stage="completed", result=result)
with (
mock.patch.object(self.module, "ACK_RECONCILE_MAX_ATTEMPTS", 3),
mock.patch.object(self.module, "ACK_RECONCILE_POLL_SECONDS", 0),
mock.patch.object(self.module, "ACK_RECONCILE_TIMEOUT_SECONDS", 60),
):
engine._run_ack_reconciliation(batch["batch_id"])
engine._run_ack_reconciliation(batch["batch_id"])
self.assertEqual([item[0] for item in calls], ["reconcile"] * 3)
current = engine.batch_by_id(batch["batch_id"])
row = current["result"]["orders"][0]
self.assertEqual(row["online_order_reconcile_attempts"], 3)
self.assertEqual(row["online_order_reconcile_state"], "deferred")
events = engine.progress_events(batch["batch_id"])
self.assertEqual(
sum(
item.get("step_code")
== "supplier.1688.background_verify.deferred"
for item in events
),
1,
)
def test_mixed_batch_skips_blocked_order_and_applies_ready_order(self):
first_order = "110208674527596779"
second_order = "110208680086305832"
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": first_order,
"state": "blocked",
"shop_name": "得物-全米",
"order_status": "异常 未设定快递",
"message": "ERP状态不符合执行条件:异常 未设定快递",
},
{
"order_id": second_order,
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常 1688一件代发",
},
],
}
self.assertEqual(
[row["order_id"] for row in payload["orders"]],
[second_order],
)
self.assertEqual(payload["progress_offset"], 1)
self.assertEqual(payload["progress_total"], 2)
return {
"status": "ok",
"orders": [
{
"order_id": second_order,
"state": "completed",
"purchase_order_no": "900002",
"online_1688_order_no": "5126531473532162000",
}
],
}
message = "\n".join(
[
"得物1688自动发货",
"| 订单号 | 收件人 | 手机号 | 收货地址 |",
"| --- | --- | --- | --- |",
f"| {first_order} | 赵 | 17819054454 | 云南省昭通市威信县扎西镇红城国际2期16栋2801 |",
f"| {second_order} | 钱 | 17819054455 | 云南省昭通市威信县扎西镇红城国际2期16栋2802 |",
]
)
engine = self.make_engine(run_erp)
answer = engine.handle(
message,
{},
self.identity(),
"c-mixed",
"u1",
"s-mixed",
)
self.assertIn(first_order, answer)
self.assertIn("异常 未设定快递", answer)
self.assertIn("900002", answer)
self.assertIn("跳过 1 条", answer)
latest = engine.latest_batch(
engine.context_key("c-mixed", "u1", "s-mixed")
)
self.assertEqual(latest["stage"], "completed")
self.assertEqual(latest["result"]["status"], "partial")
def test_safe_address_skip_does_not_stop_later_orders(self):
first_order = "110208666749084488"
second_order = "110208666749084489"
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": first_order,
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常 1688一件代发",
},
{
"order_id": second_order,
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常 1688一件代发",
},
],
}
return {
"status": "partial",
"orders": [
{
"order_id": first_order,
"state": "skipped",
"safe_to_continue": True,
"message": "收货地址识别结果不一致,本单已跳过且未保存",
},
{
"order_id": second_order,
"state": "completed",
"purchase_order_no": "900002",
"online_1688_order_no": "5126531473532162000",
"message": "完成",
},
],
}
message = "\n".join(
[
"得物1688自动发货",
"| 订单号 | 收件人 | 手机号 | 收货地址 |",
"| --- | --- | --- | --- |",
f"| {first_order} | 赵 | 17819054454 | 云南省昭通市威信县扎西镇红城国际2期16栋2801 |",
f"| {second_order} | 钱 | 17819054455 | 云南省昭通市威信县扎西镇红城国际2期16栋2802 |",
]
)
engine = self.make_engine(run_erp)
answer = engine.handle(
message,
{},
self.identity(),
"c-safe-skip",
"u1",
"s-safe-skip",
)
self.assertIn("得物1688自动发货结果", answer)
self.assertIn(first_order, answer)
self.assertIn(second_order, answer)
self.assertIn("900002", answer)
self.assertIn("跳过 1 条", answer)
latest = engine.latest_batch(
engine.context_key("c-safe-skip", "u1", "s-safe-skip")
)
self.assertEqual(latest["stage"], "completed")
self.assertEqual(latest["result"]["status"], "partial")
def test_isolated_execution_error_does_not_stop_later_orders(self):
first_order = "110208666749084488"
second_order = "110208666749084489"
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{"order_id": first_order, "state": "ready"},
{"order_id": second_order, "state": "ready"},
],
}
return {
"status": "partial",
"orders": [
{
"order_id": first_order,
"state": "skipped",
"safe_to_continue": True,
"needs_manual_review": True,
"retry_automatically": False,
"purchase_order_no": "793333",
"message": "当前订单执行异常,已隔离待人工复核;后续订单继续处理",
},
{
"order_id": second_order,
"state": "completed",
"purchase_order_no": "793334",
"online_1688_order_no": "5126531473532162001",
"message": "完成",
},
],
}
message = "\n".join(
[
"得物1688自动发货",
"| 订单号 | 收件人 | 手机号 | 收货地址 |",
"| --- | --- | --- | --- |",
f"| {first_order} | 赵 | 17819054454 | 江苏省常州市测试区测试路1号 |",
f"| {second_order} | 钱 | 17819054455 | 江苏省常州市测试区测试路2号 |",
]
)
engine = self.make_engine(run_erp)
answer = engine.handle(
message,
{},
self.identity(),
"c-isolated-error",
"u1",
"s-isolated-error",
)
latest = engine.latest_batch(
engine.context_key("c-isolated-error", "u1", "s-isolated-error")
)
self.assertEqual(latest["stage"], "completed")
self.assertEqual(latest["result"]["status"], "partial")
self.assertEqual(len(latest["result"]["orders"]), 2)
self.assertIn("隔离待人工复核", answer)
self.assertIn("793334", answer)
self.assertIn("跳过 1 条", answer)
def test_existing_purchase_resumes_without_new_confirmation(self):
def run_erp(action, payload):
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"shop_name": "得物-全米",
"order_status": "1688一件代发 / 已生成采购",
"purchase_order_no": "787629",
}
],
}
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "787629",
"online_1688_order_no": "5126531473532161999",
}
],
}
run_erp_mock = mock.Mock(side_effect=run_erp)
engine = self.make_engine(run_erp_mock)
answer = engine.handle(
self.order_message(),
{},
self.identity(),
"c-resume",
"u1",
"s-resume",
)
self.assertIn("787629", answer)
self.assertIn("5126531473532161999", answer)
self.assertEqual(run_erp_mock.call_count, 2)
def test_template_request_does_not_call_erp(self):
run_erp = mock.Mock()
engine = self.make_engine(run_erp)
answer = engine.handle(
"下载得物1688发货模板",
{},
self.identity(),
"c4",
"u1",
"s4",
)
self.assertIn("https://cdn.example.com/dewu-template.xlsx", answer)
run_erp.assert_not_called()
def test_long_batch_streams_progress_and_waits_for_terminal_result(self):
release = threading.Event()
started = threading.Event()
def run_erp(action, payload):
started.set()
release.wait(2)
if action == "apply":
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900001",
"online_1688_order_no": "5126531473532161999",
}
],
}
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
"shop_name": "得物-全米",
"order_status": "异常1688一件代发",
}
],
}
engine = self.make_engine(run_erp)
progress_events = []
response = {}
def run_request():
response["answer"] = engine.handle(
self.order_message(),
{},
self.identity(),
"c-async",
"u1",
"s-async",
progress_callback=progress_events.append,
)
request_thread = threading.Thread(target=run_request)
request_thread.start()
self.assertTrue(started.wait(1))
self.assertTrue(request_thread.is_alive())
latest = engine.latest_batch(engine.context_key("c-async", "u1", "s-async"))
engine.record_progress(
latest["batch_id"],
{
"phase": "inspect",
"completed": 1,
"total": 1,
"current_order_id": "110208666749084488",
"browser_code": "ERP-02",
},
)
deadline = time.time() + 1
while time.time() < deadline and not any(
event.get("browser_code") == "ERP-02" for event in progress_events
):
time.sleep(0.01)
self.assertTrue(
any(event.get("browser_code") == "ERP-02" for event in progress_events)
)
release.set()
request_thread.join(2)
self.assertFalse(request_thread.is_alive())
self.assertIn("得物1688自动发货结果", response["answer"])
self.assertIn("900001", response["answer"])
self.assertNotIn("查询得物1688任务进度", response["answer"])
current = engine.latest_batch(
engine.context_key("c-async", "u1", "s-async")
)
self.assertEqual(current["stage"], "completed")
def test_rapid_progress_steps_are_persisted_and_streamed_in_order(self):
engine = self.make_engine(lambda action, payload: {})
batch = engine._create_batch(
"session:progress-history",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
done = threading.Event()
streamed = []
waiter = threading.Thread(
target=engine._wait_for_job,
args=(batch["batch_id"], done, streamed.append, 0),
)
waiter.start()
expected_codes = [
"address.paste.started",
"address.parse.completed",
"address.save.completed",
"purchase.generate.success",
]
for index, step_code in enumerate(expected_codes):
engine.record_progress(
batch["batch_id"],
{
"phase": "apply",
"completed": 0,
"total": 1,
"current_order_id": "110208666749084488",
"step_code": step_code,
"step_state": "completed" if index else "started",
"message": f"步骤 {index + 1}",
"browser_code": "ERP-01",
},
)
done.set()
waiter.join(2)
self.assertFalse(waiter.is_alive())
persisted = engine.progress_events(batch["batch_id"])
self.assertEqual(
[item["step_code"] for item in persisted],
expected_codes,
)
self.assertEqual(
[
item["step_code"]
for item in streamed
if not item.get("heartbeat")
],
expected_codes,
)
self.assertEqual(
[item["event_id"] for item in persisted],
sorted(item["event_id"] for item in persisted),
)
raw = json.dumps(persisted, ensure_ascii=False)
self.assertNotIn("17819054454", raw)
self.assertNotIn("测试地址", raw)
def test_repeated_browser_queue_progress_is_throttled(self):
engine = self.make_engine(lambda action, payload: {})
batch = engine._create_batch(
"session:queue-progress-dedup",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
event = {
"step_code": "browser.queue.waiting",
"step_state": "queued",
"queue_position": 1,
"browser_slots_busy": 1,
"browser_slots_current": 1,
"browser_slots_max": 1,
"message": "等待专用 ERP 浏览器",
}
first_id = engine.record_progress(batch["batch_id"], event)
second_id = engine.record_progress(batch["batch_id"], event)
self.assertEqual(second_id, first_id)
self.assertEqual(engine.progress_event_count(batch["batch_id"]), 1)
engine.record_progress(batch["batch_id"], dict(event, queue_position=2))
self.assertEqual(engine.progress_event_count(batch["batch_id"]), 2)
def test_progress_explicitly_clears_previous_online_order_number(self):
engine = self.make_engine(lambda action, payload: {})
batch = engine._create_batch(
"session:progress-clear-online",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
engine.record_progress(
batch["batch_id"],
{
"current_order_id": "110208666749084488",
"online_1688_order_no": "5126531473532161999",
"step_code": "supplier.1688.verify.completed",
},
)
engine.record_progress(
batch["batch_id"],
{
"current_order_id": "110208666749084489",
"online_1688_order_no": "",
"step_code": "supplier.1688.batch_verify.order.waiting",
},
)
latest = engine.progress_events(batch["batch_id"])[-1]
self.assertEqual(latest["current_order_id"], "110208666749084489")
self.assertEqual(latest["online_1688_order_no"], "")
def test_blocked_result_is_clearly_labeled_and_keeps_purchase_number(self):
engine = self.make_engine(lambda action, payload: {})
batch = engine._create_batch(
"session:blocked-result",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
engine._update_batch(
batch["batch_id"],
stage="blocked",
result={
"status": "stopped",
"orders": [
{
"order_id": "110208666749084488",
"state": "stopped",
"purchase_order_no": "788535",
"message": "采购单审核确认未完成",
}
],
},
)
answer = engine._render_apply_result(
engine.latest_batch("session:blocked-result")
)
self.assertIn("得物1688自动发货未完成", answer)
self.assertIn("788535", answer)
self.assertIn("成功 0 条,停止 1 条", answer)
def test_restart_recovers_unfinished_order_automatically(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
calls = []
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
batch = first_engine._create_batch(
"session:restart-safe",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
first_engine._update_batch(batch["batch_id"], stage="apply_running")
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
return {
"status": "ok",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "ready",
}
],
}
return {
"status": "ok",
"browser_code": "ERP-01",
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900001",
"online_1688_order_no": "5126531473532161999",
}
],
}
with mock.patch.object(
self.module,
"RECOVERY_AUTO_RESUME_DELAY_SECONDS",
0.01,
):
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
deadline = time.time() + 2
while time.time() < deadline:
recovered = recovered_engine.batch_by_id(batch["batch_id"])
if recovered and recovered["stage"] == "completed":
break
time.sleep(0.01)
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "completed")
self.assertNotIn(
"服务重启中断",
json.dumps(recovered, ensure_ascii=False),
)
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
self.assertTrue(recovered["result"]["recovered_after_restart"])
def test_restart_during_recovery_requeues_and_finishes_remaining_order(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
order_ids = [
"110208666749084488",
"110208666749084489",
]
orders = [
{
"order_id": order_id,
"recipient": "测试",
"phone": "17819054454",
"address": "江苏省常州市测试区测试路1号",
}
for order_id in order_ids
]
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
batch = first_engine._create_batch(
"session:restart-during-recovery",
"u1",
orders,
)
first_engine._update_batch(
batch["batch_id"],
stage="recovery_running",
)
first_engine._record_apply_result(
batch["batch_id"],
{
"orders": [
{
"order_id": order_ids[0],
"state": "completed",
"purchase_order_no": "900020",
"online_1688_order_no": "5126531473532162020",
}
]
},
)
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
self.assertEqual(
[row["order_id"] for row in payload["orders"]],
[order_ids[1]],
)
if action == "inspect":
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": order_ids[1],
"state": "resume_purchase",
"purchase_order_no": "900021",
}
],
}
self.assertEqual(action, "apply")
return {
"status": "ok",
"browser_code": "ERP-02",
"orders": [
{
"order_id": order_ids[1],
"state": "completed",
"purchase_order_no": "900021",
"online_1688_order_no": "5126531473532162021",
}
],
}
with mock.patch.object(
self.module,
"RECOVERY_AUTO_RESUME_DELAY_SECONDS",
0.01,
):
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
deadline = time.time() + 2
while time.time() < deadline:
recovered = recovered_engine.batch_by_id(batch["batch_id"])
if recovered and recovered["stage"] == "completed":
break
time.sleep(0.01)
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "completed")
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
self.assertEqual(
[row["state"] for row in recovered["result"]["orders"]],
["completed", "completed"],
)
progress_codes = {
event["step_code"]
for event in recovered_engine.progress_events(
batch["batch_id"],
limit=100,
)
}
self.assertIn("batch.recovery.queued", progress_codes)
self.assertIn("batch.recovery.started", progress_codes)
self.assertIn("batch.recovery.completed", progress_codes)
def test_restart_uses_strict_progress_evidence_without_repeating_erp(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
batch = first_engine._create_batch(
"session:restart-strict",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
first_engine._update_batch(batch["batch_id"], stage="apply_running")
for step_code in (
"supplier.1688.submission.accepted",
"supplier.1688.dialog.success",
):
first_engine.record_progress(
batch["batch_id"],
{
"phase": "apply",
"current_order_id": "110208666749084488",
"purchase_order_no": "900002",
"step_code": step_code,
"step_state": "completed",
"message": "严格成功证据",
},
)
run_erp = mock.Mock()
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
with mock.patch.object(
recovered_engine,
"_schedule_ack_reconciliation",
):
events = recovered_engine.resume_interrupted_batches()
self.assertTrue(events[batch["batch_id"]].wait(2))
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "completed")
run_erp.assert_not_called()
row = recovered["result"]["orders"][0]
self.assertTrue(row["submission_acknowledged"])
self.assertEqual(
row["completion_evidence"],
"erp_backend_and_success_dialog",
)
def test_restart_uses_existing_online_order_notice_without_repeating_erp(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
order_id = "110208666749084488"
purchase_order_no = "900004"
batch = first_engine._create_batch(
"session:restart-existing-online-order",
"u1",
[
{
"order_id": order_id,
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
first_engine._update_batch(batch["batch_id"], stage="apply_running")
first_engine.record_progress(
batch["batch_id"],
{
"phase": "apply",
"current_order_id": order_id,
"purchase_order_no": purchase_order_no,
"step_code": "supplier.1688.already_completed",
"step_state": "completed",
"message": "ERP确认该采购单已存在线上订单",
},
)
run_erp = mock.Mock()
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
with mock.patch.object(
recovered_engine,
"_schedule_ack_reconciliation",
):
events = recovered_engine.resume_interrupted_batches()
self.assertTrue(events[batch["batch_id"]].wait(2))
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "completed")
run_erp.assert_not_called()
row = recovered["result"]["orders"][0]
self.assertTrue(row["submission_acknowledged"])
self.assertEqual(
row["completion_evidence"],
"erp_existing_online_order_notice",
)
self.assertIn("无需重复下单", row["message"])
def test_restart_only_reconciles_ambiguous_1688_submission(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
batch = first_engine._create_batch(
"session:restart-ambiguous",
"u1",
[
{
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
],
)
first_engine._update_batch(batch["batch_id"], stage="apply_running")
first_engine.record_progress(
batch["batch_id"],
{
"phase": "apply",
"current_order_id": "110208666749084488",
"purchase_order_no": "900003",
"step_code": "supplier.1688.save.started",
"step_state": "started",
"message": "正在提交",
},
)
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "resume_purchase",
"purchase_order_no": "900003",
}
],
}
self.assertEqual(action, "reconcile")
return {
"status": "ok",
"order_id": "110208666749084488",
"purchase_order_no": "900003",
"verified": False,
"online_1688_order_no": "",
}
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
with mock.patch.object(
recovered_engine,
"_schedule_recovery_verification",
):
events = recovered_engine.resume_interrupted_batches()
self.assertTrue(events[batch["batch_id"]].wait(2))
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "recovery_verifying")
self.assertEqual([item[0] for item in calls], ["inspect", "reconcile"])
self.assertNotIn("apply", [item[0] for item in calls])
self.assertIn(
"未重复下单",
recovered["result"]["orders"][0]["message"],
)
def test_restart_mixed_batch_applies_only_unfinished_safe_orders(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
db_path = str(Path(temp.name) / "state.sqlite3")
order_ids = [
"110208666749084488",
"110208666749084489",
"110208666749084490",
]
orders = [
{
"order_id": order_id,
"recipient": "测试",
"phone": "17819054454",
"address": "江苏省常州市测试区测试路1号",
}
for order_id in order_ids
]
first_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=lambda action, payload: {},
template_url=lambda: "",
)
batch = first_engine._create_batch(
"session:restart-mixed",
"u1",
orders,
)
first_engine._update_batch(batch["batch_id"], stage="apply_running")
for step_code in (
"supplier.1688.submission.accepted",
"supplier.1688.dialog.success",
):
first_engine.record_progress(
batch["batch_id"],
{
"phase": "apply",
"current_order_id": order_ids[0],
"purchase_order_no": "900010",
"step_code": step_code,
"step_state": "completed",
},
)
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
if action == "inspect":
self.assertEqual(
[row["order_id"] for row in payload["orders"]],
order_ids[1:],
)
return {
"status": "ok",
"orders": [
{
"order_id": order_ids[1],
"state": "resume_purchase",
"purchase_order_no": "900011",
},
{"order_id": order_ids[2], "state": "ready"},
],
}
self.assertEqual(action, "apply")
self.assertEqual(
[row["order_id"] for row in payload["orders"]],
order_ids[1:],
)
return {
"status": "ok",
"orders": [
{
"order_id": order_ids[1],
"state": "completed",
"purchase_order_no": "900011",
"online_1688_order_no": "5126531473532162011",
},
{
"order_id": order_ids[2],
"state": "completed",
"purchase_order_no": "900012",
"online_1688_order_no": "5126531473532162012",
},
],
}
recovered_engine = self.module.FulfillmentEngine(
db_path=db_path,
run_erp=run_erp,
template_url=lambda: "",
)
with mock.patch.object(
recovered_engine,
"_schedule_ack_reconciliation",
):
events = recovered_engine.resume_interrupted_batches()
self.assertTrue(events[batch["batch_id"]].wait(2))
recovered = recovered_engine.batch_by_id(batch["batch_id"])
self.assertEqual(recovered["stage"], "completed")
self.assertEqual([item[0] for item in calls], ["inspect", "apply"])
self.assertEqual(
[row["state"] for row in recovered["result"]["orders"]],
["completed", "completed", "completed"],
)
def test_restart_serializes_batches_with_overlapping_orders(self):
first_started = threading.Event()
release = threading.Event()
calls = []
def run_erp(action, payload):
calls.append((action, dict(payload)))
first_started.set()
self.assertTrue(release.wait(2))
return {
"status": "ok",
"orders": [
{
"order_id": "110208666749084488",
"state": "blocked",
"message": "测试只读拦截",
}
],
}
engine = self.make_engine(run_erp)
order = {
"order_id": "110208666749084488",
"recipient": "赵",
"phone": "17819054454",
"address": "云南省昭通市威信县扎西镇测试地址",
}
first = engine._create_batch("session:overlap-1", "u1", [order])
second = engine._create_batch("session:overlap-2", "u1", [order])
engine._update_batch(first["batch_id"], stage="recovery_pending")
engine._update_batch(second["batch_id"], stage="recovery_pending")
with mock.patch.object(
engine,
"_schedule_interrupted_batch_resume",
):
started = engine.resume_interrupted_batches()
self.assertEqual(len(started), 1)
self.assertTrue(first_started.wait(1))
self.assertEqual(len(calls), 1)
release.set()
self.assertTrue(next(iter(started.values())).wait(2))
remaining = [
batch_id
for batch_id in (first["batch_id"], second["batch_id"])
if (engine.batch_by_id(batch_id) or {}).get("stage")
== "recovery_pending"
]
self.assertEqual(len(remaining), 1)
class ChatApiIntegrationTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.module = load_module(CHAT_API_PATH, "dewu_1688_chat_api_test")
def test_continue_routes_persisted_batch_without_stealing_other_skill(self):
engine = mock.Mock()
engine.context_key.return_value = "session:continue-route"
engine.latest_batch.return_value = {
"batch_id": "DWCONTINUE01",
"stage": "completed",
}
with mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=engine,
), mock.patch.object(
self.module,
"recall_active_skill_context",
return_value={},
):
self.assertTrue(
self.module.is_dewu_1688_request(
"继续",
{},
"c-continue-route",
"u1",
"s-continue-route",
)
)
with mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=engine,
), mock.patch.object(
self.module,
"recall_active_skill_context",
return_value={"worker": "product-detail-planning-v2"},
):
self.assertFalse(
self.module.is_dewu_1688_request(
"继续",
{},
"c-continue-route",
"u1",
"s-continue-route",
)
)
def test_contextual_dewu_continue_bypasses_generic_task_control(self):
continue_control = ("continue", {})
engine = mock.Mock()
engine.context_key.return_value = "session:c-dewu-continue"
engine.latest_batch.return_value = {
"batch_id": "DWCONTEXT01",
"stage": "blocked",
}
with mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=engine,
), mock.patch.object(
self.module,
"is_dewu_1688_request",
return_value=False,
) as is_dewu:
self.assertTrue(
self.module.should_route_dewu_continue_before_task_control(
continue_control,
"继续",
{},
"c-dewu-continue",
"u1",
"s-dewu-continue",
)
)
is_dewu.assert_not_called()
empty_engine = mock.Mock()
empty_engine.context_key.return_value = "session:c-other-skill"
empty_engine.latest_batch.return_value = None
with mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=empty_engine,
), mock.patch.object(
self.module,
"is_dewu_1688_request",
return_value=False,
):
self.assertFalse(
self.module.should_route_dewu_continue_before_task_control(
continue_control,
"继续",
{},
"c-other-skill",
"u1",
"s-other-skill",
)
)
self.assertFalse(
self.module.should_route_dewu_continue_before_task_control(
("stop", {}),
"停止",
{},
"c-dewu-stop",
"u1",
"s-dewu-stop",
)
)
def test_router_catalog_and_deterministic_priority(self):
candidates = self.module.ai_skill_router_candidate_skills(
"得物1688自动发货",
{},
)
self.assertEqual(candidates[0], "dewu-1688-auto-fulfillment")
with mock.patch.object(
self.module,
"is_dewu_1688_request",
return_value=True,
):
self.assertTrue(
self.module.should_prioritize_deterministic_chat_route(
"得物1688自动发货"
)
)
def test_spreadsheet_attachment_and_otp_redaction(self):
payload = {
"attachments": [
{
"filename": "orders.xlsx",
"url": "https://cdn.example.com/orders.xlsx",
}
]
}
self.assertEqual(len(self.module.dewu_1688_spreadsheet_inputs(payload)), 1)
self.assertEqual(
self.module.dewu_1688_redact_message("验证码 588513"),
"验证码 [已脱敏]",
)
self.assertEqual(
self.module.dewu_1688_redact_message("588513"),
"[短信验证码已脱敏]",
)
redacted = self.module.dewu_1688_redact_message(
"得物1688自动发货\n"
"订单号:110208666749084488\n"
"收件人:赵\n"
"手机号:17819054454\n"
"收货地址:云南省昭通市测试地址"
)
self.assertIn("110208666749084488", redacted)
self.assertIn("订单数:1", redacted)
self.assertNotIn("17819054454", redacted)
self.assertNotIn("测试地址", redacted)
misrouted_table = self.module.dewu_1688_redact_message(
"请分析上传的表格内容\n"
"| 订单号 | 收件人姓名 | 收件人手机号 | 收件人详细地址 |\n"
"| --- | --- | --- | --- |\n"
"| 110208666749084488 | 测试用户 | 17819054454 | 测试地址1号 |"
)
self.assertIn("订单数:1", misrouted_table)
self.assertNotIn("17819054454", misrouted_table)
self.assertNotIn("测试地址1号", misrouted_table)
def test_chat_api_injects_existing_full_company_permission(self):
captured = {}
class FakeEngine:
def handle(self, message, payload, identity, *args):
captured.update(identity)
return "ok"
def context_key(self, *args):
return "fake-context"
def latest_batch(self, context_key):
return None
with (
mock.patch.object(self.module, "is_dewu_1688_request", return_value=True),
mock.patch.object(self.module, "identity_has_full_data_access", return_value=True),
mock.patch.object(self.module, "dewu_1688_engine", return_value=FakeEngine()),
):
answer = self.module.dewu_1688_answer(
"得物1688自动发货",
{},
{"user_id": "u-full", "user_name": "陆南亭"},
"c-auth",
"u-full",
"session-auth",
)
self.assertEqual(answer, "ok")
self.assertTrue(captured["_dewu_1688_authorized"])
def test_chat_api_recalls_active_dewu_context_before_ai_routing(self):
class FakeEngine:
@staticmethod
def context_key(*args):
return "fake-context"
@staticmethod
def latest_batch(context_key):
return None
class FakeModule:
ACTIVE_STAGES = frozenset()
@staticmethod
def is_request(message, active=False):
return active and message == "继续处理"
with (
mock.patch.object(
self.module,
"load_dewu_1688_module",
return_value=FakeModule(),
),
mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=FakeEngine(),
),
mock.patch.object(
self.module,
"recall_active_skill_context",
return_value={"worker": "dewu-1688-auto-fulfillment"},
),
):
self.assertTrue(
self.module.is_dewu_1688_request(
"继续处理",
{},
"c-active-dewu",
"u-active-dewu",
"s-active-dewu",
)
)
def test_chat_api_injects_skill_only_permission_without_company_access(self):
captured = {}
class FakeEngine:
def handle(self, message, payload, identity, *args):
captured.update(identity)
return "ok"
def context_key(self, *args):
return "fake-context"
def latest_batch(self, context_key):
return None
with (
mock.patch.object(self.module, "is_dewu_1688_request", return_value=True),
mock.patch.object(self.module, "identity_has_full_data_access", return_value=False),
mock.patch.object(self.module, "DEWU_1688_ALLOWED_USERS", ("福星", "艳蕊")),
mock.patch.object(self.module, "dewu_1688_engine", return_value=FakeEngine()),
):
answer = self.module.dewu_1688_answer(
"得物1688自动发货",
{},
{"user_id": "u-skill-only", "user_name": "福星"},
"c-skill-only",
"u-skill-only",
"session-skill-only",
)
self.assertEqual(answer, "ok")
self.assertTrue(captured["_dewu_1688_authorized"])
def test_chat_api_rejects_unlisted_user_without_company_access(self):
captured = {}
class FakeEngine:
def handle(self, message, payload, identity, *args):
captured.update(identity)
return "ok"
def context_key(self, *args):
return "fake-context"
def latest_batch(self, context_key):
return None
with (
mock.patch.object(self.module, "is_dewu_1688_request", return_value=True),
mock.patch.object(self.module, "identity_has_full_data_access", return_value=False),
mock.patch.object(self.module, "DEWU_1688_ALLOWED_USERS", ("福星", "艳蕊")),
mock.patch.object(self.module, "dewu_1688_engine", return_value=FakeEngine()),
):
answer = self.module.dewu_1688_answer(
"得物1688自动发货",
{},
{"user_id": "u-not-listed", "user_name": "未授权用户"},
"c-not-listed",
"u-not-listed",
"session-not-listed",
)
self.assertEqual(answer, "ok")
self.assertFalse(captured["_dewu_1688_authorized"])
def test_chat_api_forwards_live_progress_callback(self):
progress_events = []
class FakeEngine:
def handle(
self,
message,
payload,
identity,
*args,
progress_callback=None,
):
self.callback = progress_callback
progress_callback({"display": "正在生成采购单 · 1/2"})
return "ok"
def context_key(self, *args):
return "fake-context"
def latest_batch(self, context_key):
return None
engine = FakeEngine()
with (
mock.patch.object(self.module, "is_dewu_1688_request", return_value=True),
mock.patch.object(self.module, "identity_has_full_data_access", return_value=True),
mock.patch.object(self.module, "dewu_1688_engine", return_value=engine),
):
answer = self.module.dewu_1688_answer(
"得物1688自动发货",
{},
{"user_id": "u-full", "user_name": "陆南亭"},
"c-progress",
"u-full",
"session-progress",
progress_callback=progress_events.append,
)
self.assertEqual(answer, "ok")
self.assertEqual(
progress_events,
[{"display": "正在生成采购单 · 1/2"}],
)
def test_browser_pool_assigns_unique_slots_then_reuses_released_slot(self):
class FakeProcess:
def poll(self):
return None
def fake_start(slot):
slot["process"] = FakeProcess()
return slot
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BATCH_OWNER_KEYS.clear()
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_OWNER_KEYS.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 2),
mock.patch.object(self.module, "DEWU_1688_BROWSER_MAX_COUNT", 2),
mock.patch.object(
self.module,
"_dewu_1688_start_browser_slot",
side_effect=fake_start,
),
):
first = self.module._dewu_1688_acquire_browser_slot("batch-1")
second = self.module._dewu_1688_acquire_browser_slot("batch-2")
self.assertNotEqual(first["code"], second["code"])
acquired = {}
def acquire_third():
acquired["slot"] = self.module._dewu_1688_acquire_browser_slot(
"batch-3"
)
thread = threading.Thread(target=acquire_third)
thread.start()
time.sleep(0.05)
self.assertTrue(thread.is_alive())
self.module._dewu_1688_release_browser_slot(first)
thread.join(1)
self.assertFalse(thread.is_alive())
self.assertEqual(acquired["slot"]["code"], first["code"])
def test_browser_pool_isolates_dedicated_owner_from_shared_slots(self):
class FakeProcess:
def poll(self):
return None
class FakeEngine:
@staticmethod
def batch_by_id(batch_id):
owner = "user-zhang" if "zhang" in batch_id else "user-other"
return {
"batch_id": batch_id,
"owner_user_id": owner,
"stage": "apply_running",
}
@staticmethod
def record_progress(_batch_id, _event):
return None
def fake_start(slot):
slot["process"] = FakeProcess()
return slot
accounts = {
"user-zhang": {
"owner_user_id": "user-zhang",
"browser_code": "ERP-ZHANGHUA",
"username": "dedicated-user",
"password_file": "/test/secret",
}
}
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BATCH_OWNER_KEYS.clear()
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_OWNER_KEYS.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 1),
mock.patch.object(self.module, "DEWU_1688_BROWSER_MAX_COUNT", 1),
mock.patch.object(
self.module,
"DEWU_1688_DEDICATED_ACCOUNTS",
accounts,
),
mock.patch.object(
self.module,
"_dewu_1688_start_browser_slot",
side_effect=fake_start,
),
mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=FakeEngine(),
),
):
dedicated = self.module._dewu_1688_acquire_browser_slot(
"batch-zhang-1"
)
shared = self.module._dewu_1688_acquire_browser_slot(
"batch-other-1"
)
self.assertEqual(dedicated["code"], "ERP-ZHANGHUA")
self.assertEqual(
dedicated["dedicated_owner_user_id"],
"user-zhang",
)
self.assertEqual(shared["code"], "ERP-01")
self.assertFalse(shared["dedicated_owner_user_id"])
self.module._dewu_1688_release_browser_slot(dedicated)
self.module._dewu_1688_release_browser_slot(shared)
def test_failed_browser_warmup_uses_backoff_before_retry(self):
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BATCH_OWNER_KEYS.clear()
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_OWNER_KEYS.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 1),
mock.patch.object(self.module, "DEWU_1688_DEDICATED_ACCOUNTS", {}),
mock.patch.object(
self.module,
"_dewu_1688_start_browser_slot",
side_effect=RuntimeError("browser start failed"),
) as start,
):
first = self.module._dewu_1688_warm_browser_slot("ERP-01")
second = self.module._dewu_1688_warm_browser_slot("ERP-01")
self.assertEqual(first["status"], "failed")
self.assertEqual(second["status"], "backoff")
self.assertEqual(start.call_count, 1)
slot = self.module._DEWU_1688_BROWSER_SLOTS["ERP-01"]
self.assertEqual(slot["warm_failures"], 1)
self.assertGreater(slot["next_warm_at"], time.time())
def test_login_warmup_failure_uses_long_backoff(self):
with mock.patch.object(
self.module,
"DEWU_1688_BROWSER_REWARM_MAX_DELAY",
1800,
):
self.assertEqual(
self.module._dewu_1688_warm_failure_kind(
"Login form remained visible"
),
"login_required",
)
self.assertEqual(
self.module._dewu_1688_warm_retry_delay(
{},
"Login form remained visible",
1,
),
1800,
)
self.assertEqual(
self.module._dewu_1688_warm_retry_delay(
{},
"browser start failed",
1,
),
60,
)
def test_worker_recovers_only_known_or_blank_1688_save_overlays(self):
worker = (
ROOT
/ "dewu-1688-auto-fulfillment"
/ "scripts"
/ "erp_worker.mjs"
).read_text(encoding="utf-8")
self.assertIn("async function hideBlankEpaasPointerOverlays", worker)
self.assertIn("async function click1688SaveWithOverlayRecovery", worker)
self.assertIn("if (!confirmed && !hidden) throw error", worker)
self.assertIn("await click1688SaveWithOverlayRecovery(", worker)
def test_dedicated_browser_reads_restricted_secret_without_affecting_shared_env(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
secret_path = Path(temp.name) / "erp-password"
secret_path.write_text("test-only-secret\n", encoding="utf-8")
secret_path.chmod(0o600)
accounts = {
"user-zhang": {
"owner_user_id": "user-zhang",
"browser_code": "ERP-ZHANGHUA",
"username": "dedicated-user",
"password_file": str(secret_path),
}
}
dedicated_slot = {
"code": "ERP-ZHANGHUA",
"dedicated_owner_user_id": "user-zhang",
}
shared_slot = {
"code": "ERP-01",
"dedicated_owner_user_id": "",
}
with (
mock.patch.object(
self.module,
"DEWU_1688_DEDICATED_ACCOUNTS",
accounts,
),
mock.patch.dict(
os.environ,
{
"DEWU_1688_ERP_USERNAME": "shared-user",
"DEWU_1688_ERP_PASSWORD": "shared-secret",
},
clear=False,
),
):
dedicated_env = self.module._dewu_1688_worker_env_for_slot(
dedicated_slot
)
shared_env = self.module._dewu_1688_worker_env_for_slot(shared_slot)
self.assertEqual(
dedicated_env["DEWU_1688_ERP_USERNAME"],
"dedicated-user",
)
self.assertEqual(
dedicated_env["DEWU_1688_ERP_PASSWORD"],
"test-only-secret",
)
self.assertEqual(shared_env["DEWU_1688_ERP_USERNAME"], "shared-user")
self.assertEqual(shared_env["DEWU_1688_ERP_PASSWORD"], "shared-secret")
secret_path.chmod(0o644)
with self.assertRaisesRegex(RuntimeError, "专用ERP账号凭据不可用"):
self.module._dewu_1688_worker_env_for_slot(dedicated_slot)
def test_browser_pool_scales_on_demand_until_configured_maximum(self):
class FakeProcess:
def poll(self):
return None
def fake_start(slot):
slot["process"] = FakeProcess()
return slot
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 2),
mock.patch.object(self.module, "DEWU_1688_BROWSER_MAX_COUNT", 4),
mock.patch.object(
self.module,
"_dewu_1688_start_browser_slot",
side_effect=fake_start,
),
):
first = self.module._dewu_1688_acquire_browser_slot("batch-scale-1")
second = self.module._dewu_1688_acquire_browser_slot("batch-scale-2")
third = self.module._dewu_1688_acquire_browser_slot("batch-scale-3")
self.assertEqual(
{first["code"], second["code"], third["code"]},
{"ERP-01", "ERP-02", "ERP-03"},
)
status = self.module.dewu_1688_browser_pool_status()
self.assertEqual(status["configured_slots"], 2)
self.assertEqual(status["current_slots"], 3)
self.assertEqual(status["max_slots"], 4)
def test_browser_pool_handles_three_hundred_concurrent_batches_with_twenty_slots(self):
class FakeProcess:
def poll(self):
return None
class FakeEngine:
@staticmethod
def batch_by_id(batch_id):
return {"batch_id": batch_id, "stage": "apply_running"}
@staticmethod
def record_progress(_batch_id, _event):
return None
def fake_start(slot):
slot["process"] = FakeProcess()
return slot
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
acquired = []
failures = []
tracker_lock = threading.Lock()
start_barrier = threading.Barrier(301)
peak_active = 0
def run_batch(index):
nonlocal peak_active
start_barrier.wait()
try:
slot = self.module._dewu_1688_acquire_browser_slot(
f"batch-load-{index:03d}"
)
with tracker_lock:
acquired.append((index, slot["code"]))
active = sum(
1
for item in self.module._DEWU_1688_BROWSER_SLOTS.values()
if item.get("batch_id")
)
peak_active = max(peak_active, active)
time.sleep(0.01)
self.module._dewu_1688_release_browser_slot(slot)
except Exception as exc:
with tracker_lock:
failures.append(str(exc))
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 6),
mock.patch.object(self.module, "DEWU_1688_BROWSER_MAX_COUNT", 20),
mock.patch.object(self.module, "DEWU_1688_QUEUE_TIMEOUT", 30),
mock.patch.object(
self.module,
"_dewu_1688_start_browser_slot",
side_effect=fake_start,
),
mock.patch.object(
self.module,
"dewu_1688_engine",
return_value=FakeEngine(),
),
):
threads = [
threading.Thread(target=run_batch, args=(index,))
for index in range(300)
]
for thread in threads:
thread.start()
start_barrier.wait()
for thread in threads:
thread.join(10)
self.assertFalse(failures)
self.assertEqual(len(acquired), 300)
self.assertEqual(peak_active, 20)
self.assertEqual(len(self.module._DEWU_1688_BROWSER_SLOTS), 20)
self.assertEqual(self.module._DEWU_1688_BATCH_QUEUE, [])
def test_browser_pool_production_defaults_are_six_warm_and_twenty_max(self):
self.assertEqual(self.module.DEWU_1688_BROWSER_COUNT, 6)
self.assertEqual(self.module.DEWU_1688_BROWSER_MAX_COUNT, 20)
def test_browser_queue_fairly_interleaves_three_users_by_active_share(self):
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BATCH_OWNER_KEYS.clear()
for index in range(20):
batch_id = f"user-a-active-{index:02d}"
self.module._DEWU_1688_BROWSER_SLOTS[f"ERP-{index + 1:02d}"] = {
"batch_id": batch_id,
}
self.module._DEWU_1688_BATCH_OWNER_KEYS[batch_id] = "user-a"
queues = {
"user-a": [f"user-a-queued-{index:03d}" for index in range(80)],
"user-b": [f"user-b-queued-{index:03d}" for index in range(150)],
"user-c": [f"user-c-queued-{index:03d}" for index in range(50)],
}
for owner, batch_ids in queues.items():
for batch_id in batch_ids:
self.module._DEWU_1688_BATCH_OWNER_KEYS[batch_id] = owner
self.module._DEWU_1688_BATCH_QUEUE.extend(
queues["user-a"] + queues["user-b"] + queues["user-c"]
)
selected_owners = []
for slot_index in range(12):
slot = self.module._DEWU_1688_BROWSER_SLOTS[
f"ERP-{slot_index + 1:02d}"
]
slot["batch_id"] = ""
selected = self.module._dewu_1688_fair_queue_batch()
owner = self.module._DEWU_1688_BATCH_OWNER_KEYS[selected]
selected_owners.append(owner)
self.module._DEWU_1688_BATCH_QUEUE.remove(selected)
slot["batch_id"] = selected
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_OWNER_KEYS.clear)
self.assertEqual(
selected_owners,
[
"user-b", "user-c", "user-b", "user-c",
"user-b", "user-c", "user-b", "user-c",
"user-b", "user-c", "user-b", "user-c",
],
)
def test_browser_pool_releases_terminal_batch_reservation(self):
class FakeProcess:
def poll(self):
return None
class FakeEngine:
@staticmethod
def batch_by_id(batch_id):
return {"batch_id": batch_id, "stage": "completed"}
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BROWSER_SLOTS["ERP-01"] = {
"code": "ERP-01",
"process": FakeProcess(),
"stdout_queue": None,
"stdout_thread": None,
"batch_id": "batch-terminal",
"updated_at": 1.0,
"io_lock": threading.Lock(),
"browser_ready": True,
"warm_status": "ready",
"warm_error": "",
}
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 1),
mock.patch.object(self.module, "dewu_1688_engine", return_value=FakeEngine()),
):
with condition:
self.module._dewu_1688_prune_browser_slots(now=2.0)
slot = self.module._DEWU_1688_BROWSER_SLOTS["ERP-01"]
self.assertEqual(slot["batch_id"], "")
self.assertIsNotNone(slot["process"])
self.assertTrue(slot["browser_ready"])
def test_browser_pool_keeps_terminal_batch_during_active_reconciliation(self):
class FakeProcess:
def poll(self):
return None
class FakeEngine:
@staticmethod
def batch_by_id(batch_id):
return {"batch_id": batch_id, "stage": "completed"}
condition = self.module._DEWU_1688_SLOT_CONDITION
with condition:
self.module._DEWU_1688_BROWSER_SLOTS.clear()
self.module._DEWU_1688_BATCH_QUEUE.clear()
self.module._DEWU_1688_BROWSER_SLOTS["ERP-01"] = {
"code": "ERP-01",
"process": FakeProcess(),
"stdout_queue": None,
"stdout_thread": None,
"batch_id": "batch-reconcile-active",
"request_active": True,
"updated_at": 1.0,
"io_lock": threading.Lock(),
"browser_ready": True,
"warm_status": "ready",
"warm_error": "",
}
self.addCleanup(self.module._DEWU_1688_BROWSER_SLOTS.clear)
self.addCleanup(self.module._DEWU_1688_BATCH_QUEUE.clear)
with (
mock.patch.object(self.module, "DEWU_1688_BROWSER_COUNT", 1),
mock.patch.object(self.module, "dewu_1688_engine", return_value=FakeEngine()),
):
with condition:
self.module._dewu_1688_prune_browser_slots(now=2.0)
slot = self.module._DEWU_1688_BROWSER_SLOTS["ERP-01"]
self.assertEqual(slot["batch_id"], "batch-reconcile-active")
self.assertTrue(slot["request_active"])
def test_reconcile_request_releases_browser_immediately(self):
class FakeStdin:
def write(self, _value):
return None
def flush(self):
return None
class FakeProcess:
stdin = FakeStdin()
output = queue.Queue()
output.put(
json.dumps(
{
"type": "result",
"result": {
"status": "ok",
"verified": False,
"browser_reset": "order_list",
},
}
)
)
slot = {
"code": "ERP-01",
"process": FakeProcess(),
"stdout_queue": output,
"io_lock": threading.Lock(),
"batch_id": "batch-reconcile-release",
"request_active": True,
"updated_at": time.time(),
"browser_ready": True,
"warm_status": "ready",
}
with (
mock.patch.object(
self.module,
"_dewu_1688_acquire_browser_slot",
return_value=slot,
),
mock.patch.object(
self.module,
"_dewu_1688_release_browser_slot",
) as release,
):
result = self.module.dewu_1688_worker_request(
"reconcile",
{"batch_id": "batch-reconcile-release"},
)
self.assertEqual(result["status"], "ok")
release.assert_called_once_with(slot)
def test_worker_source_resets_to_one_order_window(self):
worker = (
ROOT
/ "dewu-1688-auto-fulfillment"
/ "scripts"
/ "erp_worker.mjs"
).read_text(encoding="utf-8")
self.assertIn("async function resetToOrderList()", worker)
self.assertIn("async function warmBrowserAtOrderList()", worker)
self.assertIn('action === "warmup"', worker)
self.assertIn("context.pages().length !== 1", worker)
self.assertIn('result.browser_reset = "order_list"', worker)
self.assertIn('orderFrame.locator("#__AppendTool_Btn")', worker)
self.assertIn("waitForFrameWithVisibleTexts(", worker)
self.assertIn('["明文", "生成采购单"]', worker)
self.assertIn('preflight.state === "resume_purchase"', worker)
resume_start = worker.index('if (preflight.state === "resume_purchase")')
resume_end = worker.index(
'if (preflight.state !== "ready")',
resume_start,
)
resume_branch = worker[resume_start:resume_end]
self.assertIn("await auditPurchase(", resume_branch)
self.assertIn("await place1688Order(", resume_branch)
self.assertIn('purchaseFrame.locator(".jmx-result-wrapper")', worker)
self.assertIn("orderFrame = await openOrderList();", worker)
self.assertIn("dropshipMatch.frame", worker)
self.assertIn('"#purchaseAddress"', worker)
self.assertIn("await waitFor1688BuyerAddress(", worker)
self.assertIn("1688 delivery address verification failed", worker)
self.assertIn("deliveryAddressMatchDiagnostics(order, currentValues)", worker)
self.assertIn("function withoutPrivacySuffix(value)", worker)
self.assertIn("function canonicalAddressText(", worker)
self.assertIn("normalizedRecipientText(recipient)", worker)
self.assertIn("administrativeParts = []", worker)
self.assertIn("await capture1688DefaultAddressState(", worker)
self.assertIn("supplier.1688.address_guard.completed", worker)
self.assertIn("defaultAddressState,", worker)
self.assertIn("currentFingerprint !== defaultFingerprint", worker)
self.assertIn("await verifyPersistedOrderAddress(", worker)
self.assertIn("Saved address verification failed", worker)
self.assertIn("phoneControlStates", worker)
self.assertIn("const deadline = Date.now() + 5000;", worker)
self.assertIn("Visible phone controls contain conflicting values", worker)
self.assertIn('verificationIssues.push("full_address")', worker)
self.assertIn("等待买家付款|等待卖家发货", worker)
self.assertIn("async function readPurchaseRow(", worker)
self.assertIn("payload.detail_only === true", worker)
self.assertIn("const action = clean(request.action, 40);", worker)
self.assertIn("const minimumComparableLength =", worker)
self.assertIn("administrativeParts.filter", worker)
self.assertIn("detail_log_text", worker)
self.assertIn('const explicitPurchaseOrderNo = cellText("po_id")', worker)
self.assertIn("explicitPurchaseOrderNo.match", worker)
self.assertNotIn("const beforeNumbers = before.match", worker)
self.assertIn('const explicitOrderStatus = cellText("status")', worker)
self.assertIn("text.match(/异常\\s*1688\\s*一件代发/)", worker)
self.assertIn("!row.hasDropshipMarker", worker)
self.assertIn('(row.orderStatus && !row.orderStatus.includes("异常"))', worker)
self.assertIn('row.orderStatus || "1688一件代发"', worker)
self.assertIn("店铺-得物", worker)
self.assertIn("purchase row after direct navigation", worker)
self.assertIn("validated purchase page", worker)
self.assertIn('purchaseFrame.locator("#po_id")', worker)
self.assertIn('purchaseFrame.locator("a.search")', worker)
self.assertIn('"purchase search button"', worker)
self.assertIn("erp_existing_online_order_notice", worker)
self.assertIn("已经存在线上订单号", worker)
self.assertIn("ancestor-or-self::*", worker)
self.assertIn("const matchingRows = rows.filter", worker)
self.assertIn("await recycleBrowserToOrderList();", worker)
self.assertIn("const maxSearchAttempts = 3;", worker)
self.assertIn("attempt === 1 ? 6000 : 10000", worker)
self.assertIn("const maxPurchaseNumberAttempts = 5;", worker)
self.assertIn('"purchase.number.detail.retry"', worker)
self.assertIn("const continuesToApply =", worker)
self.assertIn("await resetToOrderList();", worker)
self.assertIn("function isErpOrderEditorFrame(frame)", worker)
self.assertIn('url.hostname.endsWith(".erp321.com")', worker)
self.assertIn('"/app/order/order/ordereditornew.aspx"', worker)
self.assertIn("async function invokeValidatedPurchaseAudit(", worker)
self.assertIn('"CheckBeforeConfirm"', worker)
self.assertIn("window.ToConfirm(", worker)
self.assertIn('"purchase.audit.submit.validated.completed"', worker)
self.assertIn('candidate.locator("#confirm_confirm")', worker)
self.assertIn('["确认并继续", "确认", "确定", "继续"]', worker)
self.assertIn('/下单成功/.test(body)', worker)
self.assertIn('cellText("status")', worker)
self.assertIn("waitForPurchasePreviewFrame()", worker)
self.assertIn("closeVisibleEpaasDialogs()", worker)
self.assertIn("payload.progress_offset", worker)
self.assertIn("findAuditConfirmationControl()", worker)
self.assertIn("purchase.audit.confirm.native.completed", worker)
self.assertIn("purchase.audit.confirm.not_required", worker)
self.assertIn("purchase.audit.status.verified", worker)
self.assertIn("Purchase audit confirmation was not completed", worker)
self.assertIn("auditConfirmationDiagnostics()", worker)
self.assertIn("audit-controls-${purchaseOrderNo}", worker)
self.assertIn("clearPurchaseResultOverlays(", worker)
self.assertIn("purchase.audit.result.closed", worker)
self.assertIn("confirm1688DialogsAfterSave(", worker)
self.assertIn("supplier.1688.save_confirm.completed", worker)
self.assertIn("create1688NativeDialogGuard(", worker)
self.assertIn("supplier.1688.native_confirm.completed", worker)
self.assertIn("async function setLegacyCheckboxChecked(", worker)
self.assertIn("if (!(await locator.isChecked()))", worker)
self.assertIn("async function install1688AsyncRecorder(", worker)
self.assertIn('"PurchaseTo1688V3"', worker)
self.assertIn("window.jQuery.cfm = function", worker)
self.assertIn(
"请确认一键下单更改成一键代发",
worker,
)
self.assertIn("window.jQuery.msg = function", worker)
self.assertIn("exact1688SuccessMessage(", worker)
self.assertIn("latestRecorderState.confirmations", worker)
self.assertIn("if (response.isComfrimFenXiao)", worker)
self.assertIn("inspect1688SubmissionRecords(", worker)
self.assertIn("supplier.1688.submission.accepted", worker)
self.assertIn("submissionAcknowledged: true", worker)
self.assertIn(
'completionEvidence: "erp_backend_and_success_dialog"',
worker,
)
self.assertIn("async function reconcile1688Order(", worker)
self.assertIn('action === "reconcile"', worker)
self.assertIn("supplier.1688.background_verify.completed", worker)
self.assertIn(
"1688 submission confirmation was not received",
worker,
)
self.assertIn("supplier.1688.verify.started", worker)
self.assertIn("supplier.1688.verify.completed", worker)
self.assertIn("supplier.1688.verify.deferred", worker)
self.assertIn("verification_pending", worker)
self.assertIn("verifyPending1688Orders(", worker)
self.assertIn("supplier.1688.batch_verify.started", worker)
self.assertIn("supplier.1688.batch_verify.order.completed", worker)
self.assertIn(
"process.env.DEWU_1688_BATCH_VERIFY_TIMEOUT_MS || 180000",
worker,
)
self.assertIn(
"process.env.DEWU_1688_BATCH_VERIFY_POLL_INTERVAL_MS || 30000",
worker,
)
self.assertIn(
"process.env.DEWU_1688_BATCH_VERIFY_MAX_ATTEMPTS || 3",
worker,
)
self.assertIn(
'["completed", "verification_pending"].includes(result.state)',
worker,
)
self.assertIn("1688线上单号和下单状态复核通过", worker)
self.assertIn(
"1688线上单号暂未确认,本单已停止并转后续核对",
worker,
)
self.assertNotIn("submittedPending: true", worker)
self.assertNotIn("待回写", worker)
self.assertNotIn("线上单号待异步回写", worker)
self.assertIn("1688 success dialog purchase number mismatch", worker)
self.assertIn("waitFor1688AccountReady(", worker)
self.assertIn("supplier.1688.account.waiting", worker)
self.assertIn("/下单成功/.test(visibleDialogBody)", worker)
self.assertIn("is not defined", worker)
self.assertIn("referenceerror", worker)
self.assertIn("Existing purchase number was not loaded", worker)
self.assertIn("const maxVerificationAttempts = 5", worker)
self.assertIn("purchaseAuditPrimaryControl(", worker)
self.assertIn("invokePurchaseAuditFunction(", worker)
self.assertIn("clickPurchaseAuditDropdownAction(", worker)
self.assertIn("async function purchaseRowLocator(", worker)
self.assertIn("async function selectPurchaseRowForAudit(", worker)
self.assertIn("attempt <= 3", worker)
self.assertIn("Purchase row selection was not isolated", worker)
self.assertIn("await locator.check();", worker)
self.assertIn("read1688DeliveryAddressState(", worker)
self.assertIn("purchase.audit.submit.function.completed", worker)
self.assertIn("purchase.audit.submit.fallback.completed", worker)
self.assertIn('action === "inspect_1688_nav"', worker)
self.assertIn(
'for (const label of [\n "全部确认审核",',
worker,
)
self.assertIn("await auditTarget.click({ force: true })", worker)
self.assertGreaterEqual(
worker.count(
"if (!(await frameChainVisible(candidate).catch(() => false))) continue;"
),
4,
)
self.assertIn(
'candidate.getByText(label, { exact: true })',
worker,
)
self.assertIn('"全部确认审核",\n "全部审核"', worker)
self.assertIn("orderContext.purchase_order_no", worker)
self.assertIn("function safeOrderFailure(message)", worker)
self.assertIn("const deadline = Date.now() + 5000", worker)
self.assertIn("address.recipient.restore.started", worker)
self.assertIn("await recipientControl.fill(restoredRecipient)", worker)
self.assertIn("error?.safeToContinue === true", worker)
self.assertIn("const verificationPending = Boolean(", worker)
self.assertIn("needs_manual_review: needsManualReview", worker)
self.assertIn("retry_automatically: false", worker)
self.assertIn("后续订单继续处理", worker)
self.assertIn("采购单行定位暂不稳定", worker)
self.assertNotIn("当前订单执行失败,已停止后续订单", worker)
self.assertNotIn("采购单审核确认未完成,已停止后续订单", worker)
self.assertIn(
"浏览器复位失败,当前工作器无法继续后续订单",
worker,
)
self.assertIn(
'result.state === "skipped" && result.safe_to_continue',
worker,
)
self.assertIn('? "partial"', worker)
self.assertIn(
'waitForOneVisible(\n detailFrame.locator("#input_receiver_address")',
worker,
)
self.assertIn(
'waitForOneVisible(\n detailFrame.locator("#save_btn")',
worker,
)
self.assertIn("purchase_order_no: orderContext.purchase_order_no", worker)
for step_code in (
"auth.check.started",
"order.search.started",
"address.parse.completed",
"address.save.completed",
"purchase.generate.success",
"purchase.audit.success",
"supplier.1688.order.success",
"browser.reset.completed",
):
self.assertIn(step_code, worker)
def test_worker_address_normalization_accepts_dewu_privacy_format(self):
worker = (
ROOT
/ "dewu-1688-auto-fulfillment"
/ "scripts"
/ "erp_worker.mjs"
).read_text(encoding="utf-8")
start = worker.index("function normalizedContactText")
end = worker.index("const SUBMISSION_RISK_STEPS", start)
helper_source = worker[start:end]
persisted_start = worker.index(
"function administrativePartMatchesAddress"
)
persisted_end = worker.index(
"async function verifyPersistedOrderAddress",
persisted_start,
)
persisted_helper_source = worker[persisted_start:persisted_end]
delivery_start = worker.index(
"function deliveryAddressMatchDiagnostics"
)
delivery_end = worker.index(
"async function waitFor1688BuyerAddress",
delivery_start,
)
delivery_helper_source = worker[delivery_start:delivery_end]
cases = [
[
"北京北京市通州区永顺镇永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]",
"北京市北京市通州区永顺镇西潞苑二街李庄佳苑小满寓公寓A区A",
["北京市", "北京市", "通州区"],
],
[
"山西省朔州市朔城区北城街道山西省朔州市朔城区华源万和城A区[9792]",
"山西省朔州市朔城区北城街道华源万和城A区",
["山西省", "朔州市", "朔城区"],
],
[
"黑龙江省齐齐哈尔市泰来县泰来镇铁小区门卫[5774]",
"黑龙江省齐齐哈尔市泰来县泰来镇铁小区门卫",
["黑龙江省", "齐齐哈尔市", "泰来县"],
],
]
script = f"""
const vm = require('vm');
const context = {{}};
vm.createContext(context);
vm.runInContext({json.dumps(helper_source)}, context);
vm.runInContext({json.dumps(persisted_helper_source)}, context);
vm.runInContext({json.dumps(delivery_helper_source)}, context);
const cases = {json.dumps(cases, ensure_ascii=False)};
for (const item of cases) {{
if (!context.addressTextMatches(...item)) process.exit(2);
}}
if (context.normalizedRecipientText('张亚楠[9792]') !== '张亚楠') process.exit(3);
if (context.erpRecipientText('谢bro[1439]') !== '谢') process.exit(28);
if (context.erpRecipientText('陈先生[5178]') !== '陈先生') process.exit(29);
if (context.erpRecipientText('a316[9139]') !== 'a316') process.exit(30);
if (context.normalizedPhoneDigits('15786948868-9139') !== '15786948868') process.exit(6);
if (context.privacyPhoneSuffix('15786948868-9139') !== '9139') process.exit(7);
if (!context.maskedRecipientMatchesExpected('张*', '张三')) process.exit(21);
if (!context.maskedRecipientMatchesExpected('*三', '张三')) process.exit(22);
if (!context.maskedRecipientMatchesExpected('张*三', '张小三')) process.exit(23);
if (!context.maskedRecipientMatchesExpected('******', '张三')) process.exit(24);
if (context.maskedRecipientMatchesExpected('李*', '张三')) process.exit(25);
const maskedIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: 'a316[9139]', address: cases[0][0]}},
{{recipient: '******', phone: '******', address: '******', state: '北京市', city: '北京市', district: '通州区'}},
);
if (maskedIssues.length) process.exit(4);
const singleMaskedIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: '张三', address: cases[0][0]}},
{{recipient: '张*', phone: '******', address: '******', state: '北京市', city: '北京市', district: '通州区'}},
);
if (singleMaskedIssues.length) process.exit(26);
const wrongSingleMaskedIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: '张三', address: cases[0][0]}},
{{recipient: '李*', phone: '******', address: '******', state: '北京市', city: '北京市', district: '通州区'}},
);
if (!wrongSingleMaskedIssues.includes('recipient')) process.exit(27);
const chineseOnlyRecipientIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: '谢bro[1439]', address: cases[0][0], __verifiedDelivery: {{recipient: '谢'}}}},
{{recipient: '谢', phone: '******', address: '******', state: '北京市', city: '北京市', district: '通州区'}},
);
if (chineseOnlyRecipientIssues.length) process.exit(31);
const persistedMainPhoneIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: 'a316[9139]', address: cases[0][0]}},
{{recipient: 'a316', phone: '15786948868', address: '永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]', state: '北京市', city: '北京市', district: '通州区'}},
);
if (persistedMainPhoneIssues.length) process.exit(8);
const hiddenPersistedPhone = {{
recipient: 'a316',
phone: '',
phoneControlsPresent: 2,
phoneControlsAllEmpty: true,
address: '永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
state: '北京市',
city: '北京市',
district: '通州区',
}};
if (!context.persistedPhoneIsPermissionHidden(
{{
phone: '15786948868-9139',
__verifiedDelivery: {{phone: '15786948868'}},
}},
hiddenPersistedPhone,
)) process.exit(12);
if (context.persistedPhoneIsPermissionHidden(
{{
phone: '15786948868-9139',
__verifiedDelivery: {{phone: '18466515322'}},
}},
hiddenPersistedPhone,
)) process.exit(13);
const missingPrivacySuffixIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: 'a316[9139]', address: cases[0][0]}},
{{recipient: 'a316', phone: '15786948868', address: '永顺镇西潞苑二街李庄佳苑小满寓公寓A区A', state: '北京市', city: '北京市', district: '通州区'}},
);
if (!missingPrivacySuffixIssues.includes('privacy_suffix')) process.exit(9);
const previewOrder = {{
phone: '15786948868-9139',
recipient: 'a316[9139]',
address: cases[0][0],
__verifiedDelivery: {{
recipient: 'a316',
phone: '15786948868',
address: '北京市北京市通州区永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
}},
}};
if (!context.contactTextMatchesOrder(
previewOrder,
'a316 15786948868 北京市通州区永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
)) process.exit(10);
if (context.contactTextMatchesOrder(
previewOrder,
'a316 15786948868 北京市通州区永顺镇西潞苑二街李庄佳苑小满寓公寓A区A',
)) process.exit(11);
const separateDeliveryControls = [
'a316',
'15786948868',
'北京市',
'北京市',
'通州区',
'永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
];
const companyDefaultControls = [
'测试公司',
'13800138000',
'浙江省杭州市余杭区公司地址',
];
if (!context.deliveryAddressStateMatchesOrder(
previewOrder,
separateDeliveryControls,
companyDefaultControls,
)) process.exit(14);
const privacyHiddenDeliveryControls = [
'a316',
'北京市',
'北京市',
'通州区',
'永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
];
if (!context.deliveryAddressStateMatchesOrder(
previewOrder,
privacyHiddenDeliveryControls,
companyDefaultControls,
)) process.exit(16);
const privacyHiddenRecipientAndPhoneControls = [
'北京市',
'北京市',
'通州区',
'永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
];
if (!context.deliveryAddressStateMatchesOrder(
previewOrder,
privacyHiddenRecipientAndPhoneControls,
companyDefaultControls,
)) process.exit(21);
if (!context.deliveryAddressStateMatchesOrder(
previewOrder,
['a316'],
companyDefaultControls,
'a316 北京市通州区永顺镇西潞苑二街李庄佳苑小满寓公寓A区A[9139]',
'测试公司 浙江省杭州市余杭区公司地址',
)) process.exit(18);
if (context.deliveryAddressStateMatchesOrder(
previewOrder,
privacyHiddenDeliveryControls.map((value) => value.replace('9139', '0000')),
companyDefaultControls,
)) process.exit(17);
const standardDeliveryOrder = {{
phone: '17819054454',
recipient: '赵',
address: '云南省昭通市威信县扎西镇红城国际2期16栋2801',
}};
if (!context.deliveryAddressStateMatchesOrder(
standardDeliveryOrder,
['赵', '17819054454', '云南省昭通市威信县扎西镇红城国际2期16栋2801'],
companyDefaultControls,
)) process.exit(19);
if (context.deliveryAddressStateMatchesOrder(
standardDeliveryOrder,
['赵', '17819054454'],
companyDefaultControls,
)) process.exit(20);
if (context.deliveryAddressStateMatchesOrder(
previewOrder,
separateDeliveryControls,
separateDeliveryControls,
)) process.exit(15);
const wrongRegionIssues = context.orderAddressFormVerificationIssues(
{{phone: '15786948868-9139', recipient: 'a316[9139]', address: cases[0][0]}},
{{recipient: '******', phone: '******', address: '******', state: '浙江省', city: '杭州市', district: '余杭区'}},
);
if (!wrongRegionIssues.includes('state')) process.exit(5);
"""
completed = subprocess.run(
["node", "-e", script],
capture_output=True,
text=True,
check=False,
)
self.assertEqual(completed.returncode, 0, completed.stderr)
def test_chat_api_exposes_sse_and_progress_history_contract(self):
source = CHAT_API_PATH.read_text(encoding="utf-8")
self.assertIn('"X-Chat-Progress-Version", "2"', source)
self.assertIn('"request.accepted"', source)
self.assertIn('"/api/chat/dewu-progress"', source)
self.assertIn('"progress_events"', source)
self.assertIn('"progress_event_count"', source)
def test_worker_output_queue_keeps_rapid_progress_and_final_result(self):
output_queue = self.module.queue.Queue()
output_queue.put('{"type":"progress","step_code":"one"}\n')
output_queue.put('{"type":"progress","step_code":"two"}\n')
output_queue.put('{"type":"result","result":{"status":"ok"}}\n')
slot = {"stdout_queue": output_queue}
lines = [
self.module._dewu_1688_read_worker_line(slot, 0.1)
for _ in range(3)
]
self.assertEqual(
[json.loads(line)["type"] for line in lines],
["progress", "progress", "result"],
)
def test_chat_api_starts_and_reports_browser_warmup(self):
source = CHAT_API_PATH.read_text(encoding="utf-8")
engine_source = ENGINE_PATH.read_text(encoding="utf-8")
self.assertTrue(self.module.DEWU_1688_WARM_ON_START)
self.assertIn("start_dewu_1688_browser_warmup()", source)
self.assertIn("_schedule_interrupted_batch_resume()", engine_source)
self.assertIn('"browser_ready": bool(slot.get("browser_ready"))', source)
self.assertIn('"warm_status": str(slot.get("warm_status")', source)
def test_chat_api_skips_generic_feedback_for_operational_reply(self):
self.assertFalse(
self.module.should_append_feedback_prompt(
"得物1688自动发货预检结果。" * 20,
"dewu-1688-auto-fulfillment",
"completed",
)
)
source = CHAT_API_PATH.read_text(encoding="utf-8")
self.assertIn(
'if worker in {"product-stock-db", "dewu-1688-auto-fulfillment"}:',
source,
)
def test_full_result_csv_contains_every_order(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
batch = {
"batch_id": "DWRESULT01",
"browser_code": "ERP-03",
"result": {
"orders": [
{
"order_id": "110208666749084488",
"state": "completed",
"purchase_order_no": "900001",
"online_1688_order_no": "5126531473532161907",
"message": "完成",
},
{
"order_id": "110208666749084489",
"state": "stopped",
"message": "状态不匹配",
},
{
"order_id": "110208666749084490",
"state": "completed",
"purchase_order_no": "900003",
"online_1688_order_no": "",
"submission_acknowledged": True,
"submission_purchase_order_no": "900003",
"completion_evidence": (
"erp_backend_and_success_dialog"
),
"message": "完成",
},
]
},
}
with (
mock.patch.object(self.module, "DATA_EXPORT_LOCAL_DIR", temp.name),
mock.patch.object(
self.module,
"data_export_upload_file",
return_value=("https://cdn.example.com/result.csv", {}),
),
):
self.module._DEWU_1688_RESULT_URLS.clear()
url = self.module.dewu_1688_result_download_url(batch)
self.assertEqual(url, "https://cdn.example.com/result.csv")
csv_path = Path(temp.name) / "得物1688发货结果-DWRESULT01.csv"
content = csv_path.read_text(encoding="utf-8-sig")
self.assertIn("110208666749084488", content)
self.assertIn("110208666749084489", content)
self.assertIn("110208666749084490", content)
self.assertTrue(
"ERP已确认下单成功" in content
or "110208666749084490,900003,,完成,完成" in content
)
self.assertIn("ERP-03", content)
if __name__ == "__main__":
unittest.main()