255 lines
16 KiB
Python
255 lines
16 KiB
Python
"""Oral-language, US-region isolation and scoped-page regressions using synthetic facts."""
|
||
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
import re
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from logistics_chat import create_app,db
|
||
from logistics_chat.order_data import import_records
|
||
from logistics_chat.pricing import detect_dim_unit,dims_to_cm
|
||
from logistics_chat.quote_data import import_snapshot
|
||
from logistics_chat.quote_intent import parse_intent
|
||
from logistics_chat.quote_engine import quantity
|
||
from test_quotes import fixture_snapshot,frames
|
||
|
||
|
||
class CustomerGapTests(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':'qa-a','RATE_LIMIT_PER_MIN':1000})
|
||
self.client=self.app.test_client();self.client.post('/api/visitor/session',json={'key':'1'*64})
|
||
book=fixture_snapshot()
|
||
book['warehouses'].append({'code':'YYY2','zone':'西1区','state':'AA','postal':'90001',
|
||
'address':'','source_sheet':'合成分区','source_cell':'A8'})
|
||
texts=['偏远地址加收4元/kg；超偏远地址加收6元/kg','铁铝制品另外加附加费￥5/KG',
|
||
'正常货物 0KG<单件重量<22KG 0CM<最长单边长度<120CM 0CM<第二边<75CM 最长边<260CM',
|
||
'超重收+70/件','客户自行交货，不提供上门提货服务',
|
||
'禁止运输的货物：纯电池、液体','需熏蒸的货物由客户在备货阶段自行完成']
|
||
book['documents']=[{'sheet':'合成渠道','cell':f'A{i+1}','text':text} for i,text in enumerate(texts)]
|
||
self.order={'tenant_id':'qa-a','order_no':'FOCUS20260001','inbound_code':'QAFOCUS20260001',
|
||
'country_code':'US','status':'COMPLETED','package_count':2,
|
||
'tracking_json':[{'code':'IN_TRANSIT','actual_at':'2026-10-01 10:00'}]}
|
||
with self.app.app_context():
|
||
c=db.connect();import_snapshot(c,book,activate=True)
|
||
import_records(c,[self.order,{**self.order,'tenant_id':'qa-b','order_no':'OTHER20260099','inbound_code':'QAOTHER20260099'}])
|
||
c.commit();c.close()
|
||
|
||
def conversation(self):return self.client.post('/api/new_chat').json['id']
|
||
|
||
def events(self,q,cid=None,client=None):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
result=frames((client or self.client).post('/stream_chat',json={'msg':q,'conv_id':cid}))
|
||
model.assert_not_called()
|
||
self.assertEqual(result[-1],{'d':1},result)
|
||
return result
|
||
|
||
def ask(self,q,cid=None):return self.events(q,cid)[0]['t']
|
||
|
||
def login(self):
|
||
self.assertTrue(self.client.post('/api/customer/login',json={'username':'admin','password':'123456'}).json['ok'])
|
||
|
||
def test_us_natural_questions_never_enter_old_regional_model_faqs(self):
|
||
cases=('你们这个报价是按毛重还是净重？','运费能便宜一点吗？','这个价格含税含派送吗？',
|
||
'一共多少钱，给我最终总价','这个物流服务范围具体怎么执行？')
|
||
for q in cases:
|
||
with self.subTest(q=q):
|
||
cid=self.conversation();self.ask('美国美西30KG，义乌普船海派',cid)
|
||
text=self.ask(q,cid)
|
||
self.assertNotIn('21KG',text);self.assertNotIn('250',text);self.assertNotIn('7元/KG',text)
|
||
text=self.ask('我想发点货到美国，怎么收费？')
|
||
self.assertIn('元/KG',text);self.assertIn('多少公斤',text)
|
||
|
||
def test_final_total_respects_tax_labels_and_keeps_unconfirmed_fees_separate(self):
|
||
cid=self.conversation();self.ask('美国美西2箱每箱15KG，40x30x20cm，义乌普船海派',cid)
|
||
text=self.ask('一共多少钱，给我最终总价',cid)
|
||
self.assertIn('30KG = 60元',text);self.assertIn('· 包税',text)
|
||
self.assertNotIn('价格不含税',text);self.assertIn('最终应付总价',text)
|
||
self.assertIn('折扣或合作价格需人工确认',self.ask('运费能便宜一点吗',cid))
|
||
|
||
def test_negative_warehouse_and_origin_edits_do_not_keep_rejected_values(self):
|
||
cid=self.conversation();self.ask('美国30KG发ZZZ1，深圳普船海派',cid)
|
||
text=self.ask('不是发ZZZ1，是YYY2',cid)
|
||
self.assertNotIn('ZZZ1',text);self.assertIn('YYY2',text)
|
||
text=self.ask('不要深圳，还是义乌交仓',cid)
|
||
self.assertIn('· 义乌',text);self.assertNotIn('· 华南',text)
|
||
text=self.ask('50KG呢',cid)
|
||
self.assertIn('YYY2',text);self.assertNotIn('ZZZ1',text);self.assertIn('美国 50KG',text)
|
||
|
||
def test_inches_convert_before_charge_weight_and_acceptance_limits(self):
|
||
intent=parse_intent('美国30KG，60x40x30英寸，1箱普船海派',set())
|
||
self.assertEqual(intent['dim_unit'],'in')
|
||
self.assertEqual(dims_to_cm([60,40,30],'in'),[152.4,101.6,76.2])
|
||
self.assertGreater(quantity(intent,{'kind':'parcel','unit':'kg'}),Decimal('190'))
|
||
text=self.ask('美国美西30KG，60x40x30英寸，1箱普船海派，义乌')
|
||
self.assertIn('人工确认是否接收',text)
|
||
for text in ('60x40x30 inches','60x40x30in','60x40x30"'):
|
||
with self.subTest(text=text):
|
||
self.assertEqual(detect_dim_unit(text),'in')
|
||
self.assertIn('人工确认是否接收',self.ask('美国美西30KG，'+text+'，1箱普船海派，义乌'))
|
||
self.assertEqual(detect_dim_unit('60x40x30，合计1立方米'),'cm')
|
||
self.assertEqual(detect_dim_unit('60x40x30，合计1m3'),'cm')
|
||
self.assertEqual(detect_dim_unit('1.2x0.8x0.6m呢'),'m')
|
||
|
||
def test_spoken_carton_and_dimensions_are_used_without_reasking(self):
|
||
intent=parse_intent('美国30KG一个箱子，60长40宽30高，怎么算？',set())
|
||
self.assertEqual(intent['boxes'],1);self.assertEqual(intent['dims'],[60,40,30])
|
||
text=self.ask('美国美西30KG一个箱子，60长40宽30高，义乌普船海派')
|
||
self.assertIn('最高项70元/件',text)
|
||
self.assertNotIn('请补充箱数',text)
|
||
|
||
def test_box_units_are_local_and_each_axis_can_carry_its_own_unit(self):
|
||
cases=(('12英寸显示器，包装60x40x30cm',[60,40,30],'cm',Decimal('30')),
|
||
('80cm×70cm×60cm',[80,70,60],'cm',Decimal('56')),
|
||
('80cmx70cmx60cm',[80,70,60],'cm',Decimal('56')),
|
||
('800mm*700mm*600mm',[800,700,600],'mm',Decimal('56')),
|
||
('长80厘米宽70厘米高60厘米',[80,70,60],'cm',Decimal('56')),
|
||
('0.8m×70cm×600mm',[80,70,60],'cm',Decimal('56')),
|
||
('12英寸显示器，包装60x40x30',[60,40,30],'cm',Decimal('30')))
|
||
for wording,dims,unit,charge in cases:
|
||
with self.subTest(wording=wording):
|
||
intent=parse_intent('美国美西30KG，'+wording+'，1箱义乌普船海派',set())
|
||
self.assertEqual(intent['dims'],dims);self.assertEqual(intent['dim_unit'],unit)
|
||
self.assertEqual(quantity(intent,{'kind':'parcel','unit':'kg'}),charge)
|
||
text=self.ask('美国美西30KG，'+wording+'，1箱义乌普船海派')
|
||
self.assertIn(f'{charge}KG =',text)
|
||
self.assertNotIn('请补充箱数',text)
|
||
|
||
def test_oral_partial_weight_edits_never_overwrite_the_whole_shipment(self):
|
||
cid=self.conversation()
|
||
self.ask('美国美西30KG，部分铁制品，铁铝部分计费重10KG，义乌普船海派',cid)
|
||
for q,amount in (('铁铝部分还是10KG',110),('铁铝部分改成8KG',100),('铁铝部分仍是8KG',100)):
|
||
text=self.ask(q,cid)
|
||
self.assertIn('美国 30KG',text);self.assertIn(f'30KG = {amount}元',text)
|
||
self.assertIn('大于0',self.ask('铁铝部分改为-1KG',cid))
|
||
fresh=self.conversation()
|
||
self.assertIn('请确认铁铝部分的计费重',self.ask('美国美西30KG，铁铝部分还是10KG，义乌普船海派',fresh))
|
||
self.assertNotIn('×10KG',self.ask('一共多少钱，给我最终总价',fresh))
|
||
|
||
def test_long_quote_chain_retains_conditions_but_topic_changes_stop_reuse(self):
|
||
cid=self.conversation();self.ask('美国美西30KG，义乌普船海派',cid)
|
||
for _ in range(15):self.assertIn('每箱不足12KG',self.ask('这个报价按毛重还是净重？',cid))
|
||
self.assertIn('美国 50KG',self.ask('50KG呢',cid))
|
||
self.assertIn('美国渠道',self.ask('这个物流服务范围具体怎么执行？',cid))
|
||
self.assertIn('请先',self.ask('下一页',cid))
|
||
from logistics_chat.quote_service import answer_quote
|
||
from logistics_chat.quote_scope import answer_us_fallback
|
||
with self.app.test_request_context('/'):
|
||
self.assertIsNone(answer_quote('新的一票50KG发德国报价',cid))
|
||
db.save_message(cid,'user','换个话题，我要查订单')
|
||
self.assertIsNone(answer_us_fallback('50KG呢',cid))
|
||
|
||
def test_comparing_two_weights_is_not_treated_as_two_cartons(self):
|
||
cid=self.conversation();self.ask('美国30KG海派报价',cid)
|
||
text=self.ask('单箱12.5KG会不会按13KG收费？',cid)
|
||
self.assertIn('按原小数计费，不向上取整',text)
|
||
self.assertIn('12.5KG',text)
|
||
self.assertNotIn('请明确总重',text)
|
||
|
||
def test_confirmed_fractional_weight_and_partial_metal_basis_are_used(self):
|
||
self.assertIn('12.5KG = 25元',self.ask('美国美西1箱12.5KG，40x30x20cm，义乌普船海派'))
|
||
self.assertIn('12.3456KG = 24.69元',self.ask('美国美西1箱12.3456KG，40x30x20cm，义乌普船海派'))
|
||
cid=self.conversation()
|
||
text=self.ask('美国美西30KG，部分铁制品，义乌普船海派',cid)
|
||
self.assertIn('30KG = 60元',text);self.assertIn('不能按整票重量加费',text)
|
||
self.assertNotIn('30KG = 210元',text)
|
||
text=self.ask('铁铝部分计费重10KG',cid)
|
||
self.assertIn('美国 30KG',text);self.assertIn('30KG = 110元',text)
|
||
self.assertIn('×10KG=50元',text)
|
||
self.assertIn('30KG = 110元',self.ask('一共多少钱，给我最终总价',cid))
|
||
text=self.ask('改成40KG',cid)
|
||
self.assertIn('当前金额未包含该项',text);self.assertNotIn('×10KG',text)
|
||
text=self.ask('全部换成铁制品40KG',cid)
|
||
self.assertIn('40KG = 280元',text)
|
||
|
||
def test_unknown_metal_and_cbm_quotes_do_not_spread_partial_fee_over_the_whole_load(self):
|
||
text=self.ask('美国美西30KG，只是金属扣子，整票都按铁铝加费吗，义乌普船海派')
|
||
self.assertIn('只按铁铝货物的计费重',text);self.assertIn('确认金属是否属于铁铝',text)
|
||
text=self.ask('美国30KG发ZZZ1，1方不包税海卡，部分铁制品，铁铝部分计费重10KG，义乌')
|
||
self.assertIn('1CBM = 150元',text);self.assertIn('×10KG=50元',text)
|
||
self.assertIn('大于0',self.ask('美国30KG，部分铁制品，铁铝部分计费重-1KG，义乌海派'))
|
||
text=self.ask('美国20KG发ZZZ1，10KG发YYY2，部分铁制品，铁铝部分计费重10KG，义乌普船海派')
|
||
self.assertIn('不能重复套用到每个仓',text)
|
||
self.assertIn('20KG = 100元',text);self.assertIn('12KG = 24元',text)
|
||
self.assertNotIn('铁铝附加费5元/KG×10KG',text)
|
||
|
||
def test_booking_phone_invoice_and_file_capabilities_do_not_invent_workflows(self):
|
||
cases=(('就按刚才方案帮我下单','尚未创建订单'),('手机号能查吗','暂不支持手机号'),
|
||
('你们可以开发票吗，开票加几个点','开票税点'),('付款账号发我一下','收款账号'),
|
||
('可以月结吗','月结资格'),('帮我导出订单Excel','暂不提供订单Excel导出'),
|
||
('Excel能直接读里面箱规报价吗','不会自动读取Excel'),
|
||
('我换了个手机，之前聊天记录还在吗','不会自动同步恢复'))
|
||
for q,expected in cases:
|
||
with self.subTest(q=q):self.assertIn(expected,self.ask(q))
|
||
with patch('logistics_chat.order_data.lookup_order') as lookup:
|
||
self.assertIn('暂不支持手机号',self.ask('手机号19900000000能查吗'))
|
||
lookup.assert_not_called()
|
||
with self.app.app_context():
|
||
c=db.connect();self.assertEqual(c.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0],2);c.close()
|
||
|
||
def test_address_and_pickup_questions_are_not_prices_and_english_stays_english(self):
|
||
cid=self.conversation();self.ask('美国30KG海派报价',cid)
|
||
self.assertIn('地址、联系人和收货时间',self.ask('义乌仓具体地址是什么？几点可以送？',cid))
|
||
text=self.ask('Can you pick up my goods in Shenzhen for US sea freight?')
|
||
self.assertIn('does not offer pickup',text);self.assertNotIn('CNY/kg',text)
|
||
self.assertFalse(re.search(r'[\u4e00-\u9fff]',text))
|
||
|
||
def test_air_battery_question_does_not_apply_sea_battery_requirements(self):
|
||
text=self.ask('想走美国空运，包税带电的，有吗？')
|
||
self.assertIn('空运费率当前资料没有列出',text);self.assertNotIn('MSDS',text)
|
||
self.assertIn('备货阶段',self.ask('美国木制品需要熏蒸吗'))
|
||
|
||
def seed_pages(self):
|
||
with self.app.app_context():
|
||
c=db.connect();import_records(c,[{'tenant_id':'qa-a','order_no':f'PAGE2026{i:06d}',
|
||
'inbound_code':f'QAPAGE2026{i:06d}','status':'PROCESSING'} for i in range(1,26)])
|
||
c.commit();c.close()
|
||
|
||
def test_pages_use_fresh_authorized_count_and_preserve_filter_and_private_history(self):
|
||
self.seed_pages()
|
||
self.assertTrue(self.events('下一页')[0]['login_required'])
|
||
self.login();cid=self.conversation()
|
||
text=self.ask('列出我的全部订单',cid)
|
||
self.assertIn('共26票',text);self.assertEqual(text.count('- 入仓码：'),20)
|
||
self.assertNotIn('QAOTHER',text)
|
||
text=self.ask('下一页',cid)
|
||
self.assertIn('第2页',text);self.assertEqual(text.count('- 入仓码：'),6);self.assertIn('共26票',text)
|
||
self.assertIn('最后一页',self.ask('下一页',cid))
|
||
self.assertIn('第1页',self.ask('上一页',cid))
|
||
self.ask('列出履约中的订单',cid)
|
||
text=self.ask('下一页',cid)
|
||
self.assertIn('共25票',text);self.assertEqual(text.count('- 入仓码：'),5);self.assertNotIn('QAFOCUS',text)
|
||
self.assertIn('第1页',self.ask('第1页',cid))
|
||
self.client.post('/api/customer/logout')
|
||
self.assertEqual(self.client.get('/api/get_messages',query_string={'conv_id':cid}).json,[])
|
||
|
||
def test_page_commands_cannot_reuse_another_topic_or_unbounded_offset(self):
|
||
self.login();cid=self.conversation();self.ask('列出我的订单',cid)
|
||
self.assertIn('有效页码',self.ask('第0页',cid))
|
||
self.ask('列出我的订单',cid)
|
||
self.ask('美国30KG海派报价',cid)
|
||
self.assertIn('请先',self.ask('下一页',cid))
|
||
|
||
def test_count_interruptions_do_not_erase_the_active_list_page(self):
|
||
self.seed_pages();self.login();cid=self.conversation()
|
||
self.ask('列出我的全部订单',cid)
|
||
self.ask('我有多少个运单',cid)
|
||
text=self.ask('下一页',cid)
|
||
self.assertIn('第2页',text);self.assertEqual(text.count('- 入仓码：'),6)
|
||
|
||
def test_clearance_followup_modification_and_unknown_payment_do_not_call_model(self):
|
||
self.assertTrue(self.events('我的订单已清关了吗')[0]['login_required'])
|
||
self.login();cid=self.conversation();self.events('QAFOCUS20260001',cid)
|
||
event=self.events('我的订单已清关了吗？',cid)[0]
|
||
self.assertEqual(event['card']['no'],'FOCUS20260001')
|
||
self.assertEqual(event['card']['tracking_status'],'海运途中')
|
||
self.assertIn('尚未修改订单',self.ask('帮我把收货地址改一下',cid))
|
||
self.assertIn('未接入付款状态',self.ask('这票订单付过款了吗',cid))
|
||
self.assertIn('不能直接当作快递单号',self.ask('这票对应的UPS快递单号是多少',cid))
|
||
self.assertIn('暂不支持按这个物流条件',self.ask('哪些订单还没开船',cid))
|