122 lines
6.3 KiB
Python
122 lines
6.3 KiB
Python
"""Fast/slow comparisons preserve shipping conditions and comparable rate groups."""
|
||
|
||
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_engine import select_timed_rates
|
||
from test_quotes import fixture_snapshot, frames
|
||
|
||
|
||
def timing_snapshot():
|
||
book=fixture_snapshot()
|
||
rates=[]
|
||
for origin in ('义乌','华南'):
|
||
for zone,offset in (('美西',0),('美中',1),('美东',2)):
|
||
for service,price,eta in (('美森正班海派',8,'船开后15天左右'),
|
||
('合德海派',6,'船开后20-22天左右'),
|
||
('普船海派',4,'船开后25-30天左右')):
|
||
rates.append({'service':service,'kind':'parcel','origin':origin,'zone':zone,
|
||
'unit':'kg','tax':'included','min_weight':'12','price':str(price+offset),
|
||
'eta':eta,'destinations':[],'source_sheet':'合成时效','source_cell':'A1'})
|
||
book['rates']=rates
|
||
return book
|
||
|
||
|
||
class QuoteTimingTests(unittest.TestCase):
|
||
def setUp(self):
|
||
folder=tempfile.TemporaryDirectory();self.addCleanup(folder.cleanup)
|
||
root=Path(folder.name)
|
||
self.app=create_app({'TESTING':True,'DATABASE':root/'test.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':'2'*64})
|
||
with self.app.app_context():
|
||
connection=db.connect()
|
||
import_snapshot(connection,timing_snapshot(),activate=True)
|
||
connection.commit();connection.close()
|
||
|
||
def ask(self,question,cid=None):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
events=frames(self.client.post('/stream_chat',json={'msg':question,'conv_id':cid}))
|
||
model.assert_not_called()
|
||
self.assertEqual(events[-1],{'d':1},events)
|
||
return events[0]['t']
|
||
|
||
def test_fastest_and_slowest_show_each_region_origin_and_the_correct_amount(self):
|
||
text=self.ask('我想发一批货，一吨发发到美国走海运，最快的和最慢的价格分别是多少？')
|
||
fast,slow=text.split('最慢参考方案')
|
||
self.assertIn('最快参考方案',fast)
|
||
self.assertIn('美西 8元/KG，1000KG = 8000元',fast)
|
||
self.assertIn('美东 10元/KG，1000KG = 10000元',fast)
|
||
self.assertIn('美森正班海派 · 华南',fast)
|
||
self.assertNotIn('普船海派',fast)
|
||
self.assertIn('美西 4元/KG，1000KG = 4000元',slow)
|
||
self.assertIn('美东 6元/KG，1000KG = 6000元',slow)
|
||
self.assertIn('船开后25-30天左右提取',slow)
|
||
self.assertNotIn('合德海派',text)
|
||
|
||
def test_short_followups_keep_weight_origin_zone_and_selected_comparison(self):
|
||
cid=self.client.post('/api/new_chat').json['id']
|
||
self.ask('1000KG到美国走海运，最快和最慢价格分别多少',cid)
|
||
for question in ('慢的呢？','义乌交仓','美东呢','请给我具体报价'):
|
||
text=self.ask(question,cid)
|
||
self.assertIn('美国 1000KG',text)
|
||
self.assertIn('最慢参考方案',text)
|
||
self.assertNotIn('最快参考方案',text)
|
||
self.assertNotIn('华南',text)
|
||
self.assertIn('1000KG = 6000元',text)
|
||
text=self.ask('50KG呢',cid)
|
||
self.assertIn('50KG = 300元',text)
|
||
self.assertNotIn('1000KG',text)
|
||
text=self.ask('全部渠道报价',cid)
|
||
self.assertIn('美森正班海派',text)
|
||
self.assertIn('合德海派',text)
|
||
self.assertNotIn('最慢参考方案',text)
|
||
text=self.ask('新的一批100KG发美国海运报价',cid)
|
||
self.assertIn('美国 100KG',text)
|
||
self.assertIn('华南',text)
|
||
self.assertNotIn('最慢参考方案',text)
|
||
|
||
def test_numeric_timing_groups_keep_destinations_tax_and_service_kinds_separate(self):
|
||
base={'zone':'美西','kind':'parcel','unit':'kg','tax':'included'}
|
||
rates=[{**base,'eta':'10-35天','service':'wide'},
|
||
{**base,'eta':'15-25天','service':'narrow'},
|
||
{**base,'zone':'美东','eta':'40-45天','service':'east'},
|
||
{**base,'tax':'excluded','eta':'20天','service':'excluded'},
|
||
{**base,'kind':'truck','eta':'50天','service':'truck'},
|
||
{**base,'eta':'未提供时效','service':'unknown'}]
|
||
self.assertEqual({r['service'] for r in select_timed_rates(rates)},
|
||
{'wide','east','excluded','truck'})
|
||
self.assertEqual({r['service'] for r in select_timed_rates(rates,slowest=True)},
|
||
{'wide','east','excluded','truck'})
|
||
self.assertEqual(select_timed_rates([{**base,'eta':'若干天'}]),[])
|
||
|
||
def test_english_comparison_keeps_numbers_and_renders_both_speeds(self):
|
||
text=self.ask('1000kg to USA by sea, fastest and slowest prices please')
|
||
self.assertIn('Fastest reference options',text)
|
||
self.assertIn('Slowest reference options',text)
|
||
self.assertIn('1000KG = 4000 CNY',text)
|
||
|
||
def test_different_regional_times_are_rendered_on_their_own_price_lines(self):
|
||
with self.app.app_context():
|
||
connection=db.connect()
|
||
connection.execute("UPDATE quote_rates SET eta='船开后18天左右' WHERE service='美森正班海派' AND zone='美东'")
|
||
connection.commit();connection.close()
|
||
text=self.ask('美国1000KG海运最快价格，义乌')
|
||
self.assertIn('美东 10元/KG，1000KG = 10000元（船开后18天左右提取）',text)
|
||
self.assertIn('美西 8元/KG，1000KG = 8000元',text)
|
||
|
||
def test_unknown_times_do_not_become_a_fastest_or_slowest_promise(self):
|
||
with self.app.app_context():
|
||
connection=db.connect()
|
||
connection.execute("UPDATE quote_rates SET eta='若干天'")
|
||
connection.commit();connection.close()
|
||
text=self.ask('美国1000KG海运最快和最慢价格')
|
||
self.assertIn('最快参考方案暂时无法确定',text)
|
||
self.assertIn('最慢参考方案暂时无法确定',text)
|
||
self.assertNotIn('1000KG =',text)
|