批量换货脚本

scripts/batch_exchange/test_batch_exchange.py

@@ -0,0 +1,172 @@
+"""批量换货脚本的输入校验测试。"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SCRIPT_PATH = Path(__file__).with_name("batch_exchange.py")
+SPEC = importlib.util.spec_from_file_location("batch_exchange", SCRIPT_PATH)
+assert SPEC is not None and SPEC.loader is not None
+batch_exchange = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = batch_exchange
+SPEC.loader.exec_module(batch_exchange)
+
+
+class LoadExchangesTest(unittest.TestCase):
+    """验证 CSV 映射校验不会把危险批次交给接口执行。"""
+
+    def write_csv(self, content: str) -> Path:
+        """在临时目录中创建待解析的 CSV。"""
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        path = Path(directory.name) / "exchanges.csv"
+        path.write_text(content, encoding="utf-8")
+        return path
+
+    def test_loads_supported_header_and_rows(self) -> None:
+        """支持标准表头并保留原始行号。"""
+        path = self.write_csv("old_identifier,new_identifier\nold-1,new-1\nold-2,new-2\n")
+
+        exchanges = batch_exchange.load_exchanges(path)
+
+        self.assertEqual(
+            exchanges,
+            [
+                batch_exchange.ExchangeInput(2, "old-1", "new-1"),
+                batch_exchange.ExchangeInput(3, "old-2", "new-2"),
+            ],
+        )
+
+    def test_rejects_duplicate_old_identifier(self) -> None:
+        """同一旧资产不能在一批中换出两次。"""
+        path = self.write_csv("old-1,new-1\nold-1,new-2\n")
+
+        with self.assertRaisesRegex(ValueError, "旧资产.*重复"):
+            batch_exchange.load_exchanges(path)
+
+    def test_rejects_duplicate_new_identifier(self) -> None:
+        """同一新资产不能接收两次换货。"""
+        path = self.write_csv("old-1,new-1\nold-2,new-1\n")
+
+        with self.assertRaisesRegex(ValueError, "新资产.*重复"):
+            batch_exchange.load_exchanges(path)
+
+    def test_rejects_same_identifier_in_one_row(self) -> None:
+        """禁止资产换给自身。"""
+        path = self.write_csv("old-1,old-1\n")
+
+        with self.assertRaisesRegex(ValueError, "新旧资产标识相同"):
+            batch_exchange.load_exchanges(path)
+
+    def test_rejects_cross_row_asset_reuse(self) -> None:
+        """禁止资产在同批中同时作为旧资产和新资产。"""
+        path = self.write_csv("old-1,new-1\nnew-1,new-2\n")
+
+        with self.assertRaisesRegex(ValueError, "同时作为.*旧资产.*新资产"):
+            batch_exchange.load_exchanges(path)
+
+    def test_rejects_missing_column_value(self) -> None:
+        """双列中的空值必须在请求前被拦截。"""
+        path = self.write_csv("old_identifier,new_identifier\nold-1,\n")
+
+        with self.assertRaisesRegex(ValueError, "均不能为空"):
+            batch_exchange.load_exchanges(path)
+
+
+class BuildExchangePayloadTest(unittest.TestCase):
+    """验证脚本不能关闭直接换货或数据迁移。"""
+
+    def test_forces_direct_flow_and_migration(self) -> None:
+        """请求固定为 direct 且 migrate_data 为 true。"""
+        exchange = batch_exchange.ExchangeInput(2, "old-1", "new-1")
+
+        payload = batch_exchange.build_exchange_payload(
+            exchange,
+            "iot_card",
+            "批量直接换货",
+            None,
+        )
+
+        self.assertEqual(payload["flow_type"], "direct")
+        self.assertIs(payload["migrate_data"], True)
+        self.assertEqual(payload["old_asset_type"], "iot_card")
+
+
+class ExchangeAssetTest(unittest.TestCase):
+    """验证结果必须同时满足接口成功和迁移完成。"""
+
+    class FakeClient:
+        """返回指定响应的换货客户端替身。"""
+
+        def __init__(self, result: object) -> None:
+            self.result = result
+
+        def create_exchange(self, *args: object) -> object:
+            """返回测试预设的接口响应。"""
+            return self.result
+
+    def test_marks_completed_migration_as_success(self) -> None:
+        """业务成功且迁移完成时结果为成功。"""
+        result = batch_exchange.HTTPResult(
+            status=200,
+            body={
+                "code": 0,
+                "msg": "success",
+                "data": {
+                    "id": 123,
+                    "exchange_no": "EX123",
+                    "migration_completed": True,
+                    "migration_balance": 100,
+                },
+            },
+            raw_body="",
+        )
+
+        outcome = batch_exchange.exchange_asset(
+            self.FakeClient(result),
+            "token",
+            batch_exchange.ExchangeInput(2, "old-1", "new-1"),
+            "iot_card",
+            "批量直接换货",
+            None,
+        )
+
+        self.assertTrue(outcome.success)
+        self.assertEqual(outcome.exchange_no, "EX123")
+
+    def test_rejects_success_response_without_completed_migration(self) -> None:
+        """业务成功但未确认迁移完成时结果仍为失败。"""
+        result = batch_exchange.HTTPResult(
+            status=200,
+            body={
+                "code": 0,
+                "msg": "success",
+                "data": {
+                    "id": 123,
+                    "exchange_no": "EX123",
+                    "migration_completed": False,
+                    "migration_balance": 0,
+                },
+            },
+            raw_body="",
+        )
+
+        outcome = batch_exchange.exchange_asset(
+            self.FakeClient(result),
+            "token",
+            batch_exchange.ExchangeInput(2, "old-1", "new-1"),
+            "iot_card",
+            "批量直接换货",
+            None,
+        )
+
+        self.assertFalse(outcome.success)
+        self.assertIn("未确认数据迁移完成", outcome.message)
+
+
+if __name__ == "__main__":
+    unittest.main()