113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
"""快递100自动送达状态测试。
|
||
|
||
测试根据物流单号查询快递100状态，自动更新订单为已送达的逻辑。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import patch
|
||
import pytest
|
||
|
||
|
||
@pytest.mark.logistics
|
||
@pytest.mark.p0
|
||
class TestAutoDelivery:
|
||
"""测试快递100自动送达逻辑。"""
|
||
|
||
def test_auto_deliver_on_kuaidi100_signed(self, make_order, db_session):
|
||
"""当快递100返回已签收且无待结算提成时，订单应自动更新为已完成。"""
|
||
order = make_order(order_status="in_transit", commission_amount=0)
|
||
|
||
with patch('backend.app.services.logistics_service.LogisticsService._load_third_party_traces') as mock_load:
|
||
mock_load.return_value = [
|
||
{
|
||
"node_time": "2026-07-03 10:00:00",
|
||
"node_desc": "已签收",
|
||
"node_type": "signed",
|
||
"source_platform": "kuaidi100",
|
||
}
|
||
]
|
||
|
||
from backend.app.services.logistics_service import LogisticsService
|
||
logistics_service = LogisticsService()
|
||
logistics_service.check_delivery_by_tracking(db_session)
|
||
|
||
db_session.refresh(order)
|
||
assert order.order_status == "completed"
|
||
|
||
def test_auto_deliver_with_commission_moves_to_pending_settle(self, make_order, db_session):
|
||
"""当快递100返回已签收且有提成时，订单应进入待结算。"""
|
||
from backend.app.services.logistics_service import LogisticsService
|
||
|
||
order = make_order(order_status="in_transit", commission_amount=100)
|
||
|
||
with patch.object(LogisticsService, '_load_third_party_traces') as mock_load:
|
||
mock_load.return_value = [
|
||
{
|
||
"node_time": "2026-07-03 10:00:00",
|
||
"node_desc": "已签收",
|
||
"node_type": "signed",
|
||
"source_platform": "kuaidi100",
|
||
}
|
||
]
|
||
|
||
logistics_service = LogisticsService()
|
||
logistics_service.check_delivery_by_tracking(db_session)
|
||
|
||
db_session.refresh(order)
|
||
assert order.order_status == "pending_settle"
|
||
|
||
def test_auto_deliver_not_triggered_when_no_signed_status(self, make_order, db_session):
|
||
"""当快递100返回非已签收状态时，不应触发自动送达。"""
|
||
from backend.app.services.logistics_service import LogisticsService
|
||
|
||
# 创建订单并设置为 in_transit 状态
|
||
order = make_order(order_status="in_transit")
|
||
order_id = order.id
|
||
|
||
# Mock 快递100 API 返回在途中状态
|
||
with patch.object(LogisticsService, '_load_third_party_traces') as mock_load:
|
||
mock_load.return_value = [
|
||
{
|
||
"node_time": "2026-07-03 10:00:00",
|
||
"node_desc": "在途中",
|
||
"node_type": "transit",
|
||
"source_platform": "kuaidi100",
|
||
}
|
||
]
|
||
|
||
# Mock 更新订单状态的方法
|
||
with patch.object(LogisticsService, '_update_order_after_delivery') as mock_update:
|
||
logistics_service = LogisticsService()
|
||
logistics_service.check_delivery_by_tracking(db_session)
|
||
|
||
# 验证 _update_order_after_delivery 未被调用
|
||
mock_update.assert_not_called()
|
||
|
||
def test_saved_delivered_trace_skips_third_party_query(self, make_order, db_session):
|
||
"""本地已有签收轨迹时，查询轨迹不应再次调用第三方。"""
|
||
from datetime import datetime
|
||
|
||
from backend.app.services.logistics_service import LogisticsService
|
||
|
||
order = make_order(order_status="pending_settle", commission_amount=100)
|
||
logistics_service = LogisticsService()
|
||
logistics_service.logistics_repository.create_trace(
|
||
db_session,
|
||
{
|
||
"order_id": order.id,
|
||
"task_id": None,
|
||
"node_time": datetime(2026, 7, 3, 10, 0, 0),
|
||
"node_desc": "已签收",
|
||
"node_type": "signed",
|
||
"source_platform": "kuaidi100",
|
||
"remark": "",
|
||
},
|
||
)
|
||
db_session.commit()
|
||
|
||
with patch.object(LogisticsService, '_load_third_party_traces') as mock_load:
|
||
result = logistics_service.get_trace(order.id, db_session)
|
||
|
||
mock_load.assert_not_called()
|
||
assert result["trace_list"][0]["node_desc"] == "已签收"
|