287 lines
15 KiB
Python
287 lines
15 KiB
Python
"""Structured quotes preserve exact quantities and transaction-owned turn state."""
|
|
|
|
from hashlib import sha256
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from flask import g
|
|
|
|
from logistics_chat import create_app, db
|
|
from logistics_chat.conversation_state import commit_state, init_state_schema, load_state
|
|
from logistics_chat.quote_data import import_snapshot, load_book
|
|
from logistics_chat.quote_service import answer_quote
|
|
from logistics_chat.quote_tool import QuoteTurn
|
|
from test_quote_timing import timing_snapshot
|
|
|
|
|
|
class QuoteToolTests(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': '', 'CHAT_API_KEY': '', 'RATE_LIMIT_PER_MIN': 1000})
|
|
client = self.app.test_client()
|
|
client.post('/api/visitor/session', json={'key': '7' * 64})
|
|
self.cid = client.post('/api/new_chat').json['id']
|
|
self.context = self.app.test_request_context()
|
|
self.context.push()
|
|
self.addCleanup(self.context.pop)
|
|
g.visitor_id = sha256(('7' * 64).encode()).hexdigest()
|
|
connection = db.connect()
|
|
init_state_schema(connection)
|
|
import_snapshot(connection, timing_snapshot(), activate=True)
|
|
connection.commit()
|
|
connection.close()
|
|
self.book = load_book()
|
|
|
|
def arguments(self, **extra):
|
|
return {'question': '核对报价', 'request': 'quote', 'selection': 'current',
|
|
'services': [], 'policy_topic': 'auto', **extra}
|
|
|
|
def start(self, **shipment):
|
|
turn = QuoteTurn(self.cid, '一票新货', self.book)
|
|
answer = turn.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '1000',
|
|
'zone': '美西', 'origin': '义乌', **shipment}))
|
|
self.assertEqual(answer['status'], 'ok', answer)
|
|
self.persist(turn)
|
|
return answer
|
|
|
|
def persist(self, turn, question='已确认货物条件'):
|
|
connection = db.connect()
|
|
try:
|
|
connection.execute("INSERT INTO messages(conv_id,role,content,ctime) VALUES(?,'user',?,'2026-11-06 12:00:00')", (self.cid, question))
|
|
self.assertTrue(commit_state(connection, self.cid, turn.candidate, turn.expected_version))
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_same_turn_addition_applies_once_across_comparisons(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '再加100KG，最快最慢都查', self.book)
|
|
change = [{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]
|
|
for selection in ('fastest', 'slowest', 'fastest'):
|
|
answer = turn.run(self.arguments(changes=change, selection=selection))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['total_weight_kg'], '1100')
|
|
self.assertIn('1100KG', answer['t'])
|
|
self.persist(turn, '再加100KG')
|
|
later = QuoteTurn(self.cid, '再加100KG', self.book)
|
|
answer = later.run(self.arguments(changes=change))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['total_weight_kg'], '1200')
|
|
|
|
def test_conflicting_same_turn_edits_do_not_commit_partial_change(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '修改重量', self.book)
|
|
for delta in ('100', '200'):
|
|
answer = turn.run(self.arguments(changes=[{'field': 'total_weight_kg', 'operation': 'add', 'value': delta}]))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
self.assertIsNone(turn.candidate)
|
|
self.assertEqual(load_state(self.cid)['intent']['weight'], '1000')
|
|
|
|
def test_add_without_base_and_bad_dimensions_are_clarifications(self):
|
|
turn = QuoteTurn(self.cid, '加100', self.book)
|
|
answer = turn.run(self.arguments(changes=[{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
self.assertIsNone(turn.candidate)
|
|
for dims in ([20, 0, 30], [20, 30], ['NaN', 20, 30]):
|
|
answer = turn.run(self.arguments(shipment={'country': 'US', 'dimensions_cm': dims}))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
|
|
def test_structured_dimensions_use_decimal_without_reparsing_question(self):
|
|
turn = QuoteTurn(self.cid, '箱规已经提供', self.book)
|
|
with patch('logistics_chat.quote_service.intent_with_context', side_effect=AssertionError('must not parse')):
|
|
answer = turn.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '10.5',
|
|
'boxes': 1, 'dimensions_cm': ['60', '40', '30'], 'zone': '美西', 'origin': '义乌'}))
|
|
self.assertEqual(answer['status'], 'ok', answer)
|
|
self.assertIn('12KG', answer['t'])
|
|
self.assertEqual(Decimal(answer['data']['options'][0]['quantity']), Decimal('12'))
|
|
|
|
def test_carton_validation_and_weight_edit_invalidate_derived_fields(self):
|
|
turn = QuoteTurn(self.cid, '两箱', self.book)
|
|
invalid = {'country': 'US', 'total_weight_kg': '50', 'boxes': 2,
|
|
'packages': [{'count': 2, 'weight_kg': '20'}]}
|
|
self.assertEqual(turn.run(self.arguments(shipment=invalid))['status'], 'needs_clarification')
|
|
invalid['total_weight_kg'] = '40'
|
|
invalid.update(material='metal', metal_scope='partial', metal_chargeable_weight_kg='10')
|
|
self.assertEqual(turn.run(self.arguments(shipment=invalid))['status'], 'ok')
|
|
self.persist(turn)
|
|
update = QuoteTurn(self.cid, '总重改成100', self.book)
|
|
answer = update.run(self.arguments(changes=[{'field': 'total_weight_kg', 'operation': 'set', 'value': '100'}]))
|
|
self.assertNotIn('packages', answer['data']['confirmed_inputs'])
|
|
self.assertNotIn('metal_chargeable_weight_kg', answer['data']['confirmed_inputs'])
|
|
|
|
def test_country_switch_clears_old_quantities_and_destination(self):
|
|
self.start(warehouse='ZZZ1')
|
|
turn = QuoteTurn(self.cid, '这票改发英国', self.book)
|
|
answer = turn.run(self.arguments(shipment={'country': 'GB'}))
|
|
self.assertEqual(answer['status'], 'unsupported')
|
|
self.assertEqual(answer['data']['confirmed_inputs'], {'country': 'GB'})
|
|
self.assertEqual(turn.candidate['options'], [])
|
|
|
|
def test_fixed_book_survives_active_version_switch(self):
|
|
turn = QuoteTurn(self.cid, '1000KG美国', self.book)
|
|
newer = timing_snapshot()
|
|
newer['source_sha256'] = '2' * 64
|
|
for rate in newer['rates']:
|
|
rate['price'] = '99'
|
|
connection = db.connect()
|
|
import_snapshot(connection, newer, activate=True)
|
|
connection.commit()
|
|
connection.close()
|
|
with patch('logistics_chat.quote_service.load_book', side_effect=AssertionError('must pin book')):
|
|
answer = turn.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '1000', 'zone': '美西', 'origin': '义乌'}))
|
|
self.assertEqual(answer['data_version'], self.book['id'])
|
|
self.assertIn('8000元', answer['t'])
|
|
self.assertNotIn('99000元', answer['t'])
|
|
|
|
def test_stale_deleted_and_unowned_state_cannot_be_committed(self):
|
|
first, second = (QuoteTurn(self.cid, '1000KG', self.book) for _ in range(2))
|
|
for turn in (first, second):
|
|
turn.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '1000'}))
|
|
self.persist(first)
|
|
connection = db.connect()
|
|
self.assertFalse(commit_state(connection, self.cid, second.candidate, second.expected_version))
|
|
connection.close()
|
|
g.visitor_id = 'other'
|
|
forbidden = QuoteTurn(self.cid, '查报价', self.book)
|
|
self.assertEqual(forbidden.run(self.arguments())['status'], 'error')
|
|
self.assertIsNone(load_state(self.cid))
|
|
g.visitor_id = sha256(('7' * 64).encode()).hexdigest()
|
|
latest = QuoteTurn(self.cid, '1000KG', self.book)
|
|
latest.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '1000'}))
|
|
connection = db.connect()
|
|
connection.execute('DELETE FROM chat_history WHERE id=?', (self.cid,))
|
|
self.assertFalse(commit_state(connection, self.cid, latest.candidate, latest.expected_version))
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
def test_state_write_rolls_back_with_the_reply_transaction(self):
|
|
turn = QuoteTurn(self.cid, '1000KG', self.book)
|
|
turn.run(self.arguments(shipment={'country': 'US', 'total_weight_kg': '1000'}))
|
|
connection = db.connect()
|
|
connection.execute('BEGIN IMMEDIATE')
|
|
self.assertTrue(commit_state(connection, self.cid, turn.candidate, turn.expected_version))
|
|
connection.rollback()
|
|
connection.close()
|
|
self.assertEqual(load_state(self.cid)['version'], 0)
|
|
|
|
def test_legacy_fallback_updates_saved_conditions_without_double_add(self):
|
|
self.start()
|
|
answer = answer_quote('再加100KG', self.cid)
|
|
self.assertIn('1100KG', answer['t'])
|
|
self.assertEqual(answer['_quote_state_candidate']['intent']['weight'], '1100')
|
|
connection = db.connect()
|
|
connection.execute("INSERT INTO messages(conv_id,role,content,ctime) VALUES(?,'user',?,'2026-11-06 12:01:00')", (self.cid, '再加100KG'))
|
|
self.assertTrue(commit_state(connection, self.cid, answer['_quote_state_candidate'], answer['_quote_state_expected_version']))
|
|
connection.commit()
|
|
connection.close()
|
|
turn = QuoteTurn(self.cid, '最快呢', self.book)
|
|
answer = turn.run(self.arguments(shipment={}, selection='fastest'))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['total_weight_kg'], '1100')
|
|
self.assertIn('1100KG', answer['t'])
|
|
|
|
def test_old_numbered_plan_is_not_guessed_from_reply_text(self):
|
|
self.start()
|
|
answer = answer_quote('方案2多少钱', self.cid)
|
|
self.assertIn('不能根据旧回复猜测方案编号', answer['t'])
|
|
|
|
def test_option_reference_rejects_stale_price_version(self):
|
|
initial = self.start()
|
|
option = initial['data']['options'][0]['option_id']
|
|
newer = {**self.book, 'id': self.book['id'] + 100}
|
|
turn = QuoteTurn(self.cid, '按这个方案', newer)
|
|
answer = turn.run(self.arguments(option_id=option))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
self.assertIn('已失效', answer['t'])
|
|
|
|
def test_changes_only_can_migrate_legacy_history_once(self):
|
|
db.save_message(self.cid, 'user', '1000KG到美国，美西，义乌交仓')
|
|
turn = QuoteTurn(self.cid, '再加100KG', self.book)
|
|
answer = turn.run(self.arguments(changes=[{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['total_weight_kg'], '1100')
|
|
|
|
def test_later_dimensions_in_same_turn_invalidate_earlier_option_facts(self):
|
|
self.start(total_weight_kg='10')
|
|
turn = QuoteTurn(self.cid, '补充箱规', self.book)
|
|
first = turn.run(self.arguments(shipment={}, selection='fastest'))
|
|
old_ids = {item['option_id'] for item in first['data']['options']}
|
|
answer = turn.run(self.arguments(shipment={'boxes': 1, 'total_weight_kg': '10', 'dimensions_cm': [100, 100, 100]}))
|
|
self.assertEqual(answer['status'], 'ok')
|
|
self.assertTrue(old_ids.isdisjoint({item['option_id'] for item in turn.candidate['options']}))
|
|
|
|
def test_duplicate_change_in_one_call_is_not_applied_twice(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '加100KG', self.book)
|
|
change = {'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}
|
|
answer = turn.run(self.arguments(changes=[change, change]))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['total_weight_kg'], '1100')
|
|
|
|
def test_other_customer_or_deleted_conversation_cannot_reuse_turn_snapshot(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '最快呢', self.book)
|
|
g.visitor_id = 'other'
|
|
answer = turn.run(self.arguments(shipment={}))
|
|
self.assertEqual(answer['status'], 'error')
|
|
self.assertEqual(answer['data'], {})
|
|
|
|
def test_all_channels_does_not_inherit_a_previously_selected_service(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '查普船', self.book)
|
|
answer = turn.run(self.arguments(shipment={}, services=['普船']))
|
|
self.assertNotIn('美森正班海派', answer['t'])
|
|
self.persist(turn)
|
|
later = QuoteTurn(self.cid, '全部渠道', self.book)
|
|
answer = later.run(self.arguments(shipment={}, selection='all'))
|
|
self.assertIn('美森正班海派', answer['t'])
|
|
|
|
def test_new_shipment_is_reset_once_and_keeps_both_comparison_results(self):
|
|
self.start(total_weight_kg='2000')
|
|
turn = QuoteTurn(self.cid, '新一票100KG，最快最慢', self.book)
|
|
payload = {'country': 'US', 'total_weight_kg': '100', 'zone': '美西', 'origin': '义乌'}
|
|
fast = turn.run(self.arguments(shipment=payload, new_shipment=True, selection='fastest'))
|
|
slow = turn.run(self.arguments(shipment=payload, new_shipment=True, selection='slowest'))
|
|
expected = {item['option_id'] for answer in (fast, slow) for item in answer['data']['options']}
|
|
self.assertEqual({item['option_id'] for item in turn.candidate['options']}, expected)
|
|
self.assertEqual(turn.candidate['intent']['weight'], '100')
|
|
|
|
def test_policy_followup_keeps_existing_option_references(self):
|
|
initial = self.start()
|
|
turn = QuoteTurn(self.cid, '怎么算计费重', self.book)
|
|
turn.run(self.arguments(shipment={}, request='policy', policy_topic='weight_basis'))
|
|
self.assertEqual(turn.candidate['options'], initial['data']['options'])
|
|
|
|
def test_legacy_tool_question_does_not_discard_missing_unit_error(self):
|
|
self.start()
|
|
turn = QuoteTurn(self.cid, '重量改成100', self.book)
|
|
answer = turn.run(self.arguments(question='重量改成100'))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
self.assertIn('单位', answer['t'])
|
|
self.assertIsNone(turn.candidate)
|
|
|
|
def test_non_fba_address_fee_warning_survives_structured_followup(self):
|
|
initial = self.start(postal='90001', address_type='non_fba')
|
|
self.assertIn('住宅/商业派送附加费', initial['t'])
|
|
turn = QuoteTurn(self.cid, '改100KG', self.book)
|
|
answer = turn.run(self.arguments(changes=[{'field': 'total_weight_kg', 'operation': 'set', 'value': '100'}]))
|
|
self.assertEqual(answer['data']['confirmed_inputs']['address_type'], 'non_fba')
|
|
self.assertIn('住宅/商业派送附加费', answer['t'])
|
|
|
|
def test_legacy_country_switch_invalidates_old_model_context(self):
|
|
self.start()
|
|
self.assertIsNone(answer_quote('100KG发英国', self.cid))
|
|
db.save_message(self.cid, 'user', '100KG发英国')
|
|
turn = QuoteTurn(self.cid, '最快呢', self.book)
|
|
self.assertEqual(turn.context()['confirmed_inputs'], {})
|
|
self.assertEqual(turn.context()['options'], [])
|
|
answer = turn.run(self.arguments(shipment={}, selection='fastest'))
|
|
self.assertEqual(answer['status'], 'needs_clarification')
|
|
self.assertNotIn('1000KG', answer['t'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|