mcp/tests/test_query_goods_info_tool.py

import io
import unittest
from unittest.mock import Mock, patch

from app import GatewayApp
from services.output_presenter import OutputPresenter
from tools.query_goods_info import QueryGoodsInfoTool


def goods_columns():
    return [
        {'key': key, 'name': name}
        for key, name in OutputPresenter.GOODS_INFO_COLUMNS
    ]


def goods_record():
    return {key: key + '-v' for key, _name in OutputPresenter.GOODS_INFO_COLUMNS}


class QueryGoodsInfoToolTest(unittest.TestCase):
    def test_call_forwards_sku_name_and_page(self):
        client = Mock()
        client.call_tool.return_value = {'code': 'MCP_0000'}
        QueryGoodsInfoTool(client).call(
            sku=' SKU-1 ',
            cn_name=' 衬衫 ',
            page=2,
            limit=100,
            request_id='rq_goods',
            ignored=True,
        )
        client.call_tool.assert_called_once_with(
            'query_goods_info',
            '/mcp/tools/queryGoodsInfo',
            {
                'page': 2,
                'limit': 100,
                'sku': 'SKU-1',
                'cn_name': '衬衫',
            },
            'rq_goods',
        )

    def test_call_accepts_either_sku_or_name(self):
        client = Mock()
        client.call_tool.return_value = {'code': 'MCP_0000'}
        tool = QueryGoodsInfoTool(client)
        tool.call(sku='SKU-1', cn_name='   ')
        self.assertEqual({'page': 1, 'limit': 20, 'sku': 'SKU-1'}, client.call_tool.call_args[0][2])
        tool.call(cn_name='衬衫', sku=None)
        self.assertEqual(
            {'page': 1, 'limit': 20, 'cn_name': '衬衫'},
            client.call_tool.call_args[0][2],
        )

    def test_call_requires_api_client_and_a_lookup_field(self):
        with self.assertRaises(RuntimeError):
            QueryGoodsInfoTool().call(sku='SKU-1')
        tool = QueryGoodsInfoTool(api_client=Mock())
        with self.assertRaises(ValueError):
            tool.call()
        with self.assertRaises(ValueError):
            tool.call(sku='', cn_name='')

    def test_call_rejects_invalid_fields(self):
        tool = QueryGoodsInfoTool(api_client=Mock())
        with self.assertRaises(ValueError):
            tool.call(sku=1)
        with self.assertRaises(ValueError):
            tool.call(cn_name=1)
        with self.assertRaises(ValueError):
            tool.call(sku='S' * 2001)
        with self.assertRaises(ValueError):
            tool.call(cn_name='名' * 101)
        for page in (True, '1', 0, 101):
            with self.subTest(page=page):
                with self.assertRaises(ValueError):
                    tool.call(sku='SKU-1', page=page)
        with self.assertRaises(ValueError):
            tool.call(sku='SKU-1', limit=0)

    def test_cli_forwards_sku_or_name(self):
        app = GatewayApp()
        stdout = io.StringIO()
        with self.assertRaises(ValueError):
            app.run_cli(['call', '--tool', 'query_goods_info'], stdout=stdout)
        with patch.object(app, 'call_tool', return_value={'code': 'MCP_0000'}) as call_tool:
            self.assertEqual(0, app.run_cli([
                'call', '--tool', 'query_goods_info', '--sku', 'SKU-1',
            ], stdout=stdout))
            self.assertEqual('SKU-1', call_tool.call_args[0][1]['sku'])
            self.assertNotIn('cn_name', call_tool.call_args[0][1])
            self.assertEqual(0, app.run_cli([
                'call', '--tool', 'query_goods_info', '--cn-name', '衬衫',
            ], stdout=stdout))
            self.assertEqual('衬衫', call_tool.call_args[0][1]['cn_name'])
            self.assertNotIn('sku', call_tool.call_args[0][1])
            self.assertEqual(0, app.run_cli([
                'call', '--tool', 'query_goods_info',
                '--sku', 'SKU-1', '--cn-name', '衬衫',
            ], stdout=stdout))
            payload = call_tool.call_args[0][1]
            self.assertEqual('SKU-1', payload['sku'])
            self.assertEqual('衬衫', payload['cn_name'])

    def test_presenter_renders_goods_table(self):
        record = goods_record()
        result = OutputPresenter().present('query_goods_info', {
            'code': 'MCP_0000',
            'data': {'columns': goods_columns(), 'records': [record]},
            'meta': {'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_g'},
        })
        self.assertFalse(result['is_error'])
        self.assertEqual(24, len(result['structured_content']['headers']))
        self.assertEqual('sku-v', result['structured_content']['rows'][0][0])
        self.assertIn('SKU', result['text'])
        empty = OutputPresenter().present('query_goods_info', {
            'code': 'MCP_0000',
            'data': {'columns': goods_columns(), 'records': []},
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        })
        self.assertFalse(empty['is_error'])
        self.assertEqual([], empty['structured_content']['rows'])

    def test_presenter_rejects_invalid_goods_payload(self):
        presenter = OutputPresenter()
        columns = goods_columns()
        record = goods_record()
        meta = {'page': 1, 'limit': 20, 'has_more': False}
        bad_column = list(columns)
        bad_column[0] = {'key': 'sku', 'name': '错'}
        bad_record = dict(record)
        bad_record['sku'] = 1
        extra_record = dict(record)
        extra_record['secret'] = 'x'
        cases = [
            {'columns': columns},
            {'columns': 'bad', 'records': []},
            {'columns': columns, 'records': None},
            {'columns': columns[:1], 'records': []},
            {'columns': columns, 'records': []},
            {'columns': bad_column, 'records': []},
            {'columns': columns, 'records': ['x']},
            {'columns': columns, 'records': [extra_record]},
            {'columns': columns, 'records': [bad_record]},
        ]
        metas = [meta, meta, meta, meta, {'request_id': 'rq'}, meta, meta, meta, meta]
        for data, raw_meta in zip(cases, metas):
            with self.subTest(data=data):
                result = presenter.present('query_goods_info', {
                    'code': 'MCP_0000',
                    'data': data,
                    'meta': raw_meta,
                })
                self.assertTrue(result['is_error'])
                self.assertNotIn('secret', result['text'])
                self.assertNotIn('sku-v', result['text'])