"""Compound quote questions retain shipment facts and local service terms.""" from pathlib import Path import tempfile import unittest from unittest.mock import patch from logistics_chat import create_app,db from logistics_chat.quote_data import import_snapshot from logistics_chat.quote_intent import timing_with_claim from test_quote_timing import timing_snapshot from test_quotes import frames class QuoteConversationTests(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':'','RATE_LIMIT_PER_MIN':1000}) self.client=self.app.test_client() self.client.post('/api/visitor/session',json={'key':'9'*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) connection.commit() connection.close() 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.assertFalse(any(event.get('e') for event in events),events) return events[0]['t'] def test_fastest_and_cheapest_in_one_question_both_receive_prices(self): self.ask('一票3000KG发美国，义乌美东，价格怎么样') text=self.ask('最快的多少钱，最便宜的是什么') fast,cheap=text.split('最便宜参考方案') self.assertIn('最快参考方案',fast) self.assertIn('美森正班海派',fast) self.assertIn('3000KG = 30000元',fast) self.assertIn('普船海派',cheap) self.assertIn('3000KG = 18000元',cheap) self.assertNotIn('华南',text) def test_recommendation_keeps_quantity_and_can_be_followed_by_a_quantity_edit(self): self.ask('美国3000KG义乌美东海派报价') text=self.ask('请帮我推荐一下，怎么走比较好') self.assertIn('最快参考方案',text) self.assertIn('最便宜参考方案',text) self.assertIn('预算还是时效',text) self.assertIn('3000KG = 18000元',text) text=self.ask('再加100KG，其他不变') self.assertIn('3100KG = 18600元',text) self.assertIn('3100KG = 31000元',text) self.assertNotIn('华南',text) def test_recommendation_respects_an_explicit_budget_preference(self): self.ask('美国3000KG义乌美东海派报价') text=self.ask('推荐最便宜的') self.assertIn('优先运费',text) self.assertIn('普船海派',text) self.assertNotIn('美森正班海派',text) self.assertNotIn('预算还是时效',text) def test_answers_to_the_recommendation_preference_question_select_the_route(self): self.ask('美国3000KG义乌美东海派报价') self.ask('请帮我推荐一下') text=self.ask('预算优先') self.assertIn('普船海派',text) self.assertIn('3000KG = 18000元',text) self.assertNotIn('美森正班海派',text) text=self.ask('时效优先') self.assertIn('美森正班海派',text) self.assertIn('3000KG = 30000元',text) self.assertNotIn('普船海派',text) def test_no_book_for_a_named_us_service_does_not_use_legacy_regional_knowledge(self): with patch('logistics_chat.quote_service.load_book',return_value=None), \ patch('logistics_chat.quote_data.load_book',return_value=None): text=self.ask('美森正班海派多久能到') self.assertIn('美国报价表',text) self.assertNotIn('欧洲',text) self.assertNotIn('15天',text) def test_recommendation_without_shipment_details_asks_for_them(self): text=self.ask('请帮我推荐一下，怎么走比较好') self.assertIn('目的国家',text) self.assertIn('货物重量',text) self.assertNotIn('元/KG',text) def test_explicit_service_after_comparison_clears_comparison_labels(self): self.ask('美国3000KG义乌美东海派最快和最便宜报价') text=self.ask('美森正班海派多久能到') self.assertIn('15天左右提取',text) self.assertNotIn('最快参考方案',text) self.assertNotIn('最便宜参考方案',text) self.assertNotIn('普船海派',text) self.assertIn('3000KG = 30000元',self.ask('具体多少钱')) def test_timing_and_liability_are_answered_without_losing_the_quote(self): with self.app.app_context(): connection=db.connect() connection.execute("UPDATE quote_rates SET eta=eta||? WHERE service=?",('\n无时效赔偿','美森正班海派')) connection.commit() connection.close() self.ask('美国3000KG义乌美东海派报价') self.ask('请帮我推荐一下') text=self.ask('美森正班海派多久能到？如果到不了产生费用，你们公司会赔钱嘛？') self.assertIn('15天左右提取',text) self.assertIn('无时效赔偿',text) self.assertIn('人工根据实际承运条款核实',text) self.assertNotIn('请提供目的国家',text) text=self.ask('美森正班海派多久能到') self.assertIn('15天左右提取',text) self.assertNotIn('欧洲',text) self.assertNotIn('无时效赔偿',text) self.assertIn('3000KG = 30000元',self.ask('具体多少钱')) def test_mixed_claim_wording_with_punctuation_preserves_known_conditions(self): self.ask('美国3000KG义乌美东海派报价') for question in ('美森正班海派多久能到，延误会赔偿吗？', '美森正班海派多久能到？产生的费用谁负责？'): text=self.ask(question) self.assertIn('15天左右提取',text) self.assertIn('人工根据实际承运条款核实',text) self.assertIn('3000KG = 30000元',self.ask('一共多少钱')) def test_claim_processing_time_is_not_treated_as_shipping_time(self): self.assertFalse(timing_with_claim('理赔需要多久')) self.assertFalse(timing_with_claim('赔偿处理时效是几天')) self.assertTrue(timing_with_claim('美森多久到，如果晚到会赔吗')) def test_missing_liability_terms_are_not_invented(self): text=self.ask('美森正班海派多久能到，延误会赔吗？') self.assertIn('15天左右提取',text) self.assertIn('实际承运条款核实',text) self.assertNotIn('无时效赔偿',text) def test_partial_liability_terms_are_not_applied_to_every_origin(self): with self.app.app_context(): connection=db.connect() connection.execute("UPDATE quote_rates SET eta=eta||? WHERE service=? AND origin=?", ('\n无时效赔偿','美森正班海派','义乌')) connection.commit() connection.close() text=self.ask('美森正班海派多久到，延误会赔偿吗？') self.assertIn('部分匹配方案注明',text) self.assertNotIn('美森正班海派的渠道资料注明',text) def test_fresh_named_us_service_uses_local_timing_without_a_country_prompt(self): text=self.ask('美森正班海派多久能到') self.assertIn('15天左右提取',text) self.assertNotIn('欧洲',text) self.assertNotIn('请提供目的国家',text) self.assertNotIn('KG =',text) def test_existing_bad_assistant_replies_do_not_prevent_context_recovery(self): self.ask('一票3000KG发美国，义乌美东海派报价') with self.app.app_context(): for question,reply in ( ('请帮我推荐一下，怎么走比较好','请说明您想确认的渠道和费用。'), ('美森正班海派多久能到？如果晚到会赔偿吗？','请提供目的国家。'), ('美森正班海派多久能到','知识库只有欧洲海运信息。'), ): db.save_message(self.cid,'user',question) db.save_message(self.cid,'assistant',reply) text=self.ask('那美森正班海派呢') self.assertIn('15天左右提取',text) self.assertIn('3000KG = 30000元',text) self.assertNotIn('欧洲',text) def test_other_country_context_is_not_silently_replaced_by_a_service_name(self): with self.app.app_context(): db.save_message(self.cid,'user','新的一票100KG发德国海运报价') db.save_message(self.cid,'assistant','请核对对应目的地报价。') text=self.ask('美森正班海派多久能到') self.assertIn('确认这票是否发往美国',text) self.assertNotIn('15天',text) self.assertNotIn('元/KG',text) def test_known_us_warehouse_can_replace_another_destination_without_its_old_weight(self): with self.app.app_context(): db.save_message(self.cid,'user','新的一票100KG发德国海运报价') db.save_message(self.cid,'assistant','请核对对应目的地报价。') text=self.ask('ZZZ1报价') self.assertIn('美国',text) self.assertNotIn('100KG',text) self.assertNotIn('确认这票是否发往美国',text) def test_cheapest_is_compared_by_price_instead_of_slowest_timing(self): with self.app.app_context(): connection=db.connect() connection.execute("UPDATE quote_rates SET price='3' WHERE service='合德海派'") connection.commit() connection.close() self.ask('美国3000KG义乌美东海派报价') text=self.ask('最快和最便宜相差多少钱') cheap=text.split('最便宜参考方案')[1] self.assertIn('合德海派',cheap) self.assertIn('最快与最便宜方案可计算运费差额为21000元',text) self.assertNotIn('最慢参考方案',text) def test_english_fastest_and_cheapest_have_both_headings(self): text=self.ask('3000kg to USA by sea, fastest and cheapest prices please') self.assertIn('Fastest reference options',text) self.assertIn('Cheapest reference options',text)