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

import unittest

from src.step3_question_answer.recommendation import Recommender
from src.step3_question_answer.knowledge_answers import reference_summary
from src.step3_question_answer.report import markdown, render
from test_recommendation import good_analysis, handoff, retrieval


class ServiceVoiceTests(unittest.TestCase):
    def test_policy_summary_has_no_character_rejection(self):
        from src.step3_question_answer.cards import summarize
        from unittest.mock import Mock
        item = Recommender(lambda *args: good_analysis()).answer(handoff(), retrieval())['items'][0]
        text = '有效说明。' * 500
        model = Mock(return_value={'text': text, 'source_ids': [item['source_id']]})
        summary = summarize(model, '政策内容', [item], [], conversation_kind='policy_overview')
        self.assertEqual(summary['status'], 'generated')
        self.assertEqual(summary['text'], text)
        model.assert_called_once()

    def test_grounded_summary_leads_answer_and_markdown_actions_stay_last(self):
        lead = '您想学技能，可以先了解职业培训支持。'
        def model(instruction, data):
            if 'cards' in data:
                return {'text': lead + '\n\n### 可以先了解的支持\n\n- **职业培训**：提供培训服务。',
                        'source_ids': [data['cards'][0]['source_id']]}
            return good_analysis()
        result = Recommender(model).answer(handoff(), retrieval())
        self.assertEqual(result['overview'], lead)
        self.assertTrue(result['summary']['text'].startswith(lead))
        self.assertEqual(result['summary']['text'].count('### 下一步建议'), 1)
        self.assertLess(result['summary']['text'].index('### 可以先了解'), result['summary']['text'].index('### 下一步建议'))
        self.assertNotIn('由模型', result['overview'])
        self.assertEqual(result['items'][0]['is_policy_library'], True)
        html = render(result)
        self.assertIn('<h3>可以先了解的支持</h3>', html)
        self.assertIn('<strong>职业培训</strong>', html)
        self.assertIn('<ol>', html)

    def test_summary_failure_uses_existing_cited_answer_as_opening(self):
        result = Recommender(lambda *args: good_analysis()).answer(handoff(), retrieval())
        self.assertEqual(result['summary']['status'], 'fallback')
        self.assertEqual(result['overview'], good_analysis()['answer'])
        self.assertIn('### 可以先了解的支持', result['summary']['text'])

    def test_exact_qa_answer_without_source_appendix_or_rewriting(self):
        answer = '可以线上提交；特殊事项除外。'
        result = reference_summary([{'source_id': 'qa', 'title': '合成材料问题', 'answer': answer,
                                     'citations': [{'path': '/data/answer', 'quote': answer}], 'exact_qa': True}])
        self.assertTrue(result['text'].startswith('> ' + answer))
        self.assertEqual(result['citations'][0]['quote'], answer)
        self.assertEqual(result['text'], '> ' + answer)

    def test_markdown_does_not_activate_html_images_or_links(self):
        html = markdown('### 标题\n\n<script>bad()</script>\n\n![图片](https://example.invalid/a.png)\n\n[链接](javascript:bad())')
        self.assertIn('<h3>标题</h3>', html)
        self.assertNotIn('<script>', html)
        self.assertNotIn('<img', html)
        self.assertNotIn('<a ', html)

    def test_overview_summary_rejects_unrequested_qualification_question(self):
        def model(instruction, data):
            if 'cards' in data:
                self.assertEqual(data['conversation_kind'], 'policy_overview')
                return {'text': '您是否在本区登记？', 'source_ids': [data['cards'][0]['source_id']]}
            value = good_analysis()
            value['answer'] = '这项服务提供职业技能培训。'
            return value
        result = Recommender(model).answer({**handoff(), 'conversation_kind': 'policy_overview'}, retrieval())
        self.assertEqual(result['summary']['status'], 'fallback')
        self.assertNotIn('您是否', result['summary']['text'])
        self.assertEqual(result['next_actions'], [])

    def test_supplied_fact_downgrade_does_not_create_generic_reconfirmation(self):
        from src.step3_question_answer.recommendation import validate_analysis, evidence_units
        from test_recommendation import RECORD
        analysis = good_analysis()
        analysis['conditions'][0].update(status='满足', question='', user_evidence='我失业了')
        result = validate_analysis(analysis, evidence_units(RECORD), {'question': handoff()['original_question']})
        self.assertEqual(result['conditions'][0]['status'], '待确认')
        self.assertEqual(result['conditions'][0]['question'], '')