mcp/tests/test_customer_unverified_bill_details_tool.py

import json
import unittest
from io import StringIO

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.query_customer_unverified_bill_details import (
    QueryCustomerUnverifiedBillDetailsTool,
)


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

    def call_tool(self, tool_code, route_path, payload, request_id):
        self.calls.append((tool_code, route_path, payload, request_id))
        return {'code': 'MCP_0000', 'data': {}}

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['query_customer_unverified_bill_details']},
        }


class CustomerUnverifiedBillDetailsToolTest(unittest.TestCase):
    def test_schema_is_closed_requires_customer_and_forwards_call(self):
        client = RecordingApiClient()
        tool = QueryCustomerUnverifiedBillDetailsTool(client)
        schema = tool.metadata()['input_schema']

        self.assertEqual(['customer_id'], schema['required'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual({'customer_id', 'page', 'limit'}, set(schema['properties']))
        result = tool.call(customer_id=7, page=2, limit=30, request_id='rq_detail')
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual((
            'query_customer_unverified_bill_details',
            '/mcp/tools/queryCustomerUnverifiedBillDetails',
            {'customer_id': 7, 'page': 2, 'limit': 30},
            'rq_detail',
        ), client.calls[0])

    def test_invalid_arguments_and_missing_client_fail(self):
        tool = QueryCustomerUnverifiedBillDetailsTool(RecordingApiClient())
        for arguments in (
            {'customer_id': True}, {'customer_id': 0}, {'customer_id': '7'},
            {'customer_id': 7, 'page': 0}, {'customer_id': 7, 'page': 101},
            {'customer_id': 7, 'limit': False}, {'customer_id': 7, 'limit': 101},
        ):
            with self.subTest(arguments=arguments):
                with self.assertRaises(ValueError):
                    tool.call(**arguments)
        with self.assertRaises(RuntimeError):
            QueryCustomerUnverifiedBillDetailsTool().call(customer_id=7)

    def test_registries_and_cli_include_detail_tool(self):
        client = RecordingApiClient()
        local = GatewayApp(api_client=client)
        public = PublicGatewayApp(None, None)
        self.assertEqual(local.registered_tool_names(), public.registered_tool_names())
        self.assertEqual(19, len(local.registered_tool_names()))
        self.assertIn('query_customer_unverified_bill_details', local.registered_tool_names())

        output = StringIO()
        local.run_cli([
            'call', '--tool', 'query_customer_unverified_bill_details',
            '--customer-id', '7', '--page', '2', '--limit', '30',
        ], stdout=output)
        self.assertEqual('MCP_0000', json.loads(output.getvalue())['code'])
        self.assertEqual({'customer_id': 7, 'page': 2, 'limit': 30}, client.calls[-1][2])

        with self.assertRaises(ValueError):
            local.run_cli([
                'call', '--tool', 'query_customer_unverified_bill_details',
            ], stdout=StringIO())


class CustomerUnverifiedBillDetailsPresenterTest(unittest.TestCase):
    KEYS = [
        'bill_month', 'bill_no', 'business_type', 'settlement_mode',
        'unverified_amount', 'customer_receivable_date',
    ]

    def payload(self):
        return {
            'code': 'MCP_0000',
            'data': {
                'columns': [{'key': key, 'name': key} for key in self.KEYS],
                'records': [{
                    'bill_month': '2026-01',
                    'bill_no': 'DFC2601010001',
                    'business_type': '头程业务',
                    'settlement_mode': '月结30天',
                    'unverified_amount': 90.0,
                    'customer_receivable_date': '2026-01-31',
                }],
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        }

    def test_translates_all_six_fields(self):
        result = OutputPresenter().present(
            'query_customer_unverified_bill_details', self.payload()
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(6, len(result['structured_content']['headers']))
        self.assertEqual(6, len(result['structured_content']['rows'][0]))

    def test_missing_extra_malformed_or_bad_pagination_fails_closed(self):
        cases = []
        missing = self.payload()
        del missing['data']['records'][0]['bill_no']
        cases.append(missing)
        extra = self.payload()
        extra['data']['records'][0]['secret'] = 1
        cases.append(extra)
        bad_amount = self.payload()
        bad_amount['data']['records'][0]['unverified_amount'] = True
        cases.append(bad_amount)
        bad_text = self.payload()
        bad_text['data']['records'][0]['bill_month'] = []
        cases.append(bad_text)
        bad_page = self.payload()
        bad_page['meta']['page'] = 101
        cases.append(bad_page)
        extra_meta = self.payload()
        extra_meta['meta']['total'] = 1
        cases.append(extra_meta)
        extra_data = self.payload()
        extra_data['data']['internal'] = True
        cases.append(extra_data)
        bad_records = self.payload()
        bad_records['data']['records'] = 'bad'
        cases.append(bad_records)
        bad_columns = self.payload()
        bad_columns['data']['columns'] = 'bad'
        cases.append(bad_columns)
        bad_column = self.payload()
        bad_column['data']['columns'][0]['internal'] = True
        cases.append(bad_column)

        presenter = OutputPresenter()
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(presenter.present(
                    'query_customer_unverified_bill_details', payload
                )['is_error'])


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