mcp/tests/test_fuel_rate_tools.py

import unittest

from tools.query_fuel_rate_list import QueryFuelRateListTool
from tools.save_fuel_rate import SaveFuelRateTool
from tools.enable_fuel_rate import EnableFuelRateTool
from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter


class FuelRateToolTest(unittest.TestCase):
    def test_tools_are_registered_on_base_backend(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertIn('query_fuel_rate_list', local)
        self.assertIn('save_fuel_rate', local)
        self.assertIn('enable_fuel_rate', local)
        self.assertEqual('base', QueryFuelRateListTool.backend)
        self.assertEqual('base', SaveFuelRateTool.backend)
        self.assertEqual('base', EnableFuelRateTool.backend)

    def test_query_posts_exact_filters(self):
        client = FakeClient()
        QueryFuelRateListTool(api_client=client).call(
            fuel_no=' RY12 ',
            product='美国专线',
            price_type='销售燃油',
            status=3,
            start_date='2026-10-21',
            page=2,
            limit=20,
            request_id='rq_fuel_list',
        )
        self.assertEqual('query_fuel_rate_list', client.calls[0][0])
        self.assertEqual('/mcp/tools/queryFuelRateList', client.calls[0][1])
        self.assertEqual({
            'page': 2,
            'limit': 20,
            'fuel_no': 'RY12',
            'product': '美国专线',
            'price_type': '销售燃油',
            'status': '3',
            'start_date': '2026-10-21',
        }, client.calls[0][2])

    def test_save_posts_condition_item(self):
        client = FakeClient()
        SaveFuelRateTool(api_client=client).call(
            price_type=1,
            start_date='2026-10-21',
            items=[{
                'fuel_ratio': 0.15,
                'conditions': [{
                    'condition': '物流产品',
                    'group': '1',
                    'condition_value': '美国专线',
                }],
            }],
            request_id='rq_fuel_save',
        )
        self.assertEqual('/mcp/tools/saveFuelRate', client.calls[0][1])
        self.assertEqual({
            'price_type': '1',
            'start_date': '2026-10-21',
            'items': [{
                'fuel_ratio': '0.1500',
                'conditions': [{
                    'condition': '物流产品',
                    'group': '1',
                    'condition_value': '美国专线',
                }],
            }],
        }, client.calls[0][2])

    def test_save_edit_posts_fuel_no(self):
        client = FakeClient()
        SaveFuelRateTool(api_client=client).call(
            fuel_no=' RY12 ',
            price_type='销售燃油',
            start_date='2026-10-22',
            items=[{
                'fuel_ratio': '0.2000',
                'conditions': [{
                    'condition': '物流产品',
                    'group': '1',
                    'condition_value': '美国专线',
                }],
            }],
        )
        self.assertEqual('RY12', client.calls[0][2]['fuel_no'])
        self.assertEqual('销售燃油', client.calls[0][2]['price_type'])
        self.assertEqual(1, len(client.calls[0][2]['items']))

    def test_save_rejects_invalid_fuel_ratio(self):
        tool = SaveFuelRateTool(api_client=FakeClient())
        item = {
            'conditions': [{
                'condition': '物流产品',
                'group': '1',
                'condition_value': '美国专线',
            }],
        }
        for ratio in ('abc', '1.2', '0.12345', True):
            with self.assertRaises(ValueError):
                tool.call(
                    price_type=1,
                    start_date='2026-10-22',
                    items=[dict(item, fuel_ratio=ratio)],
                )

    def test_save_edit_rejects_multiple_items(self):
        tool = SaveFuelRateTool(api_client=FakeClient())
        item = {
            'fuel_ratio': '0.2000',
            'conditions': [{
                'condition': '物流产品',
                'group': '1',
                'condition_value': '美国专线',
            }],
        }
        with self.assertRaises(ValueError):
            tool.call(
                fuel_no='RY12',
                price_type=1,
                start_date='2026-10-22',
                items=[item, item],
            )

    def test_presenter_keeps_missing_condition_message(self):
        result = OutputPresenter().present('save_fuel_rate', {
            'code': 'MCP_1401',
            'msg': '第1条物流产品条件值[美国专线]不存在或未启用',
            'data': {},
            'meta': {'request_id': 'rq_fuel_save'},
        })
        self.assertIn('不存在或未启用', str(result))

    def test_presenter_shows_fuel_list(self):
        columns = [
            {'key': key, 'name': name}
            for key, name in (
                ('fuel_no', '燃油代码'),
                ('price_type_name', '燃油类型'),
                ('status_name', '状态'),
                ('source', '来源'),
            )
        ]
        result = OutputPresenter().present('query_fuel_rate_list', {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': '共查询到 1 条燃油系数',
                'columns': columns,
                'records': [{
                    'fuel_no': 'RY12',
                    'price_type_name': '销售燃油',
                    'status_name': '启用',
                    'source': 'MCP',
                }],
                'tips': [],
            },
            'meta': {'request_id': 'rq_fuel_list', 'page': 1, 'limit': 20, 'total': 1},
        })
        rendered = str(result)
        self.assertIn('RY12', rendered)
        self.assertIn('MCP', rendered)

    def test_enable_posts_fuel_numbers(self):
        client = FakeClient()
        EnableFuelRateTool(api_client=client).call(fuel_no=[' RY12 ', 'RY13', 'RY12'])
        self.assertEqual('enable_fuel_rate', client.calls[0][0])
        self.assertEqual('/mcp/tools/enableFuelRate', client.calls[0][1])
        self.assertEqual(['RY12', 'RY13'], client.calls[0][2]['fuel_no'])

    def test_enable_rejects_empty_list(self):
        with self.assertRaises(ValueError):
            EnableFuelRateTool(api_client=FakeClient()).call(fuel_no=[])

    def test_presenter_keeps_already_enabled_message(self):
        result = OutputPresenter().present('enable_fuel_rate', {
            'code': 'MCP_1401',
            'msg': '燃油系数RY12已启用',
            'data': {},
            'meta': {'request_id': 'rq_fuel_enable'},
        })
        self.assertIn('已启用', str(result))

    def test_presenter_shows_enabled_fuel(self):
        result = OutputPresenter().present('enable_fuel_rate', {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': '已启用 1 条燃油系数',
                'columns': [
                    {'key': 'fuel_no', 'name': '燃油代码'},
                    {'key': 'status_text', 'name': '状态'},
                ],
                'records': [{'fuel_no': 'RY12', 'status_text': '启用'}],
                'tips': [],
            },
            'meta': {'request_id': 'rq_fuel_enable'},
        })
        rendered = str(result)
        self.assertIn('RY12', rendered)
        self.assertIn('启用', rendered)

    def test_tools_expose_metadata_and_reject_bad_input(self):
        self.assertIn('fuel_no', QueryFuelRateListTool().metadata()['input_schema']['properties'])
        self.assertIn('items', SaveFuelRateTool().metadata()['input_schema']['properties'])
        self.assertIn('fuel_no', EnableFuelRateTool().metadata()['input_schema']['properties'])
        with self.assertRaises(RuntimeError):
            QueryFuelRateListTool().call()
        with self.assertRaises(RuntimeError):
            SaveFuelRateTool().call(price_type='1', start_date='2026-10-22', items=[])
        with self.assertRaises(RuntimeError):
            EnableFuelRateTool().call(fuel_no='RY12')

        query = QueryFuelRateListTool(api_client=FakeClient())
        for kwargs in (
            {'page': True},
            {'page': 0},
            {'limit': 101},
            {'product': 1},
            {'channel': 'C' * 101},
            {'price_type': True},
            {'status': '停用'},
        ):
            with self.assertRaises(ValueError):
                query.call(**kwargs)
        query.call(product='   ', price_type=2, status=1)

        save = SaveFuelRateTool(api_client=FakeClient())
        good = {
            'fuel_ratio': '0',
            'conditions': [{
                'condition': '物流产品',
                'group': '1',
                'condition_value': '美国专线',
                'condition_relation': 'and',
            }],
        }
        for kwargs in (
            {'price_type': True, 'start_date': '2026-10-22', 'items': [good]},
            {'price_type': '销售', 'start_date': '2026-10-22', 'items': [good]},
            {'price_type': '1', 'start_date': '', 'items': [good]},
            {'price_type': '1', 'start_date': '2026-10-22', 'end_date': 'x' * 31, 'items': [good]},
            {'price_type': '1', 'start_date': '2026-10-22', 'fuel_no': 1, 'items': [good]},
            {'price_type': '1', 'start_date': '2026-10-22', 'fuel_no': ' ', 'items': [good]},
            {'price_type': '1', 'start_date': '2026-10-22', 'fuel_no': 'R' * 21, 'items': [good]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': {}},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': []},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [good] * 21},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': ['0.1']},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'extra': 1}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '   ', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '1.2.3', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '2', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0./', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.a', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '1.001', 'conditions': good['conditions']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': {}}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': []}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': [good['conditions'][0]] * 31}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': ['物流产品']}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': [{'condition': '物流产品', 'group': '1', 'condition_value': '美国专线', 'extra': 1}]}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': [{'condition': '', 'group': '1', 'condition_value': '美国专线'}]}]},
            {'price_type': '1', 'start_date': '2026-10-22', 'items': [{'fuel_ratio': '0.1', 'conditions': [{'condition': '物流产品', 'group': '1', 'condition_value': '美国专线', 'condition_relation': 'xor'}]}]},
        ):
            with self.assertRaises(ValueError):
                save.call(**kwargs)
        client = FakeClient()
        SaveFuelRateTool(api_client=client).call(
            price_type='1',
            start_date='2026-10-22',
            end_date='2026-12-31',
            items=[{
                'fuel_ratio': 1,
                'conditions': [{
                    'condition': '物流产品',
                    'group': '1',
                    'condition_value': '美国专线',
                    'condition_relation': 'or',
                }],
            }, {
                'fuel_ratio': '1.0000',
                'conditions': good['conditions'],
            }],
        )
        self.assertEqual('2026-12-31', client.calls[0][2]['end_date'])
        self.assertEqual('1.0000', client.calls[0][2]['items'][0]['fuel_ratio'])

        enable = EnableFuelRateTool(api_client=FakeClient())
        enable.call(fuel_no=' RY12 ')
        for value in (1, [1], [' '], ['R' * 21], ['RY1'] * 21):
            with self.assertRaises(ValueError):
                enable.call(fuel_no=value)

        presenter = OutputPresenter()
        for message in ('', 'x' * 401, 'has sql', 'Exception', 'a\\b', '没有标记'):
            result = presenter.present('save_fuel_rate', {
                'code': 'MCP_1401',
                'msg': message,
                'data': {},
                'meta': {'request_id': 'rq_fuel_save'},
            })
            self.assertIn('参数不正确', str(result))


class FakeClient:
    def __init__(self):
        self.calls = []

    def call_tool(self, name, route, payload, request_id):
        self.calls.append((name, route, payload, request_id))
        return {'ok': True}


if __name__ == '__main__':
    unittest.main()