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

import unittest
from unittest.mock import Mock

from src.step3_question_answer.service_links import record_service_link
from src.step3_question_answer.finance_display import add_detail_links
from src.step3_question_answer.knowledge_answers import synthesize_answers, number_sections
from src.step3_question_answer.report import markdown


class ServiceLinksTests(unittest.TestCase):
    def test_same_entry_url_deduplicates_sources_and_existing_markdown(self):
        entries = [self.entry(), self.entry('b')]
        entries[1]['service_link'] = entries[0]['service_link']
        body = ('### 登记\n\n**办理渠道**\n\n- 网上办理：[入口](https://example.invalid/a)\n'
                '\n可通过[登记平台](https://example.invalid/a)提交申请。[[service_detail:0]]\n[[service_detail:1]]')
        mapping = [{'source_id': e['source_id'], 'heading': '登记'} for e in entries]
        text, links = add_detail_links(body, entries, ['a', 'b'], mapping)
        self.assertEqual(text.count('https://example.invalid/a'), 1)
        self.assertIn('可通过登记平台提交申请。', text)
        self.assertEqual(len(links), 1)
        again, again_links = add_detail_links(text, entries, ['a', 'b'], mapping)
        self.assertEqual(again, text)
        self.assertEqual(again_links, links)

    def test_shared_portal_keeps_each_section_but_metadata_has_one_url(self):
        entries = [self.entry(), self.entry('b')]
        entries[1]['service_link'] = entries[0]['service_link']
        text, links = add_detail_links('### 甲事项\n\n内容。\n\n### 乙事项\n\n内容。', entries, ['a', 'b'], [
            {'source_id': 'a', 'heading': '甲事项'}, {'source_id': 'b', 'heading': '乙事项'}])
        self.assertEqual(text.count('https://example.invalid/a'), 2)
        self.assertEqual(len(links), 1)
        self.assertLess(text.index('https://example.invalid/a'), text.index('### 乙事项'))
        self.assertGreater(text.rindex('https://example.invalid/a'), text.index('### 乙事项'))

    def test_raw_url_is_rebound_and_different_destinations_are_kept(self):
        text, links = add_detail_links('### 同一事项\n\n合成办理说明。\n\nhttps://example.invalid/a\n[[service_detail:0]]\n[[service_detail:1]]',
                                      [self.entry(), self.entry('b')], ['a', 'b'], [
            {'source_id': 'a', 'heading': '同一事项'}, {'source_id': 'b', 'heading': '同一事项'}])
        self.assertEqual(text.count('https://example.invalid/a'), 1)
        self.assertEqual(text.count('https://example.invalid/b'), 1)
        self.assertEqual(len(links), 2)

    def test_explicit_section_mapping_places_different_source_names_and_numbers_sections(self):
        entries = [self.entry(), self.entry('b')]
        model = Mock(return_value={
            'text': '### 登记手续\n\n**办理渠道**\n\n网上办理。\n\n### 其他许可（如涉及）\n\n**办理渠道**\n\n按情况办理。\n\n### 需要您确认的信息\n\n是否涉及。',
            'source_ids': ['a', 'b'], 'link_sections': [
                {'source_id': 'a', 'heading': '登记手续'}, {'source_id': 'b', 'heading': '其他许可（如涉及）'}]})
        result = synthesize_answers(model, '如何办理', entries, [], 'specific_need')
        text = result['text']
        self.assertIn('### 一、登记手续', text)
        self.assertIn('### 二、其他许可（如涉及）', text)
        self.assertLess(text.index('https://example.invalid/a'), text.index('### 二、'))
        self.assertGreater(text.index('https://example.invalid/b'), text.index('### 二、'))
        self.assertLess(text.index('https://example.invalid/b'), text.index('### 需要您确认'))
        self.assertEqual(number_sections(text), text)

    def test_partial_section_mapping_keeps_other_explicitly_used_guides(self):
        entries = [self.entry(chr(ord('a') + i)) for i in range(5)]
        for entry in entries:
            entry['title'] = '合成许可'
            entry['service_link']['title'] = f'合成许可{entry["source_id"]}情形'
        body = '\n\n'.join(f'### 合成许可{entry["source_id"]}情形\n\n办理说明。'
                           for entry in entries)
        mapped = [{'source_id': entry['source_id'], 'heading': f'合成许可{entry["source_id"]}情形'}
                  for entry in entries[:3]]
        text, links = add_detail_links(body, entries, [entry['source_id'] for entry in entries], mapped)
        self.assertEqual(len(links), 5)
        for entry in entries:
            self.assertEqual(text.count(f'https://example.invalid/{entry["source_id"]}'), 1)
        # A fifth source that only appeared in the inputs cannot gain an entry.
        text, links = add_detail_links(body, entries, [entry['source_id'] for entry in entries[:3]], mapped)
        self.assertEqual(len(links), 3)

    def test_section_mapping_cannot_enable_uncited_source_or_missing_heading(self):
        body = '### 甲事项\n\n内容。\n\n### 乙事项\n\n内容。'
        text, links = add_detail_links(body, [self.entry(), self.entry('b')], ['a'], [
            {'source_id': 'b', 'heading': '甲事项'}, {'source_id': 'a', 'heading': '不存在'}])
        self.assertEqual(len(links), 0)
        self.assertNotIn('https://example.invalid/b', text)
        self.assertNotIn('不存在', text)

    def test_conditional_citation_without_section_does_not_add_footer_link(self):
        body = '### 登记手续\n\n内容。\n\n特殊活动另有要求。[[service_detail:1]]'
        text, links = add_detail_links(body, [self.entry(), self.entry('b')], ['a', 'b'], [
            {'source_id': 'a', 'heading': '登记手续'}])
        self.assertEqual([link['source_id'] for link in links], ['a'])
        self.assertIn('https://example.invalid/a', text)
        self.assertNotIn('https://example.invalid/b', text)
        self.assertIn('特殊活动另有要求', text)
        self.assertNotIn('[[service_detail:', text)

    def test_adjacent_service_keeps_citation_but_cannot_enable_entry(self):
        entry = {**self.entry(), 'relevance_level': 'adjacent'}
        model = Mock(return_value={'text': '### 主线事项\n\n另有条件要求。[[service_detail:0]]',
                                  'source_ids': ['a'], 'link_sections': [{'source_id': 'a', 'heading': '主线事项'}]})
        result = synthesize_answers(model, '如何办理', [entry], [], 'specific_need')
        self.assertEqual(result['source_ids'], ['a'])
        self.assertTrue(result['citations'])
        self.assertEqual(result['detail_links'], [])
        self.assertNotIn('https://', result['text'])
        self.assertNotIn('service_marker', model.call_args.args[1]['evidence'][0])

    def test_online_entry_qa_requires_validated_answer_quote(self):
        record = {'kind': 'qa', 'data': {'question': '网上办理设立登记，入口在哪里？',
                                        'answer': '请使用网上入口（https://example.invalid/register）。'}}
        citation = {'path': '/data/answer', 'quote': record['data']['answer']}
        self.assertIsNone(record_service_link(record))
        self.assertIsNone(record_service_link(record, [{'path': '/data/question', 'quote': citation['quote']}]))
        self.assertIsNone(record_service_link(record, [{'path': '/data/answer', 'quote': 'https://evil.invalid'}]))
        link = record_service_link(record, [citation])
        self.assertEqual(link['url'], 'https://example.invalid/register')
        entry = {**self.entry(), 'knowledge_type': 'qa', 'title': record['data']['question'], 'service_link': link}
        model = Mock(return_value={'text': '### 办理渠道\n\n支持网办。[[service_detail:0]]', 'source_ids': ['a']})
        result = synthesize_answers(model, '怎样登记', [entry], [], 'specific_need')
        self.assertIn('https://example.invalid/register', result['text'])
        self.assertEqual(len(result['detail_links']), 1)

    def entry(self, source='a'):
        return {'source_id': source, 'title': '合成备案' + source, 'knowledge_type': 'service_guide',
                'answer': '支持网上办理。', 'citations': [{'path': '/data/body', 'quote': '支持网上办理。'}],
                'service_link': record_service_link({'data': {'structured_content': {'指南链接': 'https://example.invalid/' + source}}})}

    def test_explicit_online_link_precedes_guide_and_pdf_or_homepage_is_not_entry(self):
        record = {'data': {'structured_content': {'指南链接': 'https://example.invalid/guide',
                                                '网上办理链接': 'https://example.invalid/apply'}}}
        link = record_service_link(record)
        self.assertEqual(link['label'], '网上办理入口')
        self.assertEqual(link['path'], '/data/structured_content/网上办理链接')
        self.assertEqual(link['quote'], 'https://example.invalid/apply')
        self.assertIsNone(record_service_link({'source': {'url': 'https://example.invalid'},
                                              'data': {'structured_content': {'指南PDF链接': 'https://example.invalid/a.pdf'}}}))
        for value in ('javascript:alert(1)', '//example.invalid', 'https://u:p@example.invalid', ''):
            self.assertIsNone(record_service_link({'data': {'指南链接': value}}))

    def test_synthesis_places_source_bound_link_in_channel_and_enables_renderer(self):
        model = Mock(return_value={'text': '### 办理渠道\n\n- 网上办理：支持。[[service_detail:0]]\n\n### 申请材料\n\n- 合成材料。', 'source_ids': ['a']})
        result = synthesize_answers(model, '怎么办理', [self.entry()], [], 'specific_need')
        self.assertLess(result['text'].index('[合成备案a（办事指南）]'), result['text'].index('### 二、申请材料'))
        evidence = model.call_args.args[1]['evidence'][0]
        self.assertEqual(evidence['service_link_label'], '查看办事指南')
        self.assertNotIn('https://', str(evidence))
        self.assertIn('href="https://example.invalid/a"', markdown(result['text'], result['detail_links']))

    def test_uncited_missing_and_unknown_markers_never_produce_links(self):
        entries = [self.entry(), self.entry('b')]
        entries[1]['service_link'] = None
        text, links = add_detail_links('说明[[service_detail:0]][[service_detail:1]][[service_detail:99]]', entries, ['b'])
        self.assertEqual(text, '说明')
        self.assertEqual(links, [])

    def test_unlocated_sources_do_not_create_unrelated_footer_links(self):
        text, links = add_detail_links('办理说明', [self.entry(), self.entry('b')], ['a', 'b'])
        self.assertEqual(text, '办理说明')
        self.assertEqual(links, [])

    def test_missing_or_footer_markers_move_into_abbreviated_chapters(self):
        entries = [self.entry(), self.entry('b')]
        entries[0]['title'] = '上海市合成登记营业执照办理（测试区）'
        entries[1]['title'] = '合成回收经营者备案（测试区）'
        body = ('### 一、营业执照办理\n\n**办理渠道**\n\n- 支持网办。\n\n**申请材料**\n\n- 材料甲。\n\n'
                '### 二、合成回收经营者备案\n\n**办理渠道**\n\n- 支持网办。\n\n**申请材料**\n\n- 材料乙。')
        for suffix in ('', '\n\n[[service_detail:0]]\n[[service_detail:1]]'):
            text, links = add_detail_links(body + suffix, entries, ['a', 'b'])
            self.assertEqual(len(links), 2)
            self.assertGreater(text.index('https://example.invalid/a'), text.index('支持网办。'))
            self.assertLess(text.index('https://example.invalid/a'), text.index('材料甲'))
            self.assertGreater(text.index('https://example.invalid/b'), text.index('### 二、'))
            self.assertLess(text.index('https://example.invalid/b'), text.index('材料乙'))
            self.assertNotIn('[[service_detail:', text)

    def test_ambiguous_titles_do_not_guess_a_chapter(self):
        body = '### 合成备案a（甲区）\n\n内容甲。\n\n### 合成备案a（乙区）\n\n内容乙。'
        text, _ = add_detail_links(body, [self.entry()], ['a'])
        self.assertNotIn('https://example.invalid/a', text)

    def test_single_link_does_not_borrow_another_matters_channel(self):
        body = '### 另一事项\n\n#### 办理渠道\n\n窗口甲。\n\n### 未知事项\n\n内容乙。'
        text, _ = add_detail_links(body, [self.entry()], ['a'])
        self.assertNotIn('https://example.invalid/a', text)