junhong_cmp_fiber/scripts/batch_exchange/test_batch_exchange.py

"""批量换货脚本的输入校验测试。"""
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()