474 lines
29 KiB
Python
474 lines
29 KiB
Python
"""Statistics use explicit exception reasons and actual, order-level progress."""
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
import sqlite3
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from logistics_chat import create_app,db,order_data
|
||
from logistics_chat.order_tracking import current_stage
|
||
|
||
|
||
def fact(code,shipment='',forecast=False):
|
||
return {'code':code,'shipment_key':shipment,
|
||
'estimated_at' if forecast else 'actual_at':'2026-10-01 10:00'}
|
||
|
||
|
||
class OrderStatisticsTests(unittest.TestCase):
|
||
def setUp(self):
|
||
folder=tempfile.TemporaryDirectory();self.addCleanup(folder.cleanup);root=Path(folder.name)
|
||
self.app=create_app({'TESTING':True,'DATABASE':root/'data.db','UPLOAD_DIR':root/'uploads',
|
||
'BACKUP_DIR':root/'backups','LOG_DIR':root/'logs','DEEPSEEK_API_KEY':'',
|
||
'CUSTOMER_DEMO_TENANT_ID':'a','RATE_LIMIT_PER_MIN':1000})
|
||
self.client=self.app.test_client();self.client.post('/api/visitor/session',json={'key':'1'*64})
|
||
rows=[{'tenant_id':'a','order_no':f'STAT2026{i:04}','inbound_code':f'INSTAT2026{i:04}',
|
||
'status':'PROCESSING','source_updated_at':f'2026-10-{i+1:02} 10:00',
|
||
'tracking_json':[fact('IN_TRANSIT'),fact('ARRIVED_DESTINATION_PORT'),fact('EMPTY_RETURNED',forecast=True)],
|
||
'exception_reason':f'待核实资料{i}' if i<6 else None} for i in range(10)]
|
||
rows += [{'tenant_id':'a','order_no':'CLEAR20260001','status':'PROCESSING',
|
||
'tracking_json':[fact('ARRIVED_DESTINATION_PORT'),fact('PICKUP_READY')],'exception_reason':'等待客户资料'},
|
||
{'tenant_id':'a','order_no':'CLEAR20260002','status':'PROCESSING','tracking_json':[fact('PICKUP_READY')]},
|
||
{'tenant_id':'a','order_no':'MIXED20260001','tracking_json':[fact('ARRIVED_DESTINATION_PORT','A'),fact('PICKUP_READY','B')]},
|
||
{'tenant_id':'a','order_no':'FORECAST20260001','tracking_json':[fact('ARRIVED_DESTINATION_PORT',forecast=True)]},
|
||
{'tenant_id':'a','order_no':'BUSINESS20260001','status':'COMPLETED','tracking_json':'[]'},
|
||
{'tenant_id':'a','order_no':'REJECTED20260001','status':'REJECTED','exception_reason':' \t\n'},
|
||
{'tenant_id':'b','order_no':'SECRET20260001','status':'PROCESSING','tracking_json':[fact('ARRIVED_DESTINATION_PORT')],'exception_reason':'不能泄露'}]
|
||
with self.app.app_context():
|
||
c=db.connect();order_data.import_records(c,rows);c.commit();c.close()
|
||
|
||
def login(self):
|
||
self.assertTrue(self.client.post('/api/customer/login',json={'username':'admin','password':'123456'}).json['ok'])
|
||
|
||
def ask(self,q,cid=None):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
response=self.client.post('/stream_chat',json={'msg':q,'conv_id':cid})
|
||
events=[json.loads(line[6:]) for line in response.get_data(as_text=True).split('\n\n') if line.startswith('data: ')]
|
||
model.assert_not_called()
|
||
self.assertEqual(events[-1],{'d':1},events)
|
||
self.assertNotIn('_order_list_state',str(events));self.assertNotIn('_customer_data',str(events))
|
||
self.assertNotIn('_order_reference',str(events))
|
||
self.assertNotIn('SECRET',str(events));self.assertNotIn('不能泄露',str(events))
|
||
return events[0]
|
||
|
||
def test_guests_never_read_aggregate_or_exception_facts(self):
|
||
with patch('logistics_chat.order_listing.read_orders') as read:
|
||
for q in ('我有多少异常订单','到港的有多少运单','各个轨迹节点分别有多少订单','查看全部','下一页'):
|
||
self.assertTrue(self.ask(q)['login_required'])
|
||
read.assert_not_called()
|
||
|
||
def test_overview_is_requested_in_chat_and_saved_in_customer_protected_history(self):
|
||
with patch('logistics_chat.order_listing.read_overview') as read:
|
||
for question in ('我的订单概览','查看订单情况','Order overview'):
|
||
self.assertTrue(self.ask(question)['login_required'])
|
||
read.assert_not_called()
|
||
self.login()
|
||
with patch('logistics_chat.order_listing.read_overview') as read:
|
||
self.assertNotIn('overview',self.client.get('/api/customer/session?overview=1').json)
|
||
read.assert_not_called()
|
||
cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask('我的订单概览',cid)['t']
|
||
self.assertIn('共有 16 票订单，履约中 12 票，异常订单 7 票',text)
|
||
self.assertIn('船舶已到目的港：10票',text)
|
||
self.assertEqual(text.count('- 入仓码：'),5)
|
||
self.assertEqual(text.count('异常原因：'),5)
|
||
response=self.client.get('/api/get_messages',query_string={'conv_id':cid})
|
||
self.assertIn('no-store',response.headers['Cache-Control'])
|
||
self.assertTrue(any(row['role']=='assistant' and row['content']==text for row in response.json))
|
||
self.client.post('/api/customer/logout')
|
||
self.assertEqual(self.client.get('/api/get_messages',query_string={'conv_id':cid}).json,[])
|
||
self.login()
|
||
self.assertTrue(any(row['content']==text for row in self.client.get('/api/get_messages',query_string={'conv_id':cid}).json))
|
||
|
||
def test_overview_keeps_old_exception_ahead_of_new_processing_orders_and_refreshes(self):
|
||
self.login()
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
c.execute("UPDATE order_test_data SET exception_reason=NULL WHERE tenant_id='a' AND order_no<>'STAT20260000'")
|
||
c.commit();c.close()
|
||
from logistics_chat.order_listing import read_overview
|
||
with self.app.app_context():overview=read_overview(('a',))
|
||
self.assertEqual(overview['orders'][0]['order_no'],'STAT20260000')
|
||
self.assertEqual(len({order['order_no'] for order in overview['orders']}),5)
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
order_data.import_records(c,[{'tenant_id':'a','order_no':'NEW20260001','status':'PROCESSING','exception_reason':'资料待核实'}])
|
||
c.commit();c.close()
|
||
refreshed=self.ask('订单概览')['t']
|
||
self.assertIn('共有 17 票订单，履约中 13 票，异常订单 2 票',refreshed)
|
||
|
||
def test_overview_read_failure_keeps_login_available_and_no_facts_leak(self):
|
||
self.login()
|
||
with patch('logistics_chat.order_listing.read_overview',side_effect=sqlite3.OperationalError('private details')):
|
||
result=self.ask('订单概览')
|
||
self.assertTrue(self.client.get('/api/customer/session').json['authenticated'])
|
||
self.assertIn('暂时无法读取',result['t'])
|
||
self.assertNotIn('共有',result['t'])
|
||
self.assertNotIn('private details',str(result))
|
||
|
||
def test_summaries_show_at_most_five_recent_rows_and_keep_intersecting_filters(self):
|
||
self.login()
|
||
total=self.ask('我有多少个订单')['t']
|
||
self.assertIn('共有 16 票',total);self.assertEqual(total.count('- 入仓码：'),5)
|
||
abnormal=self.ask('我有多少个异常订单')['t']
|
||
self.assertIn('7 票明确标记异常',abnormal);self.assertEqual(abnormal.count('- 入仓码：'),5)
|
||
self.assertNotIn('REJECTED20260001',abnormal)
|
||
arrived=self.ask('现在到港的有多少运单')['t']
|
||
self.assertIn('10 票当前节点为“船舶已到目的港”',arrived)
|
||
self.assertEqual(arrived.count('- 入仓码：'),5)
|
||
self.assertIn('STAT20260009',arrived);self.assertNotIn('STAT20260000',arrived)
|
||
self.assertNotIn('CLEAR20260001',arrived)
|
||
text=self.ask('已清关的异常订单有多少')['t']
|
||
self.assertIn('1 票当前节点为“已清关、可提柜”',text);self.assertIn('等待客户资料',text)
|
||
|
||
def test_combined_counts_answer_each_question_and_expand_the_last_filter(self):
|
||
question='我的订单有多少个，异常有多少个'
|
||
with patch('logistics_chat.order_listing.read_orders') as read:
|
||
self.assertTrue(self.ask(question)['login_required'])
|
||
self.assertTrue(self.ask('异常的有多少条？')['login_required'])
|
||
read.assert_not_called()
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask(question,cid)['t']
|
||
self.assertIn('共有 16 票订单',text)
|
||
self.assertIn('7 票明确标记异常',text)
|
||
self.assertEqual(text.count('- 入仓码：'),5)
|
||
self.assertEqual(text.count('异常原因：'),5)
|
||
self.assertIn('共7票',self.ask('查看全部',cid)['t'])
|
||
self.assertIn('7 票明确标记异常',self.ask('异常的有多少条？',cid)['t'])
|
||
self.assertIn('共7票',self.ask('查看全部',cid)['t'])
|
||
|
||
def test_spoken_count_questions_tolerate_greetings_and_locate_all_current_nodes(self):
|
||
self.login()
|
||
text=self.ask('你好你好，现在我的订单在哪里了？有多少个？')['t']
|
||
self.assertIn('共有 16 票订单',text)
|
||
self.assertIn('船舶已到目的港：10票',text)
|
||
self.assertIn('暂无实际节点：3票',text)
|
||
self.assertIn('共有 16 票订单',self.ask('或者订单有多少条？')['t'])
|
||
|
||
def test_combined_statistics_keep_node_distribution_before_exception_count(self):
|
||
self.login()
|
||
text=self.ask('我的订单各个轨迹节点分别有多少，异常的有多少条')['t']
|
||
self.assertIn('共有 16 票订单',text)
|
||
self.assertIn('船舶已到目的港：10票',text)
|
||
self.assertIn('已清关、可提柜：2票',text)
|
||
self.assertIn('暂无实际节点：3票',text)
|
||
self.assertIn('7 票明确标记异常',text)
|
||
self.assertEqual(text.count('- 入仓码：'),5)
|
||
|
||
def test_subset_followups_keep_previous_stage_and_show_recorded_reasons(self):
|
||
self.assertTrue(self.ask('异常的呢')['login_required'])
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('已清关的订单有多少',cid)
|
||
text=self.ask('其中异常的呢',cid)['t']
|
||
self.assertIn('1 票当前节点为“已清关、可提柜”',text)
|
||
self.assertIn('等待客户资料',text)
|
||
self.assertNotIn('待核实资料',text)
|
||
self.assertIn('1 票当前节点',self.ask('这些一共多少票',cid)['t'])
|
||
text=self.ask('异常原因呢',cid)['t']
|
||
self.assertIn('等待客户资料',text)
|
||
self.assertNotIn('业务状态：',text)
|
||
self.assertIn('共1票',self.ask('查看全部',cid)['t'])
|
||
|
||
def test_ordinal_uses_the_displayed_reply_even_when_new_orders_arrive(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask('我有多少异常订单',cid)['t']
|
||
shown=re.findall(r'^- 入仓码：([^；]+)；订单号：([^；]+)；',text,re.M)
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
order_data.import_records(c,[{'tenant_id':'a','order_no':'NEWEST20260001','inbound_code':'NEWESTIN20260001',
|
||
'status':'PROCESSING','source_updated_at':'2030-01-01','exception_reason':'新记录'}])
|
||
c.commit();c.close()
|
||
card=self.ask('第一票的轨迹',cid)['card']
|
||
self.assertEqual(card['no'],shown[0][1])
|
||
card=self.ask('第二个呢',cid)['card']
|
||
self.assertEqual(card['no'],shown[1][1])
|
||
self.assertNotIn('NEWEST',str(card))
|
||
|
||
def test_selected_order_survives_acknowledgements_and_long_detail_followups(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
first=self.ask('第一个呢',cid)['card']['no']
|
||
for _ in range(8):
|
||
self.ask('谢谢',cid)
|
||
text=self.ask('异常原因呢',cid)['t']
|
||
self.assertNotIn('请提供',text)
|
||
self.assertEqual(self.ask('重量呢',cid)['card']['no'],first)
|
||
|
||
def test_short_choice_confirmation_preserves_the_selected_order_and_navigation(self):
|
||
self.login()
|
||
cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
first=self.ask('第一个呢',cid)['card']['no']
|
||
second=self.ask('下一票呢',cid)['card']['no']
|
||
self.ask('就这个',cid)
|
||
with self.app.app_context():
|
||
connection=db.connect()
|
||
reason=connection.execute('SELECT exception_reason FROM order_test_data WHERE order_no=?',(second,)).fetchone()[0]
|
||
connection.close()
|
||
self.assertIn(reason,self.ask('异常原因呢',cid)['t'])
|
||
self.assertEqual(self.ask('上一票呢',cid)['card']['no'],first)
|
||
|
||
def test_empty_or_out_of_range_ordinals_never_choose_an_unshown_order(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
self.assertIn('展示了5票',self.ask('第6票的轨迹',cid)['t'])
|
||
self.ask('已还柜的订单有多少',cid)
|
||
self.assertIn('当前列表没有',self.ask('第一个呢',cid)['t'])
|
||
self.assertIn('0 票当前节点为“已还柜”',self.ask('到哪了',cid)['t'])
|
||
|
||
def test_ordinals_and_subsets_require_the_current_owned_topic_and_login(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
self.ask('美国30KG海运报价',cid)
|
||
text=self.ask('第一票的轨迹',cid)['t']
|
||
self.assertIn('请先查看',text)
|
||
self.assertNotIn('INSTAT',text)
|
||
self.ask('我有多少异常订单',cid)
|
||
self.client.post('/api/customer/logout')
|
||
self.assertTrue(self.ask('第一个呢',cid)['login_required'])
|
||
self.assertTrue(self.ask('其中异常的呢',cid)['login_required'])
|
||
|
||
def test_conflicting_subset_conditions_request_a_scope_change(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('已清关的订单有多少',cid)
|
||
self.assertIn('确认要切换筛选条件',self.ask('其中已到港的呢',cid)['t'])
|
||
self.assertIn('10 票当前节点',self.ask('那已到港的呢',cid)['t'])
|
||
|
||
def test_subset_clarification_keeps_the_original_scope_for_the_next_answer(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('已清关的订单有多少',cid)
|
||
self.assertIn('请明确',self.ask('其中今天的异常呢',cid)['t'])
|
||
self.assertIn('1 票当前节点',self.ask('其中异常的呢',cid)['t'])
|
||
|
||
def test_a_shipping_quote_starting_with_these_is_a_new_quote_topic(self):
|
||
from logistics_chat.quote_data import import_snapshot
|
||
from test_quotes import fixture_snapshot
|
||
with self.app.app_context():
|
||
c=db.connect();import_snapshot(c,fixture_snapshot(),activate=True);c.commit();c.close()
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少个订单',cid)
|
||
text=self.ask('这些货物美国30KG海运报价',cid)['t']
|
||
self.assertIn('美国 30KG',text)
|
||
self.assertNotIn('订单范围',text)
|
||
self.assertIn('请先查看',self.ask('第一个订单呢',cid)['t'])
|
||
|
||
def test_legacy_list_state_can_resolve_ordinals_without_saved_identifiers(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask('我有多少异常订单',cid)['t']
|
||
first=re.findall(r'^- 入仓码：([^；]+)；订单号：([^；]+)；',text,re.M)[0][1]
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
state=json.loads(c.execute('SELECT order_list_state FROM chat_history WHERE id=?',(cid,)).fetchone()[0])
|
||
del state['shown']
|
||
c.execute('UPDATE chat_history SET order_list_state=? WHERE id=?',(json.dumps(state),cid))
|
||
c.commit();c.close()
|
||
self.assertEqual(self.ask('第一个呢',cid)['card']['no'],first)
|
||
|
||
def test_ordinal_uses_saved_identifiers_when_a_field_contains_display_separators(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
order_data.import_records(c,[{'tenant_id':'a','order_no':'STRUCTURED20260001',
|
||
'inbound_code':'VALUE；订单号：OTHER20260001','status':'PROCESSING',
|
||
'source_updated_at':'2030-01-01','exception_reason':'待核实'}])
|
||
c.commit();c.close()
|
||
self.ask('我有多少异常订单',cid)
|
||
self.assertEqual(self.ask('第一个呢',cid)['card']['no'],'STRUCTURED20260001')
|
||
|
||
def test_adjacent_order_navigation_keeps_the_displayed_rows_and_current_selection(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask('我有多少异常订单',cid)['t']
|
||
shown=re.findall(r'^- 入仓码：([^；]+)；订单号：([^；]+)；',text,re.M)
|
||
self.assertEqual(self.ask('第一个呢',cid)['card']['no'],shown[0][1])
|
||
self.assertEqual(self.ask('下一票呢',cid)['card']['no'],shown[1][1])
|
||
self.ask('谢谢',cid)
|
||
self.assertEqual(self.ask('上一票的轨迹',cid)['card']['no'],shown[0][1])
|
||
self.assertIn('未登记申报毛重',self.ask('那这个有多重',cid)['t'])
|
||
self.assertEqual(self.ask('下一票的轨迹',cid)['card']['no'],shown[1][1])
|
||
|
||
def test_adjacent_navigation_stops_at_visible_edges_and_resets_on_page_changes(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
order_data.import_records(c,[{'tenant_id':'a','order_no':f'PAGEX2026{i:04}','inbound_code':f'PAGEIN2026{i:04}',
|
||
'status':'PROCESSING'} for i in range(30)])
|
||
c.commit();c.close()
|
||
self.ask('列出我的全部订单',cid)
|
||
self.ask('第20票的轨迹',cid)
|
||
self.assertIn('最后一票',self.ask('下一票呢',cid)['t'])
|
||
text=self.ask('下一页',cid)['t']
|
||
first=re.findall(r'^- 入仓码：([^；]+)；订单号：([^；]+)；',text,re.M)[0][1]
|
||
self.assertEqual(self.ask('第一个呢',cid)['card']['no'],first)
|
||
self.assertIn('第一票',self.ask('上一票呢',cid)['t'])
|
||
|
||
def test_adjacent_navigation_needs_an_owned_list_and_a_selected_row(self):
|
||
for question in ('下一票呢','上一票的轨迹'):
|
||
self.assertTrue(self.ask(question)['login_required'])
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.assertIn('先查看订单列表',self.ask('下一票呢',cid)['t'])
|
||
self.ask('我有多少异常订单',cid)
|
||
self.assertIn('先在当前列表中选择',self.ask('下一票呢',cid)['t'])
|
||
self.ask('第一个呢',cid)
|
||
self.client.post('/api/customer/logout')
|
||
self.assertTrue(self.ask('下一票呢',cid)['login_required'])
|
||
|
||
def test_navigation_and_weights_do_not_reuse_orders_after_a_quote_topic(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
self.ask('第一个呢',cid)
|
||
self.ask('美国30KG海运报价',cid)
|
||
self.assertIn('先查看订单列表',self.ask('下一票呢',cid)['t'])
|
||
|
||
def test_implicit_count_without_an_established_range_asks_for_the_missing_scope(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
text=self.ask('异常的呢',cid)['t']
|
||
self.assertIn('说明要统计的订单范围',text)
|
||
self.assertNotIn('7 票',text)
|
||
|
||
def test_list_selection_keeps_the_order_when_its_old_inbound_code_is_reassigned(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
order_data.import_records(c,[{'tenant_id':'a','order_no':'ORIGINAL20260001','inbound_code':'REUSEDINBOUND1',
|
||
'status':'PROCESSING','source_updated_at':'2030-01-01','exception_reason':'原订单资料'},
|
||
{'tenant_id':'a','order_no':'OTHER20260001','status':'PROCESSING',
|
||
'source_updated_at':'2029-01-01','exception_reason':'另一票资料'}])
|
||
c.commit();c.close()
|
||
self.ask('我有多少异常订单',cid)
|
||
with self.app.app_context():
|
||
c=db.connect()
|
||
c.execute("UPDATE order_test_data SET inbound_code='NEWINBOUND1' WHERE tenant_id='a' AND order_no='ORIGINAL20260001'")
|
||
c.execute("UPDATE order_test_data SET inbound_code='REUSEDINBOUND1' WHERE tenant_id='a' AND order_no='OTHER20260001'")
|
||
c.commit();c.close()
|
||
card=self.ask('第一个呢',cid)['card']
|
||
self.assertEqual(card['no'],'ORIGINAL20260001')
|
||
self.assertIn({'key':'inbound','value':'NEWINBOUND1'},card['fields'])
|
||
self.assertIn('原订单资料',self.ask('异常原因呢',cid)['t'])
|
||
self.assertEqual(self.ask('下一票呢',cid)['card']['no'],'OTHER20260001')
|
||
|
||
def test_adjacent_selection_follows_actual_cards_after_overlapping_requests(self):
|
||
from logistics_chat.tracking import CARD_PREFIX
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
second=self.ask('第二个呢',cid)['card']
|
||
self.ask('第一个呢',cid)
|
||
with self.app.app_context():
|
||
for _ in range(2):
|
||
db.save_message(cid,'user','下一票呢')
|
||
db.save_message(cid,'assistant',CARD_PREFIX+json.dumps(second,ensure_ascii=False))
|
||
third=self.ask('下一票呢',cid)['card']['no']
|
||
self.assertEqual(third,self.ask('第三个呢',cid)['card']['no'])
|
||
|
||
def test_ambiguous_speech_answers_the_clear_count_and_requests_the_missing_meaning(self):
|
||
self.login()
|
||
text=self.ask('当前，我的订单有多少条，一层有多少条？')['t']
|
||
self.assertIn('共有 16 票订单',text)
|
||
self.assertIn('“一层有多少条”的筛选条件尚未明确',text)
|
||
self.assertNotIn('7 票明确标记异常',text)
|
||
text=self.ask('我的订单有多少个，今天异常订单有多少个')['t']
|
||
self.assertIn('共有 16 票订单',text)
|
||
self.assertIn('“今天异常订单有多少个”的筛选条件尚未明确',text)
|
||
|
||
def test_distribution_counts_each_order_once_and_never_uses_forecasts_or_business_completion(self):
|
||
self.login()
|
||
text=self.ask('各个轨迹节点分别有多少订单')['t']
|
||
for token in ('船舶已到目的港：10票','已清关、可提柜：2票','多柜进度不同：1票','暂无实际节点：3票'):
|
||
self.assertIn(token,text)
|
||
self.assertNotIn('已还柜：',text);self.assertNotIn('已签收：',text)
|
||
|
||
def test_english_homepage_questions_apply_their_intended_filters(self):
|
||
self.assertTrue(self.ask('How many exception orders do I have?')['login_required'])
|
||
self.login()
|
||
self.assertIn('7 票明确标记异常',self.ask('How many exception orders do I have?')['t'])
|
||
text=self.ask('How many orders are at each tracking stage?')['t']
|
||
self.assertIn('船舶已到目的港：10票',text)
|
||
self.assertIn('暂无实际节点：3票',text)
|
||
|
||
def test_expand_and_pages_keep_filters_and_refresh_counts(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
text=self.ask('下一页',cid)['t']
|
||
self.assertIn('第2页',text);self.assertEqual(text.count('- 入仓码：'),2)
|
||
text=self.ask('查看全部',cid)['t']
|
||
self.assertIn('共7票',text);self.assertEqual(text.count('- 入仓码：'),7)
|
||
self.assertIn('明确标记异常',text)
|
||
with self.app.app_context():
|
||
c=db.connect();c.execute('UPDATE order_test_data SET exception_reason=NULL WHERE tenant_id=?',('a',));c.commit();c.close()
|
||
self.assertIn('没有符合条件',self.ask('下一页',cid)['t'])
|
||
|
||
def test_pages_do_not_cross_conversations_or_login_boundaries(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('到港的订单有多少',cid)
|
||
self.assertIn('第2页',self.ask('下一页',cid)['t'])
|
||
self.ask('当前订单总数',cid)
|
||
self.assertIn('共16票',self.ask('查看全部',cid)['t'])
|
||
self.client.post('/api/customer/logout')
|
||
self.assertTrue(self.ask('下一页',cid)['login_required'])
|
||
self.assertEqual(self.client.get('/api/get_messages',query_string={'conv_id':cid}).json,[])
|
||
self.login();self.assertIn('共16票',self.ask('查看全部',cid)['t'])
|
||
other=self.client.post('/api/new_chat').json['id']
|
||
self.assertIn('请先',self.ask('下一页',other)['t'])
|
||
|
||
def test_unknown_or_ambiguous_filters_do_not_degenerate_to_totals(self):
|
||
self.login()
|
||
for q in ('今天有多少异常订单','未到港的订单有多少','到港和已清关分别有多少订单','已签收的有多少运单'):
|
||
with self.subTest(q=q):self.assertIn('暂不支持按这个条件统计',self.ask(q)['t'])
|
||
|
||
def test_exception_reasons_are_editable_and_old_reimports_preserve_them(self):
|
||
self.login()
|
||
with self.app.app_context():
|
||
c=db.connect();order_data.import_records(c,[{'tenant_id':'a','order_no':'CLEAR20260001','status':'PROCESSING'}],replace=True);c.commit();c.close()
|
||
self.assertIn('等待客户资料',self.ask('CLEAR20260001的异常原因是什么')['t'])
|
||
cid=self.client.post('/api/new_chat').json['id'];self.ask('CLEAR20260001',cid)
|
||
self.assertIn('等待客户资料',self.ask('为什么异常',cid)['t'])
|
||
with self.app.app_context():
|
||
c=db.connect();order_data.import_records(c,[{'tenant_id':'a','order_no':'CLEAR20260001','exception_reason':''}],replace=True);c.commit();c.close()
|
||
self.assertIn('未登记异常原因',self.ask('CLEAR20260001的异常原因是什么')['t'])
|
||
|
||
def test_missing_actuals_and_bad_tracking_do_not_count_as_arrived(self):
|
||
self.assertEqual(current_stage([fact('ARRIVED_DESTINATION_PORT','A'),fact('PICKUP_READY','B',True)]),'MIXED')
|
||
self.assertEqual(current_stage([fact('ARRIVED_DESTINATION_PORT',forecast=True)]),'NO_ACTUAL')
|
||
self.assertEqual(current_stage('{broken'),'NO_ACTUAL')
|
||
self.assertEqual(current_stage([fact('UNKNOWN')]),'NO_ACTUAL')
|
||
self.assertEqual(current_stage([fact('DOMESTIC_RECEIVED'),fact('ARRIVED_DESTINATION_PORT','A',True)]),'DOMESTIC_RECEIVED')
|
||
|
||
def test_node_names_do_not_collide_with_business_status_or_drop_negations(self):
|
||
from logistics_chat.order_query import parse_query
|
||
intent=parse_query('已完成装柜的订单有多少')
|
||
self.assertFalse(intent['unsupported']);self.assertIsNone(intent['status'])
|
||
self.assertEqual(intent['stage'],'LOADING_COMPLETED')
|
||
self.assertTrue(parse_query('未清关的订单有多少')['unsupported'])
|
||
self.assertTrue(parse_query('没有异常的订单有多少')['unsupported'])
|
||
|
||
def test_state_survives_app_restart_and_rejects_late_reply_updates(self):
|
||
self.login();cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('我有多少异常订单',cid)
|
||
restarted=create_app(dict(self.app.config,INITIALIZE_RUNTIME=False))
|
||
client=restarted.test_client()
|
||
# Copy opaque cookies without exporting them.
|
||
for cookie in self.client._cookies.values():client.set_cookie(cookie.key,cookie.value)
|
||
response=client.post('/stream_chat',json={'msg':'下一页','conv_id':cid}).get_data(as_text=True)
|
||
self.assertIn('第2页',response)
|
||
from logistics_chat.order_listing import save_state
|
||
cookies='; '.join(cookie.key+'='+cookie.value for cookie in self.client._cookies.values())
|
||
with self.app.test_request_context('/',headers={'Cookie':cookies}):
|
||
self.app.preprocess_request()
|
||
from logistics_chat.customer_service import current_customer_id
|
||
self.assertIsNotNone(current_customer_id())
|
||
c=db.connect();prior=c.execute('SELECT order_list_state FROM chat_history WHERE id=?',(cid,)).fetchone()[0]
|
||
state=json.loads(prior);c.close()
|
||
db.save_message(cid,'user','换个话题，我要报价')
|
||
save_state(cid,{'query':state['query'],'page':1,'size':5},state['message_id'])
|
||
c=db.connect();self.assertEqual(c.execute('SELECT order_list_state FROM chat_history WHERE id=?',(cid,)).fetchone()[0],prior);c.close()
|
||
|
||
def test_schema_migration_preserves_old_orders(self):
|
||
c=sqlite3.connect(':memory:')
|
||
try:
|
||
c.execute('CREATE TABLE order_test_data (id INTEGER PRIMARY KEY,tenant_id TEXT,order_no TEXT,status TEXT,inbound_code TEXT,shipment_id TEXT)')
|
||
c.execute("INSERT INTO order_test_data VALUES(1,'a','OLD20260001','PROCESSING',NULL,NULL)")
|
||
order_data.init_order_table(c);order_data.init_order_table(c)
|
||
self.assertEqual(c.execute('SELECT order_no,exception_reason,tracking_json FROM order_test_data').fetchone(),('OLD20260001',None,'[]'))
|
||
finally:c.close()
|