"""Order counts are fresh, customer-scoped facts with protected history.""" 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 class OrderCountTests(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':'', 'CUSTOMER_DEMO_TENANT_ID':'tenant-a','ORDER_DATA_TENANT_ID':'','RATE_LIMIT_PER_MIN':1000}) self.a, self.b = self.app.test_client(), self.app.test_client() for client,key in ((self.a,'1'*64),(self.b,'2'*64)): client.post('/api/visitor/session',json={'key':key}) with self.app.app_context(): connection = db.connect() order_data.import_records(connection,[ {'tenant_id':'tenant-a','order_no':'COUNT20260001','inbound_code':'INB20260001','status':'PROCESSING'}, {'tenant_id':'tenant-a','order_no':'COUNT20260002','status':'COMPLETED'}, {'tenant_id':'tenant-b','order_no':'COUNT20260001','inbound_code':'INB20260099','status':'PROCESSING'}, {'tenant_id':'ungranted','order_no':'COUNT20260003','status':'COMPLETED'}]) connection.execute('INSERT INTO customer_users(username,password_hash,display_name,tenant_id) VALUES(?,?,?,?)', ('other',self.app.config['CUSTOMER_DEMO_PASSWORD_HASH'],'Other','tenant-b')) connection.commit(); connection.close() def login(self, client=None, username='admin'): self.assertTrue((client or self.a).post('/api/customer/login',json={'username':username,'password':'123456'}).json['ok']) def events(self, message, cid=None, client=None): response = (client or self.a).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_guest_counts_require_login_before_reading_any_counts(self): with patch('logistics_chat.order_data.count_orders') as count, patch('logistics_chat.ai_client.post_completion') as model: for question in ('我有多少个运单','我的订单数量','我今天有多少订单','How many shipments do I have?'): with self.subTest(question=question): self.assertTrue(self.events(question)[0]['login_required']) count.assert_not_called(); model.assert_not_called() def test_logged_in_counts_accept_common_wording_and_refresh_after_data_changes(self): self.login() cid = self.a.post('/api/new_chat').json['id'] self.events('INB20260001',cid) with patch('logistics_chat.ai_client.post_completion') as model: for question in ('我有多少个运单','我一共有几票订单？','请帮我统计一下我的订单数量', '我的运单有多少','运单数','我有多少入仓码','我想知道我有多少个运单', '能帮我查下我的运单总数吗','帮我统计下我名下的订单','How many shipments do I have?', 'What is the number of my orders?'): with self.subTest(question=question): events = self.events(question,cid) self.assertEqual(events[0],{'t':'您当前共有 2 票订单，其中 1 票已登记入仓码。'}) self.assertEqual(events[-1],{'d':1}) model.assert_not_called() with self.app.app_context(): connection = db.connect() order_data.import_records(connection,[{'tenant_id':'tenant-a','order_no':'COUNT20260004','inbound_code':'INB20260004'}]) connection.commit(); connection.close() self.assertEqual(self.events('我有多少个运单')[0]['t'],'您当前共有 3 票订单，其中 2 票已登记入仓码。') def test_business_status_filters_are_exact_and_unknown_conditions_never_return_the_total(self): self.login() with patch('logistics_chat.ai_client.post_completion') as model: completed = self.events('我有多少个已完成的订单')[0]['t'] self.assertIn('1 票业务状态为“头程订单已完成”',completed) self.assertIn('0 票已登记入仓码',completed) self.assertNotIn('签收',completed) self.assertIn('1 票业务状态为“履约中”',self.events('履约中的运单有多少')[0]['t']) self.assertIn('0 票业务状态为“已取消”',self.events('我有多少已取消订单')[0]['t']) for question in ('我今天有多少订单','我有多少已签收的运单','美国运单有多少', '我有多少未完成的订单','已完成和已取消各有多少订单'): with self.subTest(question=question): result = self.events(question)[0] self.assertIn('暂不支持按这个条件统计',result['t']) self.assertNotIn('您当前共有',result['t']) model.assert_not_called() for question in ('这个订单重量有多少','订单运费是多少','订单数量是什么意思'): self.assertIsNone(order_data.parse_count_query(question)) def test_multiple_grants_and_environment_restrictions_use_the_same_scope_as_lookup(self): with self.app.app_context(): connection = db.connect() connection.execute("UPDATE customer_users SET tenant_id='',scope_initialized=1 WHERE username='admin'") user = connection.execute("SELECT id FROM customer_users WHERE username='admin'").fetchone()[0] connection.executemany('INSERT INTO customer_tenant_grants(customer_id,tenant_id) VALUES(?,?)', [(user,'tenant-a'),(user,'tenant-b')]) connection.commit(); connection.close() self.login() self.assertIn('共有 3 票订单，其中 2 票',self.events('我有多少个运单')[0]['t']) forged = self.a.post('/stream_chat',json={'msg':'我有多少个运单','tenant_id':'ungranted'},headers={'X-Tenant-Id':'ungranted'}) self.assertIn('共有 3 票订单',forged.get_data(as_text=True)) self.app.config['ORDER_DATA_TENANT_ID']='tenant-b' self.assertTrue(self.events('我有多少个运单')[0]['login_required']) self.login() self.assertIn('共有 1 票订单，其中 1 票',self.events('我有多少个运单')[0]['t']) with self.app.app_context(): self.assertEqual(order_data.count_orders(('tenant-a','tenant-b')),{'total':1,'with_inbound':1}) for scope in (None,(),[],['ungranted'],'tenant-b'): self.assertEqual(order_data.count_orders(scope),{'total':0,'with_inbound':0}) def test_count_history_is_bound_even_without_an_order_card(self): self.login(); cid = self.a.post('/api/new_chat').json['id'] events = self.events('我有多少个运单',cid) self.assertNotIn('_customer_data',json.dumps(events)) self.a.post('/api/customer/logout') self.assertEqual(self.a.get('/api/get_messages',query_string={'conv_id':cid}).json,[]) self.assertEqual(self.a.get('/api/get_history').json,[]) self.a.delete('/api/clear_all') self.login(username='other') self.assertEqual(self.a.get('/api/get_messages',query_string={'conv_id':cid}).json,[]) self.login(self.b) self.assertEqual(self.b.get('/api/get_messages',query_string={'conv_id':cid}).json,[]) self.login() self.assertTrue(any('共有 2 票订单' in m['content'] for m in self.a.get('/api/get_messages',query_string={'conv_id':cid}).json)) def test_count_failure_and_failed_history_binding_never_emit_facts(self): self.login(); cid = self.a.post('/api/new_chat').json['id'] with patch('logistics_chat.customer_service.bind_order_conversation',return_value=False): events = self.events('我有多少个运单',cid) self.assertIn('无法访问',events[0]['e']) self.assertNotIn('共有',json.dumps(events)) with patch('logistics_chat.db.connect',side_effect=sqlite3.OperationalError('private diagnostic')): with self.app.app_context(): with self.assertRaises(order_data.OrderDataUnavailable): order_data.count_orders(('tenant-a',)) with patch('logistics_chat.order_data.count_orders',side_effect=order_data.OrderDataUnavailable): events = self.events('我有多少个运单') self.assertIn('暂时无法读取',events[0]['t']) def test_empty_data_and_blank_codes_are_counted_truthfully(self): self.login() with self.app.app_context(): connection = db.connect() connection.execute('UPDATE order_test_data SET inbound_code=? WHERE order_no=? AND tenant_id=?', (' \t\r\n','COUNT20260002','tenant-a')) order_data.import_records(connection,[{'tenant_id':'tenant-a','order_no':'COUNT20260005','inbound_code':'INB20260001'}]) connection.commit(); connection.close() self.assertEqual(self.events('我有多少个运单')[0]['t'],'您当前共有 3 票订单，其中 2 票已登记入仓码。') with self.app.app_context(): connection = db.connect() connection.execute("DELETE FROM order_test_data WHERE tenant_id='tenant-a'") connection.commit(); connection.close() self.assertEqual(self.events('我有多少个运单')[0]['t'],'您当前共有 0 票订单，其中 0 票已登记入仓码。') if __name__ == '__main__': unittest.main()