"""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'') archive.writestr('xl/_rels/workbook.xml.rels','') archive.writestr('xl/worksheets/sheet1.xml',f'01234[1]Sheet!A1999列16384') self.assertEqual(read_sheets(path),{'合成资料':{2:{'A':'01234'}}}) if __name__ == '__main__': unittest.main()