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

import copy
import unittest

from src.step3_question_answer.next_actions import build_next_actions, append_to_summary
from src.step3_question_answer.recommendation import Recommender
from src.step3_question_answer.report import render
from test_recommendation import candidate, good_analysis, handoff, retrieval


class NextActionTests(unittest.TestCase):
    def result(self):
        def model(prompt, data):
            if 'cards' in data:
                return {'text': '可了解合成支持，登记地区仍待确认。',
                        'source_ids': [card['source_id'] for card in data['cards']]}
            return good_analysis()
        return Recommender(model).answer(handoff(), retrieval())

    def test_actions_surface_pending_and_cite_exact_original(self):
        result = self.result()
        actions = result['next_actions']
        self.assertEqual(len(actions), 3)
        self.assertEqual(actions[0]['description'], '在哪个区登记？')
        self.assertEqual(actions[0]['citations'][0]['path'], '/data/conditions')
        self.assertEqual(actions[0]['citations'][0]['quote'], '在本区登记')
        self.assertEqual(result['summary']['status'], 'generated')
        html = render(result)
        self.assertEqual(html.count('<h3>下一步建议</h3>'), 1)
        self.assertIn('1. **确认相关地区**', result['summary']['text'])
        self.assertIn('3. ', result['summary']['text'])
        self.assertNotIn('相关政策要求：', result['summary']['text'])
        self.assertIn('相关政策要求：', actions[0]['reason'])
        self.assertIn('在哪个区登记？', html)
        self.assertIn('href="#source-' + result['sources'][0]['id'], html)
        self.assertLess(html.index('下一步建议'), html.index('<h2>完整参考资料'))

    def test_failed_policy_summary_does_not_request_application_details(self):
        result = Recommender(lambda *args: good_analysis()).answer(handoff(), retrieval())
        self.assertEqual(result['summary']['status'], 'fallback')
        self.assertEqual(result['next_actions'], [])
        self.assertEqual(result['follow_up_questions'], [])
        self.assertNotIn('下一步建议', result['summary']['text'])

    def test_duplicate_conditions_merge_sources_and_limit_three(self):
        result = self.result()
        first = result['items'][0]
        # Action generation uses full internal conditions before card projection.
        first['conditions'] = copy.deepcopy(result['candidate_diagnostics'][0]['analyses'][0]['conditions'])
        second = copy.deepcopy(first)
        second['source_id'] = 'another'
        source = copy.deepcopy(result['sources'][0])
        source['id'] = 'another'
        sources = result['sources'] + [source]
        actions = build_next_actions([first, second], sources)
        self.assertEqual(actions[0]['source_ids'], [first['source_id'], 'another'])
        self.assertEqual(sum(a['kind'] == 'confirm_information' for a in actions), 1)
        for index in range(5):
            condition = dict(first['conditions'][0], requirement=f'合成不同条件{index}', question=f'问题{index}')
            first['conditions'].append(condition)
        actions = build_next_actions([first], sources)
        self.assertEqual(len(actions), 3)
        self.assertTrue(all(a['kind'] == 'confirm_information' for a in actions))

    def test_no_evidence_not_confirmed_or_overview_has_no_forced_questions(self):
        result = self.result()
        for status, quote in [('满足', '在本区登记'), ('待确认', ''), ('待确认', '不存在的条件')]:
            result['items'][0]['conditions'] = [dict(good_analysis()['conditions'][0], status=status, quote=quote)]
            actions = build_next_actions(result['items'], result['sources'])
            self.assertEqual(len(actions), 3)
            self.assertFalse(any(a['kind'] == 'confirm_information' for a in actions))
        self.assertEqual(build_next_actions([], result['sources']), [])
        self.assertEqual(build_next_actions(result['items'], result['sources'], False), [])
        for field in ('support_methods_standards', 'application_conditions'):
            items = copy.deepcopy(result['items'])
            items[0]['card'][field] = None
            self.assertEqual(len(build_next_actions(items, result['sources'])), 3)

    def test_excluded_or_no_results_and_overview_do_not_emit_actions(self):
        for mode in ('overview', 'empty', 'failure', 'ineligible'):
            context, records = handoff(), retrieval()
            analysis = good_analysis()
            if mode == 'overview':
                context['conversation_kind'] = 'policy_overview'
            elif mode in ('empty', 'failure'):
                records = retrieval([], complete=mode != 'failure')
            else:
                analysis['conditions'][0].update(status='不满足', user_evidence=context['original_question'])
            result = Recommender(lambda *args: copy.deepcopy(analysis)).answer(context, records)
            self.assertEqual(result['next_actions'], [], mode)
            self.assertNotIn('### 下一步建议', result['summary']['text'])

    def test_action_text_is_html_escaped(self):
        result = self.result()
        result['next_actions'][0]['description'] = '<script>bad()</script>'
        result['summary'] = append_to_summary({'text': '原总结', 'source_ids': []}, result['next_actions'])
        self.assertIn('&lt;script&gt;bad()', render(result))
        self.assertNotIn('<script>bad()', render(result))

    def test_existing_summary_content_metadata_and_sources_preserved(self):
        actions = self.result()['next_actions']
        original = {'text': '原结论', 'source_ids': ['qa'], 'status': 'generated', 'notices': ['原提示']}
        updated = append_to_summary(original, actions)
        self.assertTrue(updated['text'].startswith('原结论\n\n### 下一步建议'))
        self.assertEqual(updated['source_ids'], ['qa', actions[0]['source_ids'][0]])
        self.assertEqual(updated['status'], 'generated')
        self.assertEqual(updated['notices'], ['原提示'])
        self.assertEqual(original['text'], '原结论')
        self.assertIs(append_to_summary(original, []), original)