mcp/tests/test_cli_and_file_store.py

import io
import json
import os
import tempfile
import unittest

from app import GatewayApp
from services.token_store import FileTokenStore


class DummyAuthClient:
    def __init__(self, token_store):
        self.token_store = token_store
        self.exchange_calls = []

    def exchange(self, auth_code):
        self.exchange_calls.append(auth_code)
        self.token_store.save('MT_bound', '2099-01-01T00:00:00')
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'mcp_token': 'MT_bound',
                'expire_time': '2099-01-01T00:00:00',
            },
        }


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

    def call_tool(self, tool_code, route_path, payload, request_id):
        self.calls.append(
            {
                'tool_code': tool_code,
                'route_path': route_path,
                'payload': payload,
                'request_id': request_id,
            }
        )
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': 'ok',
                'records': [],
                'tips': [],
            },
            'meta': {
                'request_id': request_id,
            },
        }


class CliAndFileStoreTest(unittest.TestCase):
    def test_file_token_store_persists_session_across_instances(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            first = FileTokenStore(path, refresh_skew_seconds=60)
            first.save('MT_file', '2099-01-01T00:00:00')

            second = FileTokenStore(path, refresh_skew_seconds=60)

            self.assertEqual('MT_file', second.get()['token'])
            second.clear()
            self.assertIsNone(second.get())
            self.assertFalse(os.path.exists(path))

    def test_run_cli_bind_and_call_emit_json(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            token_store = FileTokenStore(path, refresh_skew_seconds=60)
            auth_client = DummyAuthClient(token_store)
            api_client = DummyApiClient()
            app = GatewayApp(
                auth_client=auth_client,
                api_client=api_client,
                token_store=token_store,
            )

            bind_stdout = io.StringIO()
            bind_code = app.run_cli(['bind', '--auth-code', 'AUTH123'], stdout=bind_stdout)
            bind_payload = json.loads(bind_stdout.getvalue())

            call_stdout = io.StringIO()
            call_code = app.run_cli([
                'call',
                '--tool', 'query_order',
                '--keyword', 'SO20260706001',
                '--page', '2',
                '--limit', '15',
            ], stdout=call_stdout)
            call_payload = json.loads(call_stdout.getvalue())

            self.assertEqual(0, bind_code)
            self.assertEqual('MCP_0000', bind_payload['code'])
            self.assertEqual(0, call_code)
            self.assertEqual('MCP_0000', call_payload['code'])
            self.assertEqual(['AUTH123'], auth_client.exchange_calls)
            self.assertEqual('query_order', api_client.calls[0]['tool_code'])
            self.assertEqual({'keyword': 'SO20260706001', 'page': 2, 'limit': 15}, api_client.calls[0]['payload'])

    def test_run_cli_query_track_accepts_tracking_number(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            token_store = FileTokenStore(path, refresh_skew_seconds=60)
            token_store.save('MT_bound', '2099-01-01T00:00:00')
            api_client = DummyApiClient()
            app = GatewayApp(
                auth_client=DummyAuthClient(token_store),
                api_client=api_client,
                token_store=token_store,
            )

            stdout = io.StringIO()
            exit_code = app.run_cli([
                'call',
                '--tool', 'query_track',
                '--tracking-number', '1471904540000000301',
                '--page', '1',
                '--limit', '20',
            ], stdout=stdout)

            self.assertEqual(0, exit_code)
            self.assertEqual('query_track', api_client.calls[0]['tool_code'])
            self.assertEqual('/mcp/tools/queryTrack', api_client.calls[0]['route_path'])
            self.assertEqual('1471904540000000301', api_client.calls[0]['payload']['tracking_number'])
            self.assertNotIn('order_id', api_client.calls[0]['payload'])
            self.assertNotIn('order_number', api_client.calls[0]['payload'])
    def test_run_cli_serve_stdio_handles_initialize_request(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            token_store = FileTokenStore(path, refresh_skew_seconds=60)
            token_store.save('MT_bound', '2099-01-01T00:00:00')
            app = GatewayApp(
                auth_client=DummyAuthClient(token_store),
                api_client=DummyApiClient(),
                token_store=token_store,
            )
            stdin = io.StringIO(
                json.dumps(
                    {
                        'jsonrpc': '2.0',
                        'id': 1,
                        'method': 'initialize',
                        'params': {
                            'protocolVersion': '2025-06-18',
                            'capabilities': {},
                            'clientInfo': {'name': 'workbuddy', 'version': '1.0.0'},
                        },
                    }
                )
                + '\n'
            )
            stdout = io.StringIO()

            exit_code = app.run_cli(['serve-stdio'], stdin=stdin, stdout=stdout)
            response = json.loads(stdout.getvalue().strip())

            self.assertEqual(0, exit_code)
            self.assertEqual(1, response['id'])
            self.assertEqual('2025-06-18', response['result']['protocolVersion'])

    def test_serve_public_command_is_registered(self):
        app = GatewayApp(auth_client=None, api_client=None, token_store=None)

        with self.assertRaises(SystemExit) as error:
            app.run_cli(['serve-public', '--help'])

        self.assertEqual(0, error.exception.code)

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