business-assistant/src/step3_question_answer/tests/test_agent.py

import hashlib
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
from urllib.parse import parse_qs, urlsplit

from langgraph.types import Command

from src.step3_question_answer.graph import build_graph
from src.step3_question_answer.qcc import Config, QccClient, QccError, load_config


COMPANY = {'KeyNo': 'synthetic-a', 'Name': '合成甲企业', 'CreditCode': 'SYNTHETIC-A'}
OTHER = {'KeyNo': 'synthetic-b', 'Name': '合成乙企业', 'CreditCode': 'SYNTHETIC-B'}


class Wire:
    def __init__(self, *responses):
        self.responses = list(responses)
        self.calls = []

    def __call__(self, url, headers, timeout):
        self.calls.append((url, headers, timeout))
        result = self.responses.pop(0)
        if isinstance(result, Exception):
            raise result
        return result


def ok(result):
    return {'Status': '200', 'Result': result}


def client(wire, pages=3):
    return QccClient(Config('synthetic-key', 'synthetic-secret', honor_max_pages=pages), wire, lambda: 1700000000)


class ProtocolTests(unittest.TestCase):
    def test_config_only_synthetic_secrets(self):
        with TemporaryDirectory() as directory:
            path = Path(directory) / '.env'
            path.write_text('QCC_KEY=synthetic\nQCC_SECRET_KEY=hidden\n', encoding='utf-8')
            config = load_config(path, {'QCC_TIMEOUT_SECONDS': '12'})
            self.assertEqual(config.timeout, 12)
            self.assertNotIn('hidden', repr(config))
            with self.assertRaisesRegex(ValueError, '^invalid_qcc_configuration$'):
                load_config(path, {'QCC_TIMEOUT_SECONDS': 'secret-like-invalid-value'})

    def test_configured_qcc_urls_are_used_for_all_requests(self):
        with TemporaryDirectory() as directory:
            path = Path(directory) / '.env'
            path.write_text('QCC_KEY=synthetic\nQCC_SECRET_KEY=hidden\n'
                            'QCC_SEARCH_URL=https://gateway.example.test/search\n'
                            'QCC_BASIC_URL=https://gateway.example.test/basic\n'
                            'QCC_HONOR_URL=https://gateway.example.test/honor\n', encoding='utf-8')
            config = load_config(path, {})
            wire = Wire(ok([COMPANY]), ok(COMPANY), ok({'VerifyResult': 0, 'Data': []}))
            adapter = QccClient(config, wire, lambda: 1700000000)
            adapter.search('合成')
            adapter.basic(COMPANY)
            adapter.honors(COMPANY)
            self.assertEqual([urlsplit(call[0]).path for call in wire.calls],
                             ['/search', '/basic', '/honor'])
            self.assertEqual([urlsplit(call[0]).hostname for call in wire.calls],
                             ['gateway.example.test'] * 3)
            self.assertNotIn('gateway.example.test', repr(config))
            override = load_config(path, {'QCC_SEARCH_URL': 'https://override.example.test/search'})
            self.assertEqual(override.search_url, 'https://override.example.test/search')

    def test_invalid_qcc_urls_are_rejected(self):
        for value in ('', 'http://example.test/search', 'https://user:pass@example.test/search',
                      'https://example.test/search?token=secret', 'https://example.test/search#fragment',
                      'https://example.test:bad/search', 'https://example.test/search path'):
            with self.subTest(value=value), self.assertRaisesRegex(ValueError, 'invalid_qcc_url'):
                Config('synthetic-key', 'synthetic-secret', search_url=value)

    def test_exact_full_page_stops_by_total(self):
        data = [{'Name': '合成荣誉'} for _ in range(20)]
        wire = Wire({**ok({'VerifyResult': 1, 'Data': data}), 'Paging': {'TotalRecords': 20}})
        self.assertTrue(client(wire).honors(COMPANY)['complete'])
        self.assertEqual(len(wire.calls), 1)
        candidates = [{**COMPANY, 'KeyNo': str(i)} for i in range(5)]
        result = client(Wire({**ok(candidates), 'Paging': {'TotalRecords': 5}})).search('合成')
        self.assertFalse(result['may_have_more'])

    def test_signature_query_and_provenance(self):
        wire = Wire(ok([COMPANY]))
        result = client(wire).search('合成 企业&分支', 2)
        url, headers, timeout = wire.calls[0]
        self.assertEqual(urlsplit(url).path, '/FuzzySearch/GetList')
        self.assertEqual(parse_qs(urlsplit(url).query)['searchKey'], ['合成 企业&分支'])
        self.assertEqual(parse_qs(urlsplit(url).query)['pageIndex'], ['2'])
        self.assertEqual(headers['Token'], hashlib.md5(b'synthetic-key1700000000synthetic-secret').hexdigest().upper())
        self.assertEqual(headers['Timespan'], '1700000000')
        self.assertEqual(result['source']['api_code'], '886')
        self.assertIn('2023', result['source']['fetched_at'])
        self.assertNotIn('synthetic-key', repr(client(wire).config))

    def test_empty_distinct_from_service_error(self):
        self.assertEqual(client(Wire(ok([]))).search('合成')['candidates'], [])
        for status in ('201', 201):
            for extra in ({}, {'Result': None}, {'Result': []}):
                with self.subTest(status=status, extra=extra):
                    result = client(Wire({'Status': status, **extra})).search('合成')
                    self.assertEqual(result['candidates'], [])
                    self.assertFalse(result['may_have_more'])
                    self.assertEqual(result['source']['api_code'], '886')
                    self.assertEqual(result['source']['total_records'], 0)
        for payload in ({'Status': '201', 'Result': [COMPANY]}, {'Status': '102'}, {'Status': '999'}):
            with self.assertRaises(QccError):
                client(Wire(payload)).search('合成')
        with self.assertRaisesRegex(QccError, 'provider_failure'):
            client(Wire({'Status': '201'})).basic(COMPANY)
        for payload in [{'Status': '500', 'Message': 'private-secret'}, {}, ok(None), ok({})]:
            with self.subTest(payload=payload):
                with self.assertRaises(QccError) as raised:
                    client(Wire(payload)).search('合成')
                self.assertNotIn('private-secret', str(raised.exception))

    def test_basic_identity(self):
        wire = Wire(ok(COMPANY))
        result = client(wire).basic(COMPANY)
        self.assertEqual(result['source']['company_key'], COMPANY['KeyNo'])
        self.assertEqual(parse_qs(urlsplit(wire.calls[0][0]).query)['keyword'], [COMPANY['Name']])
        for row in [OTHER, {**COMPANY, 'CreditCode': 'other'}, None]:
            with self.assertRaises(QccError):
                client(Wire(ok(row))).basic(COMPANY)

    def test_honor_pagination_fields_and_partial(self):
        data = [{'Name': f'合成荣誉{i}', 'BegingDate': '2020-01-01', 'DeadLine': None} for i in range(20)]
        wire = Wire(ok({'VerifyResult': 1, 'Data': data}), ok({'VerifyResult': 1, 'Data': [data[0]]}))
        result = client(wire).honors(COMPANY)
        self.assertTrue(result['complete'])
        self.assertEqual(len(result['records']), 21)
        self.assertIsNone(result['records'][0]['data']['DeadLine'])
        self.assertEqual(result['records'][-1]['source']['page_index'], 2)
        self.assertEqual(parse_qs(urlsplit(wire.calls[1][0]).query)['pageSize'], ['20'])
        partial = client(Wire(ok({'VerifyResult': 1, 'Data': data}), QccError('network_failure'))).honors(COMPANY)
        self.assertFalse(partial['complete'])
        self.assertEqual(len(partial['records']), 20)
        self.assertEqual(partial['error'], 'network_failure')
        limited = client(Wire(ok({'VerifyResult': 1, 'Data': data})), pages=1).honors(COMPANY)
        self.assertEqual(limited['error'], 'page_limit')

    def test_honor_empty_and_invalid(self):
        self.assertTrue(client(Wire(ok({'VerifyResult': 0, 'Data': None}))).honors(COMPANY)['complete'])
        for data in [{'VerifyResult': 1, 'Data': None}, {'VerifyResult': 0, 'Data': [{}]}, {}]:
            result = client(Wire(ok(data))).honors(COMPANY)
            self.assertFalse(result['complete'])
            self.assertIsNotNone(result['error'])


