220 lines
14 KiB
Python
220 lines
14 KiB
Python
"""Statistics use explicit exception reasons and actual, order-level progress."""
|
|
|
|
import json
|
|
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('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_session_overview_is_opt_in_scoped_prioritized_and_not_cached(self):
|
|
with patch('logistics_chat.order_listing.read_overview') as read:
|
|
guest=self.client.get('/api/customer/session?overview=1&tenant_id=b',headers={'X-Tenant-Id':'b'})
|
|
self.assertFalse(guest.json['authenticated'])
|
|
self.assertNotIn('overview',guest.json)
|
|
read.assert_not_called()
|
|
self.login()
|
|
self.assertNotIn('overview',self.client.get('/api/customer/session').json)
|
|
response=self.client.get('/api/customer/session?overview=1&tenant_id=b',headers={'X-Tenant-Id':'b'})
|
|
overview=response.json['overview']
|
|
self.assertIn('no-store',response.headers['Cache-Control'])
|
|
self.assertEqual((overview['total'],overview['processing'],overview['exceptions']),(16,12,7))
|
|
self.assertEqual(sum(stage['count'] for stage in overview['stages']),12)
|
|
self.assertEqual(len(overview['orders']),5)
|
|
self.assertTrue(all(order['exception_reason'] for order in overview['orders']))
|
|
self.assertNotIn('SECRET',response.get_data(as_text=True))
|
|
self.assertNotIn('tenant_id',response.get_data(as_text=True))
|
|
self.client.post('/api/customer/logout')
|
|
self.assertNotIn('overview',self.client.get('/api/customer/session?overview=1').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()
|
|
overview=self.client.get('/api/customer/session?overview=1').json['overview']
|
|
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.client.get('/api/customer/session?overview=1').json['overview']
|
|
self.assertEqual((refreshed['total'],refreshed['processing'],refreshed['exceptions']),(17,13,2))
|
|
|
|
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.client.get('/api/customer/session?overview=1').json
|
|
self.assertTrue(result['authenticated'])
|
|
self.assertEqual(result['overview_error'],'unavailable')
|
|
self.assertNotIn('overview',result)
|
|
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_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()
|