"""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()