"""Mixed-domain answers and displayed quote references preserve scope and facts.""" from pathlib import Path import tempfile import unittest from unittest.mock import patch from logistics_chat import create_app,db,order_data from logistics_chat.quote_data import import_snapshot from test_quote_timing import timing_snapshot from test_quotes import frames class ConversationCompositionTests(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':'a','RATE_LIMIT_PER_MIN':1000}) self.client=self.app.test_client() self.client.post('/api/visitor/session',json={'key':'3'*64}) self.cid=self.client.post('/api/new_chat').json['id'] with self.app.app_context(): connection=db.connect() import_snapshot(connection,timing_snapshot(),activate=True) order_data.import_records(connection,[ {'tenant_id':'a','order_no':'MIXORDER20260001','inbound_code':'MIXIN20260001','status':'PROCESSING', 'exception_reason':'等待资料','source_updated_at':'2026-11-02 10:00'}, {'tenant_id':'a','order_no':'MIXORDER20260002','inbound_code':'MIXIN20260002','status':'PROCESSING', 'source_updated_at':'2026-11-01 10:00'}, {'tenant_id':'b','order_no':'SECRETORDER20260001','exception_reason':'SECRET_REASON'}, ]) connection.commit() connection.close() def login(self): self.assertTrue(self.client.post('/api/customer/login',json={'username':'admin','password':'123456'}).json['ok']) def ask(self,question): with patch('logistics_chat.ai_client.post_completion') as model: events=frames(self.client.post('/stream_chat',json={'msg':question,'conv_id':self.cid})) model.assert_not_called() self.assertEqual(events[-1],{'d':1},events) self.assertNotIn('SECRET',str(events)) self.assertNotIn('plan_option',str(events)) return events[0] def test_guest_gets_public_quote_and_login_prompt_without_order_reads(self): with patch('logistics_chat.order_listing.read_orders') as read: answer=self.ask('我有多少个订单，同时美国1000KG义乌美东海派最快多少钱') read.assert_not_called() self.assertTrue(answer['login_required']) self.assertIn('1000KG = 10000元',answer['t']) self.assertNotIn('MIXORDER',answer['t']) def test_customer_gets_both_answers_and_logout_protects_the_combined_history(self): self.login() answer=self.ask('我有多少个订单，同时美国1000KG义乌美东海派最快多少钱') self.assertIn('共有 2 票订单',answer['t']) self.assertIn('1000KG = 10000元',answer['t']) self.assertTrue(self.client.get('/api/get_messages',query_string={'conv_id':self.cid}).json) self.client.post('/api/customer/logout',json={}) self.assertEqual(self.client.get('/api/get_messages',query_string={'conv_id':self.cid}).json,[]) def test_reverse_domain_order_and_followup_quantity_keep_both_results(self): self.login() answer=self.ask('美国1000KG义乌美东海派最快多少钱，顺便我有多少异常订单') self.assertIn('1 票明确标记异常',answer['t']) self.assertIn('1000KG = 10000元',answer['t']) self.assertIn('1100KG = 11000元',self.ask('再加100KG')['t']) def test_mixed_reply_requires_domain_for_bare_ordinal_and_supports_explicit_ones(self): self.login() self.ask('我有多少个订单，同时美国1000KG义乌美东海派最快和最便宜多少钱') ambiguous=self.ask('第二个呢')['t'] self.assertIn('订单',ambiguous) self.assertIn('报价方案2',ambiguous) self.assertEqual(self.ask('第二个订单')['card']['no'],'MIXORDER20260002') quote=self.ask('第二个方案')['t'] self.assertIn('1000KG = 6000元',quote) self.assertNotIn('美森正班海派',quote) def test_second_displayed_plan_remains_selected_through_acknowledgement_and_edits(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') selected=self.ask('第二个方案呢')['t'] self.assertIn('当前按方案2继续核算',selected) self.assertIn('1000KG = 6000元',selected) self.ask('就选这个') self.assertIn('25-30天左右提取',self.ask('多久能提取')['t']) self.assertIn('1100KG = 6600元',self.ask('再加100KG')['t']) first=self.ask('第一个方案')['t'] self.assertIn('1100KG = 11000元',first) self.assertNotIn('普船海派',first) def test_prompted_plan_number_is_accepted_and_ambiguous_confirmation_is_clarified(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') self.assertIn('方案1、方案2',self.ask('就这个')['t']) selected=self.ask('我选方案2')['t'] self.assertIn('1000KG = 6000元',selected) self.assertIn('普船海派',selected) def test_priority_change_after_a_numbered_selection_can_switch_carriers(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') self.ask('方案1') text=self.ask('那我想省钱')['t'] self.assertIn('普船海派',text) self.assertNotIn('美森正班海派',text) self.assertIn('1000KG = 6000元',text) def test_english_displayed_options_remain_english(self): self.ask('1000kg to USA by sea, fastest and cheapest prices please') selected=self.ask('Option 2')['t'] self.assertIn('Regular sea parcel',selected) self.assertIn('Continuing with option 2',selected) self.assertNotRegex(selected,r'[\u4e00-\u9fff]') def test_english_compact_difference_contains_no_chinese_text(self): from logistics_chat.quote_data import load_book from logistics_chat.quote_engine import parse_intent,variants,render_rates with self.app.app_context(): book=load_book() intent=parse_intent('1000kg to USA by sea fastest and slowest prices',set()) intent['mode']='difference' destinations=variants(intent,book) text=render_rates(intent,book,destinations,[None]*len(destinations)) self.assertNotRegex(text,r'[\u4e00-\u9fff]') self.assertIn('4000 CNY',text) def test_out_of_range_plan_does_not_choose_a_route(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') text=self.ask('第9个方案')['t'] self.assertIn('展示了2个方案',text) self.assertNotIn('KG =',text) def test_displayed_choice_preserves_carrier_identity_after_prices_change(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') with self.app.app_context(): connection=db.connect() connection.execute("UPDATE quote_rates SET price='3' WHERE service='美森正班海派'") connection.commit() connection.close() selected=self.ask('第二个方案')['t'] self.assertIn('普船海派',selected) self.assertIn('1000KG = 6000元',selected) self.assertNotIn('美森正班海派',selected) def test_removed_displayed_carrier_requires_a_new_selection_through_followups(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') with self.app.app_context(): connection=db.connect() connection.execute("DELETE FROM quote_rates WHERE service='普船海派'") connection.commit() connection.close() self.assertIn('暂无可用报价',self.ask('方案2')['t']) self.assertIn('暂无可用报价',self.ask('那具体多少钱')['t']) text=self.ask('全部渠道报价')['t'] self.assertIn('美森正班海派',text) self.assertNotIn('暂无可用报价',text) def test_new_quote_root_replaces_the_old_plan_catalog(self): self.ask('美国1000KG义乌美东海派最快和最便宜多少钱') self.ask('新的一批200KG美国义乌美西海派报价') selected=self.ask('第二个方案')['t'] self.assertIn('合德海派',selected) self.assertIn('200KG = 1200元',selected) self.assertNotIn('1000KG',selected) def test_legacy_unnumbered_rows_keep_their_positions_after_a_carrier_is_removed(self): from logistics_chat.quote_plans import displayed_options from logistics_chat.quote_data import load_book content='美东：\n- 旧渠道 · 义乌 · 包税：5元/KG\n- 普船海派 · 义乌 · 包税：6元/KG' with self.app.app_context(): options=displayed_options(content,load_book()) self.assertEqual([option['number'] for option in options],[1,2]) self.assertEqual(options[0]['rows'],[]) self.assertEqual(options[1]['rows'][0]['service'],'普船海派') def test_removed_warehouse_does_not_reuse_the_previous_destination(self): from logistics_chat.quote_plans import displayed_options from logistics_chat.quote_data import load_book content='美东：\n- 普船海派 · 义乌 · 包税：6元/KG\nABC1 · 东1区：\n- 普船海派 · 义乌 · 包税：6元/KG' with self.app.app_context(): options=displayed_options(content,load_book()) self.assertEqual(options[1]['warehouse'],'ABC1') self.assertTrue(options[1]['destination_unavailable']) def test_different_visitor_cannot_use_private_plan_bindings(self): self.login() self.ask('我有多少订单，同时美国1000KG义乌美东海派最快和最便宜多少钱') other=self.app.test_client() other.post('/api/visitor/session',json={'key':'4'*64}) with patch('logistics_chat.quote_plans.history_bindings') as read: response=frames(other.post('/stream_chat',json={'msg':'方案2','conv_id':self.cid})) read.assert_not_called() self.assertTrue(any(event.get('e') for event in response)) def test_known_box_count_and_weights_are_not_requested_again(self): text=self.ask('美国10箱每箱20KG义乌美西普船海派报价')['t'] question=text.splitlines()[-1] self.assertIn('每箱长宽高',question) self.assertIn('目的仓码或邮编',question) self.assertNotIn('箱数',question) self.assertNotIn('重量',question) text=self.ask('每箱50x40x30cm')['t'] question=text.splitlines()[-1] self.assertNotIn('长宽高',question) self.assertNotIn('重量',question) def test_all_known_shipment_fields_produce_no_repeated_followup(self): text=self.ask('美国10箱每箱20KG每箱50x40x30cm，ZZZ1，义乌普船海派报价')['t'] self.assertNotIn('请补充',text) def test_selected_warehouse_variant_preserves_its_zone_and_assigned_weight(self): snapshot=timing_snapshot() snapshot['source_sha256']='2'*64 snapshot['warehouses'].append({**snapshot['warehouses'][0],'code':'ALT1','zone':'西1区','postal':'90001'}) with self.app.app_context(): connection=db.connect() import_snapshot(connection,snapshot,activate=True) connection.commit() connection.close() self.ask('美国ZZZ1 100KG，ALT1 200KG，义乌海派报价，总体积10CBM') # ZZZ1 has two zones, with three parcel options in each zone. selected=self.ask('方案4')['t'] self.assertIn('ZZZ1 · 东2区',selected) self.assertNotIn('东1区',selected) self.assertNotIn('ALT1',selected) self.assertIn('100KG = 1100元',selected) self.assertNotIn('300KG',selected) def test_difference_is_short_and_details_expand_without_losing_conditions(self): self.ask('美国1000KG海派最快和最慢报价') text=self.ask('快的贵多少')['t'] self.assertLess(len(text),350) self.assertIn('差额为4000元',text) self.assertIn('15天、25–30天',text) self.assertNotIn('1000KG =',text) details=self.ask('展开报价明细')['t'] self.assertIn('1000KG = 10000元',details) self.assertIn('1000KG = 6000元',details)