174 lines
10 KiB
Python
174 lines
10 KiB
Python
"""Synthetic evidence verifies scope, references and bounded retrieval without I/O."""
|
|
|
|
from copy import deepcopy
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from logistics_chat.knowledge_service import search_business_knowledge, read_business_knowledge
|
|
|
|
|
|
def fixture_book():
|
|
return {'id': 7, 'source_name': 'private-company-price.xlsx',
|
|
'rates': [{'service': '美森正班海卡', 'kind': 'truck'},
|
|
{'service': 'EXX海卡', 'kind': 'truck'},
|
|
{'service': '普船海派', 'kind': 'parcel'}],
|
|
'documents': [
|
|
{'book_id': 7, 'sheet': '美森正班海卡', 'source_cell': 'A8',
|
|
'content': '外箱必须贴入仓条码和唛头。旧条码须移除，但产品原产地标识不得遮挡。'},
|
|
{'book_id': 7, 'sheet': 'EXX海卡', 'source_cell': 'B9',
|
|
'content': '外箱标签需包含货件编号；只有本渠道接受两面贴标。'},
|
|
{'book_id': 7, 'sheet': '海派', 'source_cell': 'C10',
|
|
'content': '本海派渠道不提供上门提货服务，客户须送货到指定仓库。'},
|
|
{'book_id': 7, 'sheet': '合成说明', 'source_cell': 'D11',
|
|
'content': '发票申请需提交结算凭证，开票条件需要人工核实。'},
|
|
]}
|
|
|
|
|
|
class KnowledgeServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.book = fixture_book()
|
|
|
|
def test_reviewed_currency_replaces_the_conflicting_wording_with_traceable_source(self):
|
|
self.book['documents'].append({'book_id': 7, 'sheet': '美森正班海卡', 'source_cell': 'A53',
|
|
'content': '海外进口清关超过5个品名的部分，加收USD30元RMB/个的附加费。'})
|
|
result = search_business_knowledge('海外进口清关品名续页费用', country='US', services=['美森正班海卡'], book=self.book)
|
|
self.assertIn('USD30/个', result['t'])
|
|
self.assertNotIn('USD30元RMB', result['t'])
|
|
source = next(item for item in result['sources'] if item.get('cell') == 'A53')
|
|
self.assertEqual(source['business_rule_revision'], 'customs-extra-name-usd-20261106')
|
|
read = read_business_knowledge(source['reference'], country='US', services=['美森正班海卡'], book=self.book)
|
|
self.assertIn('USD30/个', read['t'])
|
|
self.assertIn('USD30元RMB', self.book['documents'][-1]['content'])
|
|
|
|
def test_synonyms_find_full_policy_and_can_read_same_reference(self):
|
|
result = search_business_knowledge('可以上门取件吗', country='us',
|
|
services=['普船海派'], book=self.book)
|
|
self.assertEqual(result['status'], 'ok')
|
|
match = result['data']['matches'][0]
|
|
self.assertIn('不提供上门提货', match['text'])
|
|
read = read_business_knowledge(match['reference'], country='USA',
|
|
services=['普船海派'], book=self.book)
|
|
self.assertEqual(read['data']['matches'][0]['text'], match['text'])
|
|
self.assertEqual(result['data_version'], 7)
|
|
self.assertEqual(result['data']['resolved_country'], 'US')
|
|
self.assertEqual(result['sources'][0]['cell'], 'C10')
|
|
|
|
def test_labels_keep_tail_exception_and_service_scope(self):
|
|
result = search_business_knowledge('labels 怎么贴', country='US',
|
|
services=['美森正班'], book=self.book)
|
|
self.assertEqual(result['status'], 'ok')
|
|
self.assertEqual(len(result['data']['matches']), 1)
|
|
self.assertIn('不得遮挡', result['t'])
|
|
self.assertNotIn('两面贴标', result['t'])
|
|
ref = result['sources'][0]['reference']
|
|
self.assertEqual(read_business_knowledge(ref, country='US', services=['EXX'],
|
|
book=self.book)['status'], 'not_found')
|
|
|
|
def test_us_never_receives_legacy_european_prices(self):
|
|
result = search_business_knowledge('美国海运费用和报价多少', country='US', book=self.book)
|
|
self.assertNotIn('12.3元', result['t'])
|
|
self.assertNotIn('欧洲', result['t'])
|
|
self.assertEqual(result['status'], 'not_found')
|
|
legacy = search_business_knowledge('欧洲海运快线价格', country='DE')
|
|
self.assertEqual(legacy['status'], 'ok')
|
|
self.assertIn('12.3元', legacy['t'])
|
|
|
|
def test_unknown_country_does_not_assume_book_or_unclassified_faq_scope(self):
|
|
result = search_business_knowledge('海运费用报价', book=self.book)
|
|
self.assertEqual(result['status'], 'not_found')
|
|
inferred = search_business_knowledge('美国外箱标签要求', book=self.book)
|
|
self.assertEqual(inferred['status'], 'ok')
|
|
self.assertEqual(inferred['data']['resolved_country'], 'US')
|
|
ref = inferred['sources'][0]['reference']
|
|
self.assertEqual(read_business_knowledge(ref, book=self.book)['status'], 'not_found')
|
|
with patch('logistics_chat.knowledge_service.knowledge_list',
|
|
[{'keys': ['新业务'], 'text': '新业务的海运费用为每票9元。'}]):
|
|
self.assertEqual(search_business_knowledge('新业务费用', country='US')['status'], 'not_found')
|
|
|
|
def test_unknown_service_does_not_inherit_an_unscoped_policy(self):
|
|
result = search_business_knowledge('发票怎么申请', country='US',
|
|
services=['未知渠道'], book=self.book)
|
|
self.assertEqual(result['status'], 'not_found')
|
|
unscoped = search_business_knowledge('发票怎么申请', country='US', book=self.book)
|
|
self.assertEqual(unscoped['status'], 'ok')
|
|
self.assertFalse(unscoped['data']['matches'][0]['scope_confirmed'])
|
|
self.assertIn('适用范围', unscoped['t'])
|
|
|
|
def test_read_rechecks_version_content_and_country(self):
|
|
result = search_business_knowledge('外箱标签', country='US', book=self.book)
|
|
ref = result['sources'][0]['reference']
|
|
changed = deepcopy(self.book)
|
|
changed['id'] = 8
|
|
for row in changed['documents']:
|
|
row['book_id'] = 8
|
|
self.assertEqual(read_business_knowledge(ref, country='US', book=changed)['status'], 'not_found')
|
|
changed = deepcopy(self.book)
|
|
for row in changed['documents']:
|
|
row['content'] += ' 新版补充要求。'
|
|
self.assertEqual(read_business_knowledge(ref, country='US', book=changed)['status'], 'not_found')
|
|
self.assertEqual(read_business_knowledge(ref, country='UK', book=self.book)['status'], 'not_found')
|
|
self.assertEqual(read_business_knowledge('C:/private/data.xlsx', country='US', book=self.book)['status'], 'not_found')
|
|
|
|
def test_no_result_is_explicit_and_never_reads_a_database(self):
|
|
with patch('logistics_chat.quote_data.load_book', side_effect=AssertionError('must use supplied snapshot')):
|
|
result = search_business_knowledge('火星轨道燃料', country='US')
|
|
self.assertEqual(result['status'], 'not_found')
|
|
self.assertEqual(result['sources'], [])
|
|
self.assertIn('未查到', result['t'])
|
|
general = search_business_knowledge('周末上班吗', country='US')
|
|
self.assertEqual(general['status'], 'ok')
|
|
self.assertIn('8:00', general['t'])
|
|
|
|
def test_duplicate_cells_do_not_fill_top_k(self):
|
|
duplicate = dict(self.book['documents'][0], source_cell='A9')
|
|
self.book['documents'].append(duplicate)
|
|
result = search_business_knowledge('外箱标签', country='US', book=self.book, limit=99)
|
|
values = [item['text'] for item in result['data']['matches']]
|
|
self.assertEqual(len(values), len(set(values)))
|
|
self.assertLessEqual(len(values), 5)
|
|
|
|
def test_sensitive_source_and_contacts_are_not_returned(self):
|
|
texts = ['贴标请联系联系人：合成人员，手机号：13812345678。',
|
|
'贴标资料在 C:\\private\\customer.xlsx，请查看。',
|
|
'贴标请发邮件给 private@example.com。',
|
|
'贴标联系合成人员 +1 (909) 123-4567。']
|
|
self.book['documents'] = [{'sheet': '美森正班海卡', 'source_cell': f'A{i+1}',
|
|
'content': text} for i, text in enumerate(texts)]
|
|
result = search_business_knowledge('贴标', country='US', book=self.book)
|
|
self.assertEqual(result['status'], 'not_found')
|
|
serialized = json.dumps(result, ensure_ascii=False)
|
|
for secret in ('13812345678', 'private', '合成人员', 'company-price'):
|
|
self.assertNotIn(secret, serialized)
|
|
|
|
def test_mismatched_book_rows_and_new_regional_faq_are_not_general(self):
|
|
self.book['documents'][0]['book_id'] = 8
|
|
self.book['documents'][1]['content'] = '英国海运外箱标签要求两面贴标。'
|
|
result = search_business_knowledge('外箱标签', country='US', book=self.book)
|
|
self.assertEqual(result['status'], 'not_found')
|
|
with patch('logistics_chat.knowledge_service.knowledge_list',
|
|
[{'keys': ['人工客服'], 'text': '美国人工客服代办费20元。'}]):
|
|
self.assertEqual(search_business_knowledge('人工客服代办', country='US')['status'], 'not_found')
|
|
|
|
def test_long_passages_are_not_truncated_before_exceptions(self):
|
|
self.book['documents'][0]['content'] = '贴标要求说明。' * 400 + '但此渠道禁止带电货。'
|
|
result = search_business_knowledge('贴标', country='US', services=['美森正班'], book=self.book)
|
|
self.assertEqual(result['status'], 'not_found')
|
|
self.assertTrue(result['data']['warnings'])
|
|
self.assertEqual(search_business_knowledge('贴标' * 400, country='US', book=self.book)['status'], 'error')
|
|
self.assertEqual(search_business_knowledge('贴标', limit=True)['status'], 'error')
|
|
|
|
def test_total_response_budget_and_snapshot_are_unchanged(self):
|
|
base = '贴标要求必须保留产品标识。' * 65
|
|
self.book['documents'] = [{'sheet': '美森正班海卡', 'source_cell': f'A{i+1}',
|
|
'content': base + f'适用包装编号{i}。'} for i in range(20)]
|
|
original = deepcopy(self.book)
|
|
result = search_business_knowledge('贴标要求', country='US', book=self.book, limit=20)
|
|
self.assertLessEqual(len(result['t']), 7000)
|
|
self.assertLessEqual(len(result['data']['matches']), 5)
|
|
self.assertEqual(self.book, original)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|