243 lines
12 KiB
Python
243 lines
12 KiB
Python
"""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)
|