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

import copy
import unittest

from src.step3_question_answer.recommendation import Recommender
from test_recommendation import candidate, good_analysis, handoff, retrieval


class FailureNoticeTests(unittest.TestCase):
    def test_default_analysis_visits_every_chunk_past_old_limit(self):
        item = candidate()
        item['record']['data'] = {'conditions': '合成长文认定标准。' * 2000 + '尾段独立条件'}
        calls = []
        def model(prompt, payload):
            calls.append(copy.deepcopy(payload))
            return {'relevant': False}
        result = Recommender(model)._analyze(item, handoff())
        self.assertGreater(result['chunk_count'], 12)
        self.assertTrue(result['complete'])
        self.assertEqual(len(calls), result['chunk_count'])
        self.assertEqual([p['evidence_scope']['chunk_index'] for p in calls],
                         list(range(1, len(calls) + 1)))
        units = [u for p in calls for u in p['evidence'] if u['path'] == '/data/conditions']
        self.assertEqual(''.join(u['text'] for u in units), item['record']['data']['conditions'])
        self.assertEqual(units[-1]['offset'] + len(units[-1]['text']), len(item['record']['data']['conditions']))
        calls.clear()
        limited = Recommender(model, max_chunks=12)._analyze(item, handoff())
        self.assertFalse(limited['complete'])
        self.assertEqual(limited['chunk_count'], result['chunk_count'])
        self.assertEqual(limited['error'], 'analysis_budget_limit')
        self.assertEqual(calls, [])

    def test_partial_failure_keeps_verified_card_and_one_scoped_notice(self):
        for retrieval_complete in (True, False):
            with self.subTest(retrieval_complete=retrieval_complete):
                failed = candidate()
                failed['record']['id'] = 'failed'
                engine = Recommender(lambda *args: good_analysis())
                analyze = engine._prepare_analysis

                def mixed(item, context, chunk_executor=None):
                    if item['record']['id'] == 'failed':
                        return {'candidate': item, 'analyses': [], 'complete': False,
                                'error': 'model_request_timeout'}
                    return analyze(item, context)

                engine._prepare_analysis = mixed
                result = engine.answer(handoff(), retrieval([candidate(), failed], retrieval_complete))
                self.assertEqual(len(result['items']), 1)
                self.assertFalse(result['retrieval_analysis_complete'])
                self.assertEqual(result['analysis_errors'][0]['code'], 'model_request_timeout')
                self.assertEqual(len(result['summary']['notices']), 1)
                self.assertIn('核验', result['summary']['notices'][0])
                self.assertNotIn(result['summary']['notices'][0], result['overview'])
                self.assertNotIn('部分候选', result['overview'])
                self.assertNotIn('逐条分析', str(result['summary']))

    def test_attachment_failure_is_not_reported_as_candidate_failure(self):
        context = handoff()
        context['attachments'] = [{'id': 'attachment', 'title': '合成附件',
                                  'status': 'failed', 'error': 'document_no_text'}]
        result = Recommender(lambda *args: good_analysis()).answer(context, retrieval())
        self.assertEqual(len(result['items']), 1)
        self.assertFalse(result['complete'])
        self.assertEqual(len(result['summary']['notices']), 1)
        self.assertIn('合成附件未能读取', result['summary']['notices'][0])
        self.assertNotIn('部分候选', result['overview'])
        self.assertNotIn('部分资料', str(result['summary']))

    def test_total_failure_remains_distinct_from_no_results(self):
        bad = copy.deepcopy(good_analysis())
        bad['citations'][0]['quote'] = '伪造引文'
        result = Recommender(lambda *args: bad).answer(handoff(), retrieval())
        self.assertEqual(result['items'], [])
        self.assertEqual(result['summary']['status'], 'unavailable')
        self.assertIn('未完整完成', result['summary']['text'])
        self.assertEqual(result['summary']['notices'], [])
        self.assertTrue(result['analysis_errors'])


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