127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""mappers 状态映射单测。
|
|
|
|
运行:uv run python -m unittest discover -s tests -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from flows.shared import mappers
|
|
from flows.shared.mappers import (
|
|
map_riin_goods,
|
|
map_riin_goods_status,
|
|
map_riin_system_status,
|
|
map_sds_goods_status,
|
|
map_sds_system_status,
|
|
)
|
|
|
|
TZ_CN = timezone(timedelta(hours=8))
|
|
|
|
|
|
class WarnUnknownTest(unittest.TestCase):
|
|
def setUp(self):
|
|
mappers._warned_values.clear()
|
|
|
|
def _capture(self):
|
|
return io.StringIO()
|
|
|
|
def test_未知值告警且同值只报一次(self):
|
|
buf = self._capture()
|
|
with redirect_stdout(buf):
|
|
self.assertIsNone(map_riin_goods_status("SOMETHING_NEW"))
|
|
self.assertIsNone(map_riin_goods_status("SOMETHING_NEW")) # 第二次不再报
|
|
lines = [ln for ln in buf.getvalue().splitlines() if "SOMETHING_NEW" in ln]
|
|
self.assertEqual(len(lines), 1)
|
|
self.assertIn("[mapper] unknown RIIN goodsStatus", lines[0])
|
|
|
|
|
|
class RiinGoodsStatusTest(unittest.TestCase):
|
|
def setUp(self):
|
|
mappers._warned_values.clear()
|
|
|
|
def test_全部已知枚举可映射(self):
|
|
cases = {
|
|
"NOT_SHIPPED": "待发货",
|
|
"SHIPPED": "已发货",
|
|
"CANCEL": "已取消",
|
|
"CANCELLED": "已取消",
|
|
"REFUNDING": "退款中",
|
|
"COMPLETE": "已完成", # 本次补齐(曾导致 NULL)
|
|
"PAUSE": "暂停", # 本次补齐(曾导致 NULL)
|
|
"CLOSE": "已取消", # 回填重拉时告警发现的新枚举
|
|
}
|
|
for raw, want in cases.items():
|
|
with self.subTest(raw=raw):
|
|
self.assertEqual(map_riin_goods_status(raw), want)
|
|
|
|
def test_空值返回None(self):
|
|
self.assertIsNone(map_riin_goods_status(None))
|
|
self.assertIsNone(map_riin_goods_status(""))
|
|
|
|
def test_map_riin_goods_接线_PAUSE与COMPLETE不再落NULL(self):
|
|
# 复现线上订单 GSU1S555400MFLL:4 行商品 goods_status=PAUSE/COMPLETE
|
|
created = datetime(2026, 9, 14, 9, 39, 24, tzinfo=TZ_CN)
|
|
base = dict(
|
|
platform_oid="GSU1S555400MFLL", product_code="DG001-BK",
|
|
color_code="BK", size_code="L", style_code=None, num=1,
|
|
)
|
|
g1 = map_riin_goods(SimpleNamespace(id="P1", goods_status="PAUSE", **base),
|
|
"GSU1S555400MFLL", created)
|
|
g2 = map_riin_goods(SimpleNamespace(id="P2", goods_status="COMPLETE", **base),
|
|
"GSU1S555400MFLL", created)
|
|
self.assertEqual(g1.status, "暂停")
|
|
self.assertEqual(g2.status, "已完成")
|
|
self.assertEqual(g1.productionNo, "P1")
|
|
self.assertEqual(g1.quantity, 1)
|
|
self.assertEqual(g1.itemName, "DG001")
|
|
|
|
|
|
class RiinSystemStatusTest(unittest.TestCase):
|
|
def setUp(self):
|
|
mappers._warned_values.clear()
|
|
|
|
def test_已知与空值(self):
|
|
self.assertEqual(map_riin_system_status(12), "已发货")
|
|
self.assertEqual(map_riin_system_status(-4), "待推送")
|
|
self.assertIsNone(map_riin_system_status(None))
|
|
|
|
def test_未知值返回None并告警(self):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
self.assertIsNone(map_riin_system_status(999))
|
|
self.assertIn("unknown RIIN systemStatus", buf.getvalue())
|
|
|
|
|
|
class SdsStatusTest(unittest.TestCase):
|
|
def setUp(self):
|
|
mappers._warned_values.clear()
|
|
|
|
def test_商品状态含已删除(self):
|
|
self.assertEqual(map_sds_goods_status(99), "已删除")
|
|
self.assertEqual(map_sds_goods_status(98), "已取消")
|
|
self.assertEqual(map_sds_goods_status(4), "已完成")
|
|
self.assertIsNone(map_sds_goods_status(None))
|
|
|
|
def test_未知商品状态返回None并告警(self):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
self.assertIsNone(map_sds_goods_status(6))
|
|
self.assertIn("unknown SDS goodsStatus", buf.getvalue())
|
|
|
|
def test_订单状态与未知告警(self):
|
|
self.assertEqual(map_sds_system_status(3), "备货中")
|
|
self.assertIsNone(map_sds_system_status(None))
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
self.assertIsNone(map_sds_system_status(42))
|
|
self.assertIn("unknown SDS systemStatus", buf.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|