233 lines
14 KiB
Python
233 lines
14 KiB
Python
"""Local order snapshots, exact lookup and conversational order queries."""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from logistics_chat import create_app, db, order_data
|
||
from logistics_chat.tracking import CARD_PREFIX, card_to_text
|
||
|
||
|
||
class OrderDataTests(unittest.TestCase):
|
||
def setUp(self):
|
||
folder = tempfile.TemporaryDirectory()
|
||
self.addCleanup(folder.cleanup)
|
||
root = Path(folder.name)
|
||
self.app = create_app({'TESTING':True, 'DATABASE':root/'history.db',
|
||
'UPLOAD_DIR':root/'uploads', 'BACKUP_DIR':root/'backups', 'LOG_DIR':root/'logs',
|
||
'DEEPSEEK_API_KEY':'', 'ORDER_DATA_TENANT_ID':'', 'RATE_LIMIT_PER_MIN':1000})
|
||
self.client = self.app.test_client()
|
||
self.client.post('/api/visitor/session', json={'key':'1'*64})
|
||
self.context = self.app.app_context()
|
||
self.context.push()
|
||
self.addCleanup(self.context.pop)
|
||
self.record = {'source_order_id':'999999999999999999', 'tenant_id':'tenant-a',
|
||
'order_no':'FL202610290001', 'inbound_code':'INB20261029001-01', 'shipment_id':'shipment00208',
|
||
'status':'PROCESSING', 'country_code':'US', 'transport_mode':'SEA', 'warehouse_code':'TEST1',
|
||
'order_role':'MASTER', 'package_count':3, 'gross_weight_kg':'30.000', 'volume_cbm':'0.2500',
|
||
'history_json':[{'time':'2026-10-28 10:00', 'status':'SUBMITTED'},
|
||
{'time':'2026-10-29 10:00', 'status':'PROCESSING'}]}
|
||
self.import_rows([self.record])
|
||
self.assertTrue(self.client.post('/api/customer/login',json={'username':'admin','password':'123456'}).json['ok'])
|
||
self.client.get('/api/customer/session')
|
||
|
||
def import_rows(self, rows, replace=False):
|
||
connection = db.connect()
|
||
try:
|
||
with connection:
|
||
return order_data.import_records(connection, rows, replace=replace)
|
||
finally:
|
||
connection.close()
|
||
|
||
def events(self, message, cid=None):
|
||
response = self.client.post('/stream_chat', json={'msg':message, 'conv_id':cid})
|
||
self.assertEqual(response.status_code, 200)
|
||
return [json.loads(frame[6:]) for frame in response.get_data(as_text=True).split('\n\n') if frame.startswith('data: ')]
|
||
|
||
def test_identifiers_are_exact_case_insensitive_and_preserve_suffixes(self):
|
||
for value in ('fl202610290001', 'INB20261029001-01', 'shipment00208'):
|
||
self.assertEqual(order_data.lookup_order(value)[0]['order_no'], self.record['order_no'])
|
||
self.assertEqual(self.events('查询 '+value)[0]['card']['no'], self.record['order_no'])
|
||
self.assertEqual(order_data.lookup_order('FL20261029000'), [])
|
||
self.assertEqual(order_data.extract_reference('查询 INB20261029001-01'), ['INB20261029001-01'])
|
||
self.import_rows([{**self.record,'order_no':'BOX100x100x100'}])
|
||
self.assertEqual(self.events('查询 BOX100x100x100')[0]['card']['no'],'BOX100x100x100')
|
||
|
||
def test_weight_dates_and_attachment_urls_are_not_order_identifiers(self):
|
||
for text in ('2000KG海运报价', '100000 kg海运报价', '30.000000 KG', '2026-10-29',
|
||
'2026-10-29T16:53:00', '100x100x100cm', '1200 × 800 × 600 mm',
|
||
'1.2x0.8x0.6m', '[文件]/uploads/20261029100145_6787.txt'):
|
||
self.assertEqual(order_data.extract_reference(text), [])
|
||
events = self.events('帮我查询物流轨迹')
|
||
self.assertIn('入仓码', events[0]['t'])
|
||
self.assertEqual(events[-1], {'d':1})
|
||
|
||
def test_direct_inbound_lookup_and_followup_keep_the_inbound_identity(self):
|
||
cid=self.client.post('/api/new_chat').json['id']
|
||
card=self.events(self.record['inbound_code'],cid)[0]['card']
|
||
self.assertEqual(card['no'],self.record['order_no'])
|
||
self.assertTrue(card['summary'].startswith('入仓码 '+self.record['inbound_code']))
|
||
self.assertEqual(next(item['value'] for item in card['fields'] if item['key']=='number'),self.record['order_no'])
|
||
self.import_rows([{**self.record,'gross_weight_kg':'42.000'}],replace=True)
|
||
followup=self.events('现在到哪了',cid)[0]['card']
|
||
self.assertTrue(followup['summary'].startswith('入仓码 '+self.record['inbound_code']))
|
||
self.assertEqual(next(item['value'] for item in followup['fields'] if item['key']=='weight'),'42.000 kg')
|
||
|
||
def test_missing_inbound_code_is_not_replaced_by_the_internal_order_number(self):
|
||
self.import_rows([{**self.record,'inbound_code':None}],replace=True)
|
||
card=self.events(self.record['order_no'])[0]['card']
|
||
self.assertFalse(any(item['key']=='inbound' for item in card['fields']))
|
||
self.assertTrue(card['summary'].startswith('订单号 '+self.record['order_no']))
|
||
|
||
def test_order_identifiers_remain_queryable_alongside_quantities_and_dates(self):
|
||
for message in ('查询 FL202610290001，2026-10-29T16:53:00 提交的',
|
||
'FL202610290001 30.000000 KG', 'FL202610290001 100x100x100cm',
|
||
'FL202610290001 运费多少钱？'):
|
||
with self.subTest(message=message):
|
||
event = self.events(message)[0]
|
||
if '运费' in message:
|
||
self.assertIn('未记录实际运费', event['t'])
|
||
else:
|
||
self.assertEqual(event['card']['no'], self.record['order_no'])
|
||
|
||
def test_unknown_orders_never_produce_fake_tracking_or_call_model(self):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
events = self.events('查询 UNKNOWN20260001')
|
||
self.assertIn('未找到', events[0]['t'])
|
||
self.assertFalse(any('card' in event for event in events))
|
||
self.assertEqual(events[-1], {'d':1})
|
||
model.assert_not_called()
|
||
|
||
def test_unimplemented_bulk_queries_ask_for_identifier_instead_of_inventing_records(self):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
events = self.events('列出所有订单')
|
||
self.assertIn('请提供入仓码', events[0]['t'])
|
||
model.assert_not_called()
|
||
|
||
def test_server_tenant_scope_and_ambiguous_identifiers(self):
|
||
self.import_rows([{**self.record, 'tenant_id':'tenant-b', 'gross_weight_kg':'99'}])
|
||
fields = self.events(self.record['order_no'])[0]['card']['fields']
|
||
self.assertEqual(next(field['value'] for field in fields if field['key']=='weight'), '30.000 kg')
|
||
self.app.config['ORDER_DATA_TENANT_ID'] = 'tenant-a'
|
||
rows = order_data.lookup_order(self.record['order_no'])
|
||
self.assertEqual(len(rows), 1)
|
||
self.assertEqual(rows[0]['gross_weight_kg'], '30.000')
|
||
self.app.config['ORDER_DATA_TENANT_ID'] = 'unknown-tenant'
|
||
self.assertEqual(order_data.lookup_order(self.record['order_no']), [])
|
||
|
||
def test_manual_changes_are_visible_and_reimport_preserves_them_by_default(self):
|
||
connection = db.connect()
|
||
connection.execute("UPDATE order_test_data SET gross_weight_kg='42.000' WHERE order_no=?", (self.record['order_no'],))
|
||
connection.commit()
|
||
connection.close()
|
||
self.assertEqual(self.import_rows([self.record]), {'imported':0, 'preserved':1})
|
||
fields = self.events(self.record['order_no'])[0]['card']['fields']
|
||
self.assertEqual(next(field['value'] for field in fields if field['key']=='weight'), '42.000 kg')
|
||
self.import_rows([self.record], replace=True)
|
||
self.assertEqual(order_data.lookup_order(self.record['order_no'])[0]['gross_weight_kg'], '30.000')
|
||
|
||
def test_new_table_records_can_be_added_without_touching_source_database(self):
|
||
self.import_rows([{**self.record, 'order_no':'LOCAL20260002', 'shipment_id':'localship002'}])
|
||
self.assertEqual(self.events('查询 LOCAL20260002')[0]['card']['no'], 'LOCAL20260002')
|
||
|
||
def test_business_status_history_does_not_claim_delivery_or_invent_transport_events(self):
|
||
self.import_rows([{**self.record, 'status':'COMPLETED'}], replace=True)
|
||
card = self.events(self.record['order_no'])[0]['card']
|
||
self.assertEqual(card['status'], '头程订单已完成')
|
||
self.assertNotIn('已签收', json.dumps(card, ensure_ascii=False))
|
||
self.assertEqual([node['time'] for node in card['nodes']], ['2026-10-29 10:00','2026-10-28 10:00'])
|
||
self.assertTrue(all(not node['place'] for node in card['nodes']))
|
||
|
||
def test_children_do_not_reuse_parent_total_weight(self):
|
||
self.import_rows([{**self.record, 'order_no':'CHILD20260001-01', 'order_role':'CHILD',
|
||
'gross_weight_kg':None, 'package_count':None, 'volume_cbm':None,
|
||
'parent_order_no':self.record['order_no'], 'data_note':'子单独立货物数据未提供'}])
|
||
card = self.events('CHILD20260001-01')[0]['card']
|
||
self.assertFalse(any(field['key']=='weight' for field in card['fields']))
|
||
self.assertEqual(next(field['value'] for field in card['fields'] if field['key']=='parent'), self.record['order_no'])
|
||
|
||
def test_tracking_queries_and_followups_read_linked_facts_without_calling_model(self):
|
||
self.import_rows([{**self.record, 'tracking_json': [
|
||
{'code': 'DEPARTED_ORIGIN_PORT', 'actual_at': '2026-10-28 10:00', 'shipment_key': '1'},
|
||
{'code': 'ARRIVED_DESTINATION_PORT', 'estimated_at': '2026-11-02 10:00', 'shipment_key': '1'},
|
||
]}], replace=True)
|
||
cid = self.client.post('/api/new_chat').json['id']
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
for message in (self.record['order_no']+' 物流轨迹', '到哪里了？', '那到港了吗？', '这票签收了吗？',
|
||
'轨迹呢？', '现在到哪了？', '什么时候到港？', '预计什么时候到？', '还要多久？'):
|
||
card = self.events(message, cid)[0]['card']
|
||
self.assertEqual(card['node_kind'], 'logistics')
|
||
self.assertEqual(card['tracking_status'], '船舶已离开起运港')
|
||
self.assertEqual(card['nodes'][0]['text'], '船舶已离开起运港')
|
||
self.assertEqual(card['estimated_nodes'][0]['text'], '预计到达目的港')
|
||
self.assertIn('预计到达目的港', card_to_text(CARD_PREFIX+json.dumps(card)))
|
||
self.assertNotIn('已签收', json.dumps(card, ensure_ascii=False))
|
||
model.assert_not_called()
|
||
self.assertIsNone(order_data.answer_query('美国30KG报价', cid))
|
||
|
||
def test_followups_refresh_local_data_and_remain_within_conversation(self):
|
||
cid = self.client.post('/api/new_chat').json['id']
|
||
self.events(self.record['order_no'], cid)
|
||
connection = db.connect()
|
||
connection.execute("UPDATE order_test_data SET gross_weight_kg='45.000' WHERE order_no=?", (self.record['order_no'],))
|
||
connection.commit()
|
||
connection.close()
|
||
card = self.events('这票重量是多少？',cid)[0]['card']
|
||
self.assertEqual(card['no'], self.record['order_no'])
|
||
self.assertEqual(next(field['value'] for field in card['fields'] if field['key']=='weight'), '45.000 kg')
|
||
self.assertIn('未记录实际运费', self.events('这票运费多少钱？',cid)[0]['t'])
|
||
self.assertEqual(self.events('那状态呢？',cid)[0]['card']['no'], self.record['order_no'])
|
||
self.events('查询 UNKNOWN20260001',cid)
|
||
self.assertIn('未找到', self.events('这票重量呢？',cid)[0]['t'])
|
||
other_cid = self.client.post('/api/new_chat').json['id']
|
||
self.assertIsNone(order_data.answer_query('这票重量是多少？', other_cid))
|
||
|
||
def test_new_shipping_questions_do_not_reuse_previous_order(self):
|
||
cid = self.client.post('/api/new_chat').json['id']
|
||
self.events(self.record['order_no'], cid)
|
||
for message in ('现在有30KG货物想发美国，运费多少？', '美国海运费用怎么算？', '报价30公斤到德国', '物流状态是什么意思？'):
|
||
with self.subTest(message=message):
|
||
self.assertIsNone(order_data.answer_query(message, cid))
|
||
db.save_message(cid, 'user', '美国海运费用怎么算？')
|
||
db.save_message(cid, 'assistant', '海运费用与重量及目的地有关。')
|
||
self.assertIsNone(order_data.answer_query('那状态呢？', cid))
|
||
|
||
def test_short_order_followups_read_facts_without_calling_model(self):
|
||
cid = self.client.post('/api/new_chat').json['id']
|
||
self.events(self.record['order_no'], cid)
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
for message in ('多少公斤？', '有几箱？', '体积呢？', '那状态呢？'):
|
||
with self.subTest(message=message):
|
||
self.assertEqual(self.events(message, cid)[0]['card']['no'], self.record['order_no'])
|
||
model.assert_not_called()
|
||
|
||
def test_card_storage_and_model_context_contain_only_selected_facts(self):
|
||
card = self.events(self.record['order_no'])[0]['card']
|
||
text = card_to_text(CARD_PREFIX+json.dumps(card))
|
||
self.assertIn('30.000', text)
|
||
self.assertIn('US', text)
|
||
self.assertNotIn('测试', text)
|
||
self.assertNotIn('tenant_id', text)
|
||
self.assertNotIn('customer_name', json.dumps(card))
|
||
|
||
def test_invalid_history_is_safe_and_read_failure_is_explicit(self):
|
||
connection = db.connect()
|
||
connection.execute("UPDATE order_test_data SET history_json='invalid'")
|
||
connection.commit()
|
||
self.assertEqual(self.events(self.record['order_no'])[0]['card']['nodes'], [])
|
||
connection.execute('DROP TABLE order_test_data')
|
||
connection.commit()
|
||
connection.close()
|
||
self.assertIn('暂时无法读取', self.events(self.record['order_no'])[0]['t'])
|
||
|
||
def test_import_failure_rolls_back_partial_rows(self):
|
||
with self.assertRaises(ValueError):
|
||
self.import_rows([{**self.record,'order_no':'ROLLBACK20260001'}, {'order_no':''}])
|
||
self.assertEqual(order_data.lookup_order('ROLLBACK20260001'), [])
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|