mcp/tests/test_gateway_query_order.py

import unittest
from unittest.mock import patch

from app import GatewayApp
from services.api_client import ApiClient, JsonTransport
from services.token_store import InMemoryTokenStore
from tools.query_order import QueryOrderTool


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

    def post_json(self, url, payload, headers, timeout):
        self.calls.append(
            {
                'url': url,
                'payload': payload,
                'headers': headers,
                'timeout': timeout,
            }
        )
        if url.endswith('/mcp/tools/listEnabledTools'):
            return {
                'code': 'MCP_0000',
                'msg': 'success',
                'data': {
                    'tool_codes': [
                        'query_order',
                        'query_track',
                        'query_order_exact',
                        'list_order_filter_options',
                    ],
                },
            }
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': 'ok',
                'records': [],
                'tips': [],
            },
            'meta': {
                'request_id': headers.get('X-Request-Id', ''),
            },
        }



class BomHttpResponse:
    def __init__(self, body):
        self.body = body

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        return False

    def read(self):
        return self.body
class GatewayQueryOrderTest(unittest.TestCase):
    def test_gateway_lists_query_order_tool(self):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        store.save('MT_demo', '2099-01-01T00:00:00')
        app = GatewayApp(api_client=ApiClient(
            base_url='http://tools.example.test',
            token_store=store,
            transport=DummyTransport(),
        ))

        tools = app.list_tools()

        by_name = {tool['name']: tool for tool in tools}
        self.assertIn('query_order', by_name)
        self.assertIn('query_order_exact', by_name)
        self.assertIn('list_order_filter_options', by_name)
        self.assertIn('keyword', by_name['query_order']['input_schema']['required'])

    def test_json_transport_accepts_utf8_bom_response(self):
        transport = JsonTransport()
        body = b'\xef\xbb\xbf{"code":"MCP_0000","data":{"summary":"ok"}}'

        with patch('urllib.request.urlopen', return_value=BomHttpResponse(body)):
            response = transport.post_json(
                'http://tools.example.test/mcp/tools/queryOrder',
                {'keyword': 'SO20260706001'},
                {},
                8,
            )

        self.assertEqual('MCP_0000', response['code'])
        self.assertEqual('ok', response['data']['summary'])
    def test_api_client_adds_mcp_headers_when_calling_tool(self):
        transport = DummyTransport()
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        store.save('MT_demo', '2099-01-01T00:00:00')
        client = ApiClient(
            base_url='http://tools.example.test',
            token_store=store,
            transport=transport,
            timeout=8,
        )

        response = client.call_tool(
            tool_code='query_order',
            route_path='/mcp/tools/queryOrder',
            payload={'keyword': 'SO20260706001', 'page': 1, 'limit': 20},
            request_id='rq_demo',
        )

        self.assertEqual('MCP_0000', response['code'])
        self.assertEqual(1, len(transport.calls))
        self.assertEqual('Bearer MT_demo', transport.calls[0]['headers']['Authorization'])
        self.assertEqual('query_order', transport.calls[0]['headers']['X-MCP-Tool-Code'])
        self.assertEqual('rq_demo', transport.calls[0]['headers']['X-Request-Id'])
        self.assertEqual('http://tools.example.test/mcp/tools/queryOrder', transport.calls[0]['url'])

    def test_api_client_lists_enabled_tools_without_tool_headers(self):
        transport = DummyTransport()
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        store.save('MT_demo', '2099-01-01T00:00:00')
        client = ApiClient(
            base_url='http://tools.example.test',
            token_store=store,
            transport=transport,
            timeout=8,
        )

        response = client.list_enabled_tools('rq_enabled_tools')

        self.assertEqual('MCP_0000', response['code'])
        self.assertEqual(
            'http://tools.example.test/mcp/tools/listEnabledTools',
            transport.calls[0]['url'],
        )
        self.assertEqual({}, transport.calls[0]['payload'])
        self.assertEqual({
            'Authorization': 'Bearer MT_demo',
            'X-Request-Id': 'rq_enabled_tools',
        }, transport.calls[0]['headers'])

    def test_query_order_tool_normalizes_input_before_forwarding(self):
        transport = DummyTransport()
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        store.save('MT_demo', '2099-01-01T00:00:00')
        client = ApiClient(
            base_url='http://tools.example.test',
            token_store=store,
            transport=transport,
            timeout=8,
        )
        tool = QueryOrderTool(api_client=client)

        tool.call(keyword=' SO20260706001 ', page=2, limit=15, request_id='rq_tool')

        self.assertEqual(
            {
                'keyword': 'SO20260706001',
                'page': 2,
                'limit': 15,
            },
            transport.calls[0]['payload'],
        )
        self.assertEqual('http://tools.example.test/mcp/tools/queryOrder', transport.calls[0]['url'])


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