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)