class GraphTests(unittest.TestCase):
    def test_unique_exact_name_or_code_skips_confirmation(self):
        for hint in (COMPANY['Name'], COMPANY['CreditCode']):
            with self.subTest(hint=hint):
                wire = Wire(ok([OTHER, COMPANY]), ok(COMPANY), ok({'VerifyResult': 0, 'Data': []}))
                graph = build_graph(client(wire), lambda q: {
                    'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': hint})
                question = f'我们是{hint}，想扩产有什么支持？'
                result = graph.invoke({'question': question}, self.config())
                self.assertNotIn('__interrupt__', result)
                self.assertEqual(result['selected_company']['KeyNo'], COMPANY['KeyNo'])
                self.assertEqual(result['compang_info']['profile']['data'], COMPANY)
                self.assertEqual(result['handoff']['original_question'], question)
                self.assertEqual(len(wire.calls), 3)

    def test_duplicate_exact_names_still_require_confirmation(self):
        wire = Wire(ok([COMPANY, {**OTHER, 'Name': COMPANY['Name']}]))
        graph = build_graph(client(wire), lambda q: {
            'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': COMPANY['Name']})
        result = graph.invoke({'question': COMPANY['Name'] + '想扩产'}, self.config())
        self.assertIn('__interrupt__', result)
        self.assertIsNone(result['selected_company'])
        self.assertEqual(len(wire.calls), 1)

    def test_refined_exact_name_continues_without_second_confirmation(self):
        wire = Wire(ok([]), ok([COMPANY]), ok(COMPANY), ok({'VerifyResult': 0, 'Data': []}))
        graph = build_graph(client(wire), lambda q: {
            'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': '简称'})
        graph.invoke({'question': '简称想扩产'}, self.config())
        result = graph.invoke(Command(resume={'action': 'refine', 'keyword': COMPANY['Name']}), self.config())
        self.assertNotIn('__interrupt__', result)
        self.assertEqual(result['handoff']['original_question'], '简称想扩产')
        self.assertEqual(result['searches'], 2)

    def test_auto_selected_identity_still_validated_by_basic(self):
        wire = Wire(ok([COMPANY]), ok(OTHER))
        graph = build_graph(client(wire), lambda q: {
            'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': COMPANY['Name']})
        result = graph.invoke({'question': COMPANY['Name'] + '想扩产'}, self.config())
        self.assertIn('basic:company_identity_mismatch', result['errors'])
        self.assertIsNone(result['honors'])
        self.assertEqual(result['compang_info'], {})
        self.assertEqual(len(wire.calls), 2)

    def test_model_invented_full_name_cannot_trigger_auto_selection(self):
        wire = Wire()
        graph = build_graph(client(wire), lambda q: {
            'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': COMPANY['Name']})
        result = graph.invoke({'question': '简称想扩产'}, self.config())
        self.assertIn('intent_analysis_failed', result['errors'])
        self.assertIsNone(result['selected_company'])
        self.assertEqual(wire.calls, [])

    def config(self, thread='one'):
        return {'configurable': {'thread_id': thread}, 'recursion_limit': 100}

    def test_confirm_required_even_single_and_no_replayed_search(self):
        wire = Wire(ok([COMPANY]), ok(COMPANY), ok({'VerifyResult': 0, 'Data': []}))
        graph = build_graph(client(wire), lambda q: {'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': '合成'})
        result = graph.invoke({'question': '合成公司如何招人'}, self.config())
        self.assertIn('__interrupt__', result)
        self.assertEqual(len(wire.calls), 1)
        invalid = graph.invoke(Command(resume={'action': 'select', 'key_no': OTHER['KeyNo']}), self.config())
        self.assertIn('__interrupt__', invalid)
        self.assertEqual(len(wire.calls), 1)
        final = graph.invoke(Command(resume={'action': 'select', 'key_no': COMPANY['KeyNo']}), self.config())
        self.assertEqual(len(wire.calls), 3)
        self.assertEqual(final['handoff']['original_question'], '合成公司如何招人')
        self.assertEqual(final['handoff']['qualification_status'], 'not_evaluated')
        self.assertEqual(final['handoff']['policy_evidence'], [])

    def test_personal_skip_and_reset_and_threads(self):
        wire = Wire(ok([COMPANY]), ok(COMPANY), ok({'VerifyResult': 0, 'Data': []}))
        graph = build_graph(client(wire), lambda q: {'conversation_kind': 'specific_need', 'needs_company': '公司' in q, 'company_hint': '合成' if '公司' in q else ''})
        graph.invoke({'question': '合成公司咨询'}, self.config())
        personal = graph.invoke({'question': '个人办事'}, self.config('two'))
        self.assertIsNone(personal['selected_company'])
        self.assertEqual(len(wire.calls), 1)
        graph.invoke(Command(resume={'action': 'select', 'key_no': COMPANY['KeyNo']}), self.config())
        reset = graph.invoke({'question': '个人咨询'}, self.config())
        self.assertIsNone(reset['profile'])
        self.assertIsNone(reset['honors'])
        self.assertIsNone(reset['selected_company'])
        self.assertEqual(len(wire.calls), 3)

    def test_refine_empty_and_cancel(self):
        wire = Wire(ok([]), ok([COMPANY, OTHER]))
        graph = build_graph(client(wire))
        graph.invoke({'question': '扩产想找支持'}, self.config())
        graph.invoke(Command(resume={'needs_company': True, 'company_hint': '模糊名称'}), self.config())
        result = graph.invoke(Command(resume={'action': 'refine', 'keyword': '新名称'}), self.config())
        self.assertEqual(len(result['search_result']['candidates']), 2)
        result = graph.invoke(Command(resume={'action': 'cancel'}), self.config())
        self.assertFalse(result['handoff']['ready_for_retrieval'])
        self.assertEqual(len(wire.calls), 2)

    def test_identity_mismatch_stops_honors(self):
        wire = Wire(ok([COMPANY]), ok(OTHER))
        graph = build_graph(client(wire), lambda q: {'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': '合成'})
        graph.invoke({'question': '合成融资'}, self.config())
        result = graph.invoke(Command(resume={'action': 'select', 'key_no': COMPANY['KeyNo']}), self.config())
        self.assertIsNone(result['honors'])
        self.assertFalse(result['handoff']['ready_for_retrieval'])
        self.assertIn('basic:company_identity_mismatch', result['errors'])
        self.assertEqual(len(wire.calls), 2)

    def test_untrusted_model_hint_requires_clarification(self):
        wire = Wire()
        graph = build_graph(client(wire), lambda q: {'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': '编造全名'})
        result = graph.invoke({'question': '我厂扩产缺资金'}, self.config())
        self.assertNotIn('__interrupt__', result)
        self.assertEqual(result['conversation_kind'], 'policy_overview')
        self.assertEqual(result['recommendation'], {})
        self.assertEqual(wire.calls, [])

    def test_search_errors_retry_bounded(self):
        wire = Wire(*[QccError('network_failure') for _ in range(5)])
        graph = build_graph(client(wire), lambda q: {'conversation_kind': 'specific_need', 'needs_company': True, 'company_hint': '合成'})
        result = graph.invoke({'question': '合成融资'}, self.config())
        for _ in range(5):
            result = graph.invoke(Command(resume={'action': 'retry'}), self.config())
        self.assertNotIn('__interrupt__', result)
        self.assertIn('search_limit', result['errors'])
        self.assertEqual(len(wire.calls), 5)


class IntentTests(unittest.TestCase):
    def test_model_json_and_truncation(self):
        from src.step3_question_answer.intent import deepseek_analyzer
        from src.step1_knowledge_prepare.summarize import Config as ModelConfig
        config = ModelConfig('fake-secret', 'https://example.invalid', 'synthetic-model')
        response = {'choices': [{'finish_reason': 'stop', 'message': {
            'content': '{"needs_company":false,"company_hint":"","cipa":false}'}}]}
        with patch('src.step3_question_answer.intent.load_config', return_value=config), patch(
                'src.step3_question_answer.intent.transport', return_value=response) as send:
            analyzer = deepseek_analyzer(Path('unused'))
            self.assertFalse(analyzer('个人咨询')['needs_company'])
            self.assertEqual(send.call_args.args[0]['messages'][-1]['content'], '个人咨询')
            response['choices'][0]['finish_reason'] = 'length'
            with self.assertRaisesRegex(ValueError, 'incomplete_intent'):
                analyzer('个人咨询')


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