222 lines
12 KiB
Python
222 lines
12 KiB
Python
"""Synthetic prices exercise the quote contract; no customer workbook is committed."""
|
||
|
||
from copy import deepcopy
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
import json
|
||
import sqlite3
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import patch
|
||
from zipfile import ZipFile
|
||
|
||
from logistics_chat import create_app, db
|
||
from logistics_chat.quote_data import import_snapshot, load_book
|
||
from logistics_chat.quote_engine import quantity
|
||
|
||
|
||
def fixture_snapshot():
|
||
rates = []
|
||
for origin, extra in (('义乌',0), ('华南',1)):
|
||
for zone, price in (('美西',2), ('美中',3), ('美东',4)):
|
||
for minimum, discount in ((12,0),(51,1),(101,1.5)):
|
||
rates.append({'service':'普船海派','kind':'parcel','origin':origin,'zone':zone,
|
||
'unit':'kg','tax':'included','min_weight':str(minimum), 'price':str(price+extra-discount),
|
||
'eta':'船开后若干天','destinations':[],'source_sheet':'合成海派','source_cell':'C9'})
|
||
for unit, tax, price in (('kg','included','3'),('cbm','excluded','100')):
|
||
rates.append({'service':'合成海卡','kind':'truck','origin':origin,'zone':'','unit':unit,
|
||
'tax':tax,'min_weight':'0','price':price,'eta':'船开后若干天','destinations':['ZZZ1'],
|
||
'source_sheet':'合成海卡','source_cell':'C8'})
|
||
return {'schema':1,'source_name':'synthetic-quotes.xlsx','source_sha256':'1'*64,'rates':rates,
|
||
'warehouses':[{'code':'ZZZ1','zone':zone,'state':'AA','postal':'01234','address':'',
|
||
'source_sheet':'合成分区','source_cell':f'A{index+3}'} for index,zone in enumerate(('东1区','东2区'))],
|
||
'postcodes':[{'postal':'01234','severity':'remote','fee':'1'},
|
||
{'postal':'99901','severity':'extreme','fee':None},
|
||
{'postal':'12345','severity':'extreme','fee':None}],
|
||
'documents':[{'sheet':'合成说明','cell':'A1','text':'合成业务说明，不含真实数据'}]}
|
||
|
||
|
||
def frames(response):
|
||
return [json.loads(f[6:]) for f in response.get_data(as_text=True).split('\n\n') if f.startswith('data: ')]
|
||
|
||
|
||
class QuoteTests(unittest.TestCase):
|
||
def setUp(self):
|
||
directory = tempfile.TemporaryDirectory(); self.addCleanup(directory.cleanup)
|
||
root = Path(directory.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':'1'*64})
|
||
with self.app.app_context():
|
||
connection = db.connect()
|
||
import_snapshot(connection,fixture_snapshot(),activate=True)
|
||
connection.commit(); connection.close()
|
||
|
||
def ask(self, question, cid=None):
|
||
events = frames(self.client.post('/stream_chat',json={'msg':question,'conv_id':cid}))
|
||
self.assertEqual(events[-1],{'d':1},events)
|
||
self.assertFalse(any(e.get('e') for e in events),events)
|
||
return events[0]['t']
|
||
|
||
def conversation(self):
|
||
return self.client.post('/api/new_chat').json['id']
|
||
|
||
def test_guest_us_30kg_always_gets_prices_without_model_or_login(self):
|
||
with patch('logistics_chat.ai_client.post_completion') as model:
|
||
for question in ('发送美国30KG','30kg发US仓海运报价','30kg发USFBA仓海运报价','30kg to US'):
|
||
with self.subTest(question=question):
|
||
text = self.ask(question)
|
||
self.assertIn('美西 2元/KG，30KG = 60元',text)
|
||
self.assertIn('美东 4元/KG，30KG = 120元',text)
|
||
self.assertIn('华南',text)
|
||
model.assert_not_called()
|
||
|
||
def test_ambiguous_warehouse_returns_both_zones_and_own_rates(self):
|
||
text = self.ask('美国30kg发ZZZ1海派，义乌交仓')
|
||
for zone in ('东1区','东2区'):
|
||
self.assertIn('ZZZ1 · '+zone,text)
|
||
self.assertEqual(text.count('30KG = 150元'),2)
|
||
self.assertIn('海派已计入偏远附加费 1元/KG',text)
|
||
self.assertNotIn('华南',text)
|
||
|
||
def test_context_resumes_after_warehouse_origin_weight_and_dimensions(self):
|
||
cid = self.conversation()
|
||
self.ask('美国30kg报价',cid)
|
||
text = self.ask('ZZZ1',cid)
|
||
self.assertIn('美国 30KG',text)
|
||
self.assertIn('合成海卡',text)
|
||
text = self.ask('华南呢',cid)
|
||
self.assertIn('30KG',text)
|
||
self.assertIn('华南',text)
|
||
self.assertNotIn('· 义乌',text)
|
||
text = self.ask('50kg呢',cid)
|
||
self.assertIn('美国 50KG',text)
|
||
text = self.ask('2箱每箱15kg，40x30x20cm呢',cid)
|
||
self.assertIn('美国 30KG',text)
|
||
fresh = self.ask('ZZZ1')
|
||
self.assertNotIn('美国 30KG',fresh)
|
||
self.assertIn('多少公斤',fresh)
|
||
|
||
def test_new_topics_do_not_reuse_old_quote_or_cross_order_privacy(self):
|
||
from logistics_chat.quote_service import answer_quote
|
||
cid = self.conversation()
|
||
self.ask('美国30kg报价',cid)
|
||
with self.app.test_request_context('/'):
|
||
self.assertIsNone(answer_quote('新的一票50kg发德国报价',cid))
|
||
db.save_message(cid,'user','美国货物丢失如何理赔？')
|
||
self.assertIsNone(answer_quote('50kg呢',cid))
|
||
events = frames(self.client.post('/stream_chat',json={'msg':'查询入仓码 INB20260001','conv_id':cid}))
|
||
self.assertTrue(events[0]['login_required'])
|
||
|
||
def test_replaced_destination_stays_replaced_after_other_followups(self):
|
||
cid = self.conversation()
|
||
self.ask('美国30kg发ZZZ1海派，义乌',cid)
|
||
text = self.ask('换成邮编12345',cid)
|
||
self.assertNotIn('ZZZ1',text)
|
||
self.ask('那空运呢',cid)
|
||
text = self.ask('海运呢',cid)
|
||
self.assertNotIn('ZZZ1',text)
|
||
self.assertIn('12345',text)
|
||
self.assertIn('美国 30KG',text)
|
||
|
||
def test_new_quote_tables_start_empty_and_existing_orders_are_untouched(self):
|
||
with self.app.app_context():
|
||
connection = db.connect()
|
||
before = connection.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0]
|
||
import_snapshot(connection,fixture_snapshot(),activate=True)
|
||
self.assertEqual(connection.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0],before)
|
||
self.assertEqual(connection.execute("SELECT postal FROM quote_postcodes WHERE severity='remote'").fetchone()[0],'01234')
|
||
connection.execute('UPDATE quote_books SET active=0'); connection.commit(); connection.close()
|
||
text = self.ask('美国30kg海运报价')
|
||
self.assertIn('尚未包含',text)
|
||
|
||
def test_tier_boundaries_and_per_box_minimum_use_decimal_calculation(self):
|
||
for weight, amount in ((50,100),(51,51),(100,100),(101,50.5)):
|
||
with self.subTest(weight=weight):
|
||
self.assertIn(f'{amount:g}元',self.ask(f'美国美西{weight}kg海派，义乌交仓'))
|
||
text = self.ask('美国美西2箱每箱5kg，义乌海派')
|
||
self.assertIn('美国 10KG',text)
|
||
self.assertIn('24KG = 48元',text)
|
||
with self.app.app_context():
|
||
book = load_book()
|
||
rate = next(r for r in book['rates'] if r['kind']=='parcel')
|
||
intent = {'weight':'10','cbm':None,'dims':[60,50,40],'dim_unit':'cm','boxes':2}
|
||
self.assertEqual(quantity(intent,rate),Decimal(40))
|
||
|
||
def test_non_tax_volume_uses_density_rule_and_missing_volume_still_replies(self):
|
||
text = self.ask('美国30kg发ZZZ1，不包税，义乌')
|
||
self.assertIn('100元/CBM',text)
|
||
self.assertNotIn('30CBM',text)
|
||
text = self.ask('美国363kg发ZZZ1，不包税，0.5方，义乌')
|
||
self.assertIn('1CBM = 100元',text)
|
||
|
||
def test_extreme_remote_air_missing_and_invalid_inputs_finish_cleanly(self):
|
||
text = self.ask('美国30kg邮编12345海派，义乌')
|
||
self.assertIn('遥远地址',text)
|
||
self.assertIn('30KG = 120元',text)
|
||
self.assertIn('不接收',self.ask('美国30kg邮编99901海派'))
|
||
self.assertIn('空运费率当前资料没有列出',self.ask('美国30kg空运报价'))
|
||
self.assertIn('大于 0',self.ask('美国0kg海运报价'))
|
||
|
||
def test_quote_history_is_saved_and_isolated_by_visitor(self):
|
||
cid = self.conversation()
|
||
text = self.ask('美国30KG报价',cid)
|
||
history = self.client.get('/api/get_messages',query_string={'conv_id':cid}).json
|
||
self.assertTrue(any(m['content']==text for m in history))
|
||
other = self.app.test_client(); other.post('/api/visitor/session',json={'key':'2'*64})
|
||
self.assertEqual(other.get('/api/get_messages',query_string={'conv_id':cid}).json,[])
|
||
|
||
def test_import_activation_is_fresh_idempotent_and_invalid_import_preserves_current(self):
|
||
with self.app.app_context():
|
||
connection = db.connect()
|
||
result = import_snapshot(connection,fixture_snapshot(),activate=True)
|
||
self.assertTrue(result['existing'])
|
||
self.assertEqual(connection.execute('SELECT COUNT(*) FROM quote_rates').fetchone()[0],22)
|
||
newer = deepcopy(fixture_snapshot()); newer['source_sha256']='2'*64
|
||
for rate in newer['rates']:
|
||
rate['price']='9'
|
||
import_snapshot(connection,newer,activate=True); connection.commit()
|
||
invalid = deepcopy(newer); invalid['source_sha256']='3'*64; invalid['rates'][0]['price']='NaN'
|
||
with self.assertRaises(ValueError):
|
||
import_snapshot(connection,invalid,activate=True)
|
||
self.assertEqual(connection.execute('SELECT COUNT(*) FROM quote_books WHERE active=1').fetchone()[0],1)
|
||
self.assertEqual(connection.execute('SELECT source_sha256 FROM quote_books WHERE active=1').fetchone()[0],'2'*64)
|
||
connection.close()
|
||
self.assertIn('美西 9元/KG，30KG = 270元',self.ask('美国30kg海派，义乌'))
|
||
|
||
def test_failed_insert_rolls_back_entire_version_and_import_does_not_commit_caller(self):
|
||
with self.app.app_context():
|
||
connection = db.connect()
|
||
broken = deepcopy(fixture_snapshot()); broken['source_sha256']='2'*64
|
||
broken['postcodes'].append(deepcopy(broken['postcodes'][0]))
|
||
with self.assertRaises(sqlite3.IntegrityError):
|
||
import_snapshot(connection,broken,activate=True)
|
||
self.assertEqual(connection.execute('SELECT COUNT(*) FROM quote_books').fetchone()[0],1)
|
||
connection.rollback()
|
||
newer = deepcopy(fixture_snapshot()); newer['source_sha256']='3'*64
|
||
import_snapshot(connection,newer,activate=True)
|
||
connection.rollback()
|
||
self.assertEqual(connection.execute('SELECT source_sha256 FROM quote_books WHERE active=1').fetchone()[0],'1'*64)
|
||
connection.close()
|
||
|
||
def test_lowercase_english_pronoun_is_not_a_us_destination(self):
|
||
from logistics_chat.quote_service import answer_quote
|
||
with self.app.app_context():
|
||
self.assertIsNone(answer_quote('Can you help us quote 30kg to Germany?',None))
|
||
|
||
def test_local_xlsx_reader_ignores_external_formulas_and_fake_full_width_headers(self):
|
||
from logistics_chat.quote_workbook import read_sheets
|
||
with tempfile.TemporaryDirectory() as folder:
|
||
path = Path(folder)/'fixture.xlsx'
|
||
ns = 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'
|
||
with ZipFile(path,'w') as archive:
|
||
archive.writestr('xl/workbook.xml',f'<workbook xmlns="{ns}" xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets><sheet name="合成资料" sheetId="1" r:id="rId1"/></sheets></workbook>')
|
||
archive.writestr('xl/_rels/workbook.xml.rels','<Relationships><Relationship Id="rId1" Target="worksheets/sheet1.xml"/></Relationships>')
|
||
archive.writestr('xl/worksheets/sheet1.xml',f'<worksheet xmlns="{ns}"><sheetData><row r="2"><c r="A2" t="inlineStr"><is><t>01234</t></is></c><c r="B2"><f>[1]Sheet!A1</f><v>999</v></c><c r="XFD2" t="inlineStr"><is><t>列16384</t></is></c></row></sheetData></worksheet>')
|
||
self.assertEqual(read_sheets(path),{'合成资料':{2:{'A':'01234'}}})
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|