233 lines
11 KiB
Python
233 lines
11 KiB
Python
"""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_natural_cost_preference_selects_the_cheapest_after_recommendation(self):
|
|
self.ask('美国3000KG义乌美东海派报价')
|
|
self.ask('请帮我推荐一下')
|
|
text=self.ask('那我想省钱')
|
|
self.assertIn('3000KG = 18000元',text)
|
|
self.assertNotIn('美森正班海派',text)
|
|
|
|
def test_short_choice_confirmation_preserves_the_last_selected_quote(self):
|
|
self.ask('美国3000KG义乌美东海派报价')
|
|
self.ask('请帮我推荐一下')
|
|
self.ask('时效优先')
|
|
self.ask('就选这个')
|
|
self.assertIn('15天左右提取',self.ask('多久能提取'))
|
|
self.assertIn('3000KG = 30000元',self.ask('那一共多少钱'))
|
|
|
|
def test_quantity_changes_continue_after_a_short_confirmation(self):
|
|
self.ask('美国3000KG义乌美东美森正班海派报价')
|
|
self.ask('再加500KG')
|
|
self.ask('就按这个')
|
|
self.assertIn('2500KG = 25000元',self.ask('再减一吨'))
|
|
|
|
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('最快和最便宜相差多少钱')
|
|
self.assertIn('最快与最便宜方案可计算运费差额为21000元',text)
|
|
self.assertNotIn('最慢参考方案',text)
|
|
detail=self.ask('展开报价明细')
|
|
self.assertIn('合德海派',detail.split('最便宜参考方案')[1])
|
|
|
|
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)
|