"""Common pre-sale and order questions; all data and monetary values are synthetic.""" from pathlib import Path 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.quote_data import import_snapshot from test_quotes import fixture_snapshot,frames class CustomerQuestionTests(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':'A10'}) documents=['铁铝制品另外加附加费￥5/KG','客户自行送货，不提供上门提货服务', '外箱要贴两张入仓条码与我司唛头标签，代贴收操作费', '产品上和包装上必须有Made in China产地标识', '单独报关需提供资料：装箱单、发票、报关委托书、合同', '如遇查验，不包税货物的查验费用按客户所占货物比例均摊', '正常货物 0KG<单件重量<22KG 0CM<最长单边长度<120CM 0CM<第二边<75CM 最长边<260CM', '超重收+70/件'] book['documents']=[{'sheet':'合成渠道','cell':f'A{i}','text':text} for i,text in enumerate(documents)] self.order={'tenant_id':'qa-a','order_no':'COMMON20260001','inbound_code':'QACOMMON20260001', 'status':'COMPLETED','package_count':2,'country_code':'US', 'tracking_json':[{'code':'DEPARTED_ORIGIN_PORT','actual_at':'2026-10-01 10:00'}, {'code':'ARRIVED_DESTINATION_PORT','estimated_at':'2026-10-20 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':'COMMON20260099','inbound_code':'QACOMMON20260099'}]) c.commit();c.close() def conversation(self):return self.client.post('/api/new_chat').json['id'] def events(self,text,cid=None): with patch('logistics_chat.ai_client.post_completion') as model: result=frames(self.client.post('/stream_chat',json={'msg':text,'conv_id':cid})) model.assert_not_called() self.assertEqual(result[-1],{'d':1},result) return result def ask(self,text,cid=None):return self.events(text,cid)[0]['t'] def login(self): self.assertTrue(self.client.post('/api/customer/login',json={'username':'admin','password':'123456'}).json['ok']) def test_explicit_weights_are_billed_by_destination_and_not_reused_as_each_total(self): for text in ('美国20KG发ZZZ1，10KG发YYY2，义乌普船海派', '美国ZZZ1：20KG，YYY2：10KG，义乌普船海派'): answer=self.ask(text) self.assertIn('美国 30KG',answer) self.assertIn('20KG = 100元',answer) self.assertIn('12KG = 24元',answer) self.assertNotIn('30KG =',answer) self.assertIn('分配重量不一致',self.ask('美国总重40KG，20KG发ZZZ1，10KG发YYY2，海派')) def test_allocation_follows_origin_changes_but_a_single_new_destination_clears_it(self): cid=self.conversation() self.ask('美国20KG发ZZZ1，10KG发YYY2，义乌普船海派',cid) text=self.ask('深圳交仓呢',cid) self.assertIn('20KG = 120元',text);self.assertIn('12KG = 36元',text) text=self.ask('换成YYY2',cid) self.assertNotIn('ZZZ1',text);self.assertIn('美国 30KG',text) def test_allocated_cartons_keep_individual_minimums_and_overweight_fees(self): text=self.ask('美国2箱每箱5KG发ZZZ1，1箱20KG发YYY2，义乌普船海派') self.assertIn('美国 30KG',text);self.assertIn('24KG = 120元',text);self.assertIn('20KG = 40元',text) text=self.ask('美国1箱30KG发ZZZ1，1箱10KG发YYY2，义乌普船海派') self.assertIn('30KG = 220元',text);self.assertIn('最高项70元/件',text) text=self.ask('美国1箱10KG发ZZZ1，60x50x40cm，1箱20KG发YYY2，40x30x20cm，义乌普船海派') self.assertIn('20KG = 100元',text);self.assertIn('20KG = 40元',text) def test_total_weight_correction_removes_old_per_carton_weight_even_on_next_followup(self): cid=self.conversation() self.ask('美国美西2箱每箱30KG，义乌普船海派',cid) text=self.ask('30公斤是两箱加起来的，不是每箱30公斤',cid) self.assertIn('美国 30KG',text);self.assertNotIn('60KG',text) text=self.ask('深圳交仓呢',cid) self.assertIn('美国 30KG',text);self.assertNotIn('60KG',text) self.assertIn('美国 50KG',self.ask('不是30KG，是50KG',cid)) def test_uniform_carton_count_change_preserves_piece_weight_across_followups(self): cid=self.conversation() self.ask('美国美西2箱每箱15KG，义乌普船海派',cid) self.assertIn('美国 60KG',self.ask('改成4箱',cid)) self.assertIn('美国 60KG',self.ask('深圳交仓呢',cid)) self.assertIn('美国 60KG',self.ask('改成6箱，总重不变',cid)) self.assertIn('美国 60KG',self.ask('义乌交仓呢',cid)) def test_mixed_carton_count_requires_new_weights_then_accepts_a_new_total(self): cid=self.conversation() self.ask('美国美西1箱10KG，1箱20KG，义乌普船海派',cid) self.assertIn('请确认新的总重',self.ask('改成4箱',cid)) text=self.ask('总重40KG',cid) self.assertIn('美国 40KG',text);self.assertIn('48KG = 96元',text) other=self.conversation() self.ask('美国美西1箱10KG，1箱20KG，义乌普船海派',other) self.ask('改成4箱，总重不变',other) self.assertIn('美国 30KG',self.ask('深圳交仓呢',other)) def test_new_dimensions_do_not_resurrect_stale_volume(self): cid=self.conversation() self.ask('美国363KG发ZZZ1，1箱，1.5方，义乌不包税海卡',cid) self.assertIn('1CBM = 100元',self.ask('改成每箱50x50x40cm',cid)) self.assertIn('1CBM = 100元',self.ask('深圳交仓呢',cid)) def test_cbm_metal_surcharge_is_separate_instead_of_silently_included(self): text=self.ask('美国铁制品30KG发ZZZ1，1方，义乌不包税海卡') self.assertIn('1CBM = 100元',text) self.assertIn('铁铝附加费为5元/KG',text) self.assertIn('当前按方运费未包含',text) self.assertNotIn('= 105元',text) def test_preparation_questions_use_us_material_without_promising_ship_dates(self): cases=(('美国可以上门提货吗','不提供上门提货'),('美国FBA外箱怎么贴标签','两张入仓条码'), ('美国单独报关资料需要什么','装箱单'),('美国申报比例怎么算','不能按统一比例'), ('美国不包税查验费用谁承担','比例均摊'),('美国这周哪天开船','联系人工核对')) for question,expected in cases: with self.subTest(question=question):self.assertIn(expected,self.ask(question)) cid=self.conversation();self.ask('美国30KG海派报价',cid) self.assertIn('唛头',self.ask('外箱标签呢',cid)) self.assertIn('目的国家',self.ask('可以上门提货吗')) self.assertIn('普船海派',self.ask('你们能发美国吗，有哪些渠道')) text=self.ask('美国美西30KG最便宜的，月底必须到，义乌海派') self.assertIn('30KG = 60元',text);self.assertIn('不能作为',text) def test_order_business_status_does_not_imply_signing_shelving_or_piece_delivery(self): self.assertIn('业务完成不等于',self.ask('头程订单已完成是不是签收了')) self.assertTrue(self.events('这票上架了吗')[0]['login_required']) self.login();cid=self.conversation() self.events(self.order['inbound_code'],cid) text=self.ask('这票已完成为什么还没上架',cid) self.assertIn('已离开起运港',text) self.assertIn('不等于收件人签收',text) self.assertNotIn('已经到港',text) text=self.ask('怎么三天都没更新',cid) self.assertIn('2026-10-01',text);self.assertIn('原因无法',text) text=self.ask('到了几箱？少了一箱',cid) self.assertIn('登记总件数为2件',text);self.assertIn('没有逐箱签收明细',text) text=self.ask('帮我催一下',cid) self.assertIn('尚未提交',text) self.assertIn('未找到',self.ask('QACOMMON20260099上架了吗')) self.client.post('/api/customer/logout') self.assertEqual(self.client.get('/api/get_messages',query_string={'conv_id':cid}).json,[])