477 lines
28 KiB
Python
477 lines
28 KiB
Python
"""HTTP and SSE contracts across the application factory and extracted services."""
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import requests
|
|
|
|
from logistics_chat import create_app, db
|
|
|
|
|
|
class ModelResponse:
|
|
def __init__(self, text='物流测试回答，费用与时效请以报价表为准。', status=200):
|
|
self.content = text
|
|
self.status_code = status
|
|
self.closed = False
|
|
self.text = 'model error'
|
|
|
|
def iter_lines(self):
|
|
yield ('data: ' + json.dumps({'choices': [{'delta': {'content': self.content}}]})).encode()
|
|
yield b'data: [DONE]'
|
|
|
|
def json(self):
|
|
return {'choices': [{'message': {'content': self.content}}], 'error': {'message': 'model error'}}
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def frames(response):
|
|
return [json.loads(part[6:]) for part in response.get_data(as_text=True).split('\n\n') if part.startswith('data: ')]
|
|
|
|
|
|
class ApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.root = Path(self.directory.name)
|
|
self.app = self.application(self.root)
|
|
self.client = self.app.test_client()
|
|
from logistics_chat.order_data import import_records
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
import_records(connection, [{'tenant_id':'fixture', 'order_no':'15252125',
|
|
'status':'PROCESSING', 'country_code':'US', 'gross_weight_kg':'30',
|
|
'history_json':[{'time':'2026-10-28 10:00', 'status':'SUBMITTED'}]}])
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
def application(self, root):
|
|
return create_app({'TESTING': True, 'DATABASE': root/'history.db',
|
|
'UPLOAD_DIR': root/'uploads', 'BACKUP_DIR': root/'backups',
|
|
'LOG_DIR': root/'logs', 'DEEPSEEK_API_KEY': ''})
|
|
|
|
def conversation(self):
|
|
return self.client.post('/api/new_chat').json['id']
|
|
|
|
def test_pages_static_files_and_cache_headers(self):
|
|
for path in ('/', '/admin', '/admin/dashboard'):
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('no-store', response.headers['Cache-Control'])
|
|
with self.client.get('/static/fa.jpg') as image:
|
|
self.assertEqual(image.status_code, 200)
|
|
|
|
def test_page_styles_and_scripts_are_accessible(self):
|
|
class Assets(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.urls = []
|
|
def handle_starttag(self, tag, attrs):
|
|
attrs = dict(attrs)
|
|
if tag == 'script' and attrs.get('src'):
|
|
self.urls.append(attrs['src'])
|
|
if tag == 'link' and attrs.get('rel') == 'stylesheet':
|
|
self.urls.append(attrs['href'])
|
|
for path in ('/', '/admin', '/admin/dashboard'):
|
|
parser = Assets()
|
|
parser.feed(self.client.get(path).get_data(as_text=True))
|
|
self.assertGreaterEqual(len(parser.urls), 3)
|
|
for url in parser.urls:
|
|
resource = self.client.get(url)
|
|
self.assertEqual(resource.status_code, 200, url)
|
|
self.assertGreater(len(resource.data), 0)
|
|
resource.close()
|
|
|
|
def test_history_card_roundtrip_and_deletion(self):
|
|
cid = self.conversation()
|
|
response = self.client.post('/stream_chat', json={'msg': '15252125', 'conv_id': cid})
|
|
events = frames(response)
|
|
self.assertEqual(events[-1], {'d': 1})
|
|
self.assertEqual(events[0]['card']['no'], '15252125')
|
|
messages = self.client.get(f'/api/get_messages?conv_id={cid}').json
|
|
self.assertEqual(len(messages), 3)
|
|
self.assertTrue(messages[-1]['content'].startswith('__CARD__'))
|
|
self.assertEqual(self.client.get('/api/get_history').json[0]['title'], '15252125')
|
|
self.assertEqual(self.client.delete(f'/api/del_history?id={cid}').json, {'ok': True})
|
|
self.assertEqual(self.client.get(f'/api/get_messages?conv_id={cid}').json, [])
|
|
|
|
def test_empty_image_and_missing_key_streams(self):
|
|
self.assertIn('e', frames(self.client.post('/stream_chat', json={'msg': ''}))[0])
|
|
image_reply = frames(self.client.post('/stream_chat', json={'msg': '[图片]/uploads/test.png'}))
|
|
self.assertIn('收到您的图片', image_reply[0]['t'])
|
|
self.assertEqual(image_reply[-1], {'d': 1})
|
|
self.assertIn('API Key', frames(self.client.get('/stream_chat?msg=hello'))[0]['e'])
|
|
|
|
def test_attachment_filename_is_not_used_as_a_tracking_number(self):
|
|
attachment = '[文件]/uploads/20261029100145_6787.txt'
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '请查看附件 ' + attachment}))
|
|
self.assertFalse(any('card' in event for event in events))
|
|
self.assertIn('API Key', events[0]['e'])
|
|
cid = self.conversation()
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '查询15252125 ' + attachment, 'conv_id': cid}))
|
|
self.assertEqual(events[0]['card']['no'], '15252125')
|
|
self.assertEqual(self.client.get(f'/api/get_messages?conv_id={cid}').json[1]['content'], '查询15252125 ' + attachment)
|
|
|
|
def test_image_attachments_do_not_swallow_order_questions_and_are_saved(self):
|
|
cid = self.conversation()
|
|
image = '[图片]/uploads/20261029100145_6787.png'
|
|
events = frames(self.client.post('/stream_chat', json={'msg':'查询15252125 '+image, 'conv_id':cid}))
|
|
self.assertEqual(events[0]['card']['no'], '15252125')
|
|
events = frames(self.client.post('/stream_chat', json={'msg':image, 'conv_id':cid}))
|
|
self.assertIn('收到您的图片', events[0]['t'])
|
|
messages = self.client.get(f'/api/get_messages?conv_id={cid}').json
|
|
self.assertEqual(messages[-2]['content'], image)
|
|
self.assertIn('收到您的图片', messages[-1]['content'])
|
|
|
|
def test_bad_chat_payloads_produce_readable_errors_without_calling_model(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
invalid = ([1], 'hello', {'msg':7}, {'msg':['test']}, {'msg':'物流咨询', 'conv_id':{}},
|
|
{'msg':'物流咨询', 'conv_id':-1}, {'msg':'物流咨询', 'conv_id':1.5}, {'msg':'物流咨询','conv_id':False},
|
|
{'msg':'物流咨询','conv_id':999999999999999999999999},
|
|
{'msg':'物流咨询','conv_id':'²'}, {'msg':'物流咨询','conv_id':'9'*5000})
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
for index, payload in enumerate(invalid):
|
|
with self.subTest(case=index):
|
|
events = frames(self.client.post('/stream_chat', json=payload))
|
|
self.assertIn('e', events[0])
|
|
model.assert_not_called()
|
|
|
|
def test_deleted_conversations_close_human_sessions_and_reject_new_chat_messages(self):
|
|
cid = self.conversation()
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
from datetime import datetime
|
|
clock.now.return_value = datetime(2026,10,29,12)
|
|
sid = self.client.post('/api/human/request',json={'conv_id':cid}).json['session_id']
|
|
self.client.post('/api/human/send',json={'sid':sid,'content':'原会话内容'})
|
|
self.client.delete(f'/api/del_history?id={cid}')
|
|
self.assertEqual(self.client.get('/api/human/queue').json, [])
|
|
self.assertEqual(self.client.get(f'/api/human/poll?sid={sid}').json['messages'], [])
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat',json={'msg':'15252125','conv_id':cid}))
|
|
self.assertIn('已删除',events[0]['e'])
|
|
model.assert_not_called()
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
clock.now.return_value = datetime(2026,10,29,12)
|
|
self.assertEqual(self.client.post('/api/human/request',json={'conv_id':cid}).status_code,404)
|
|
|
|
def test_uk_quotes_and_followups_keep_the_uk_rates(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
cid = self.conversation()
|
|
for message, expected in (('30kg英国空运报价','英国空运专线'), ('那海运呢？','英国海运'),
|
|
('那空运呢？','英国空运专线'), ('那50公斤呢？','英国空运专线')):
|
|
with self.subTest(message=message):
|
|
with patch('logistics_chat.ai_client.post_completion',return_value=ModelResponse()) as model, patch('logistics_chat.chat_service.gen_followups',return_value=[]):
|
|
frames(self.client.post('/stream_chat',json={'msg':message,'conv_id':cid}))
|
|
prompt = model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn(expected,prompt)
|
|
self.assertNotIn('欧洲空运专线',prompt)
|
|
self.assertNotIn('欧洲海运快线',prompt)
|
|
|
|
def test_us_quotes_never_reuse_european_price_tables(self):
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat',json={'msg':'30KG发美国FBA仓，海运报价','conv_id':self.conversation()}))
|
|
self.assertIn('目的地',events[0]['t'])
|
|
self.assertIn('人工',events[0]['t'])
|
|
model.assert_not_called()
|
|
|
|
def test_mixed_country_codes_select_the_correct_rates(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
for message, channel, amount in (('30kg发UK海运报价','英国海运','384'),
|
|
('30kg发uk海运报价','英国海运','384'),
|
|
('30kg发GB海运报价','英国海运','384'),
|
|
('30kg发DE仓海运报价','欧洲海运','369')):
|
|
with self.subTest(message=message):
|
|
with patch('logistics_chat.ai_client.post_completion', return_value=ModelResponse()) as model, patch('logistics_chat.chat_service.gen_followups',return_value=[]):
|
|
events=frames(self.client.post('/stream_chat',json={'msg':message}))
|
|
self.assertEqual(events[-1],{'d':1})
|
|
prompt=model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn(channel,prompt)
|
|
self.assertIn(amount,prompt)
|
|
for message in ('30kg发US仓海运报价','30kg发us仓海运报价','30kg发USFBA仓海运报价',
|
|
'30kg发阿根廷海运报价','30kg海运报价'):
|
|
with self.subTest(message=message), patch('logistics_chat.ai_client.post_completion') as model:
|
|
events=frames(self.client.post('/stream_chat',json={'msg':message}))
|
|
self.assertIn('目的',events[0]['t'])
|
|
model.assert_not_called()
|
|
|
|
def test_short_channel_followups_and_retries_keep_weight_and_region(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
cid=self.conversation()
|
|
sequence=(('30kg英国海运报价','英国海运','384'), ('海运呢','英国海运','384'),
|
|
('空运呢','英国空运','1,440'), ('50公斤呢','英国空运','2,400'))
|
|
for message, channel, amount in sequence:
|
|
with self.subTest(message=message):
|
|
with patch('logistics_chat.ai_client.post_completion',return_value=ModelResponse()) as model, patch('logistics_chat.chat_service.gen_followups',return_value=[]):
|
|
frames(self.client.post('/stream_chat',json={'msg':message,'conv_id':cid}))
|
|
prompt=model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn(channel,prompt)
|
|
self.assertIn(amount,prompt)
|
|
with patch('logistics_chat.ai_client.post_completion',side_effect=requests.exceptions.ConnectionError):
|
|
self.assertIn('e',frames(self.client.post('/stream_chat',json={'msg':'海运呢','conv_id':cid}))[0])
|
|
with patch('logistics_chat.ai_client.post_completion',return_value=ModelResponse()) as model:
|
|
frames(self.client.post('/stream_chat',json={'msg':'海运呢','conv_id':cid}))
|
|
prompt=model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn('英国海运',prompt)
|
|
self.assertIn('640',prompt)
|
|
|
|
def test_complete_new_questions_do_not_inherit_an_old_quote(self):
|
|
from logistics_chat.chat_service import shipping_intent_with_context
|
|
cid=self.conversation()
|
|
with self.app.app_context():
|
|
db.save_message(cid,'user','30kg英国海运报价')
|
|
db.save_message(cid,'assistant','报价结果')
|
|
for message in ('新的一票50kg德国空运报价','空运理赔应该怎么办？','海运时效一般多久？'):
|
|
with self.subTest(message=message):
|
|
intent=shipping_intent_with_context(message,cid)
|
|
self.assertNotEqual(intent['region'],'uk')
|
|
db.save_message(cid,'user','空运理赔应该怎么办？')
|
|
self.assertIsNone(shipping_intent_with_context('海运呢',cid)['weight'])
|
|
|
|
def test_dimension_quotes_and_changes_do_not_become_order_identifiers(self):
|
|
self.app.config['DEEPSEEK_API_KEY']='test-model-key'
|
|
cid=self.conversation()
|
|
for message, amount in (('30kg英国海运，100x100x100cm','1,817'), ('50x40x30cm呢','384'),
|
|
('1.2x0.8x0.6m呢','1,229'),
|
|
('那另外一票50kg德国海运','615')):
|
|
with self.subTest(message=message):
|
|
with patch('logistics_chat.ai_client.post_completion',return_value=ModelResponse()) as model, patch('logistics_chat.chat_service.gen_followups',return_value=[]):
|
|
events=frames(self.client.post('/stream_chat',json={'msg':message,'conv_id':cid}))
|
|
self.assertEqual(events[-1],{'d':1})
|
|
self.assertTrue(model.called)
|
|
prompt=model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn(amount,prompt)
|
|
if '另外一票' in message:
|
|
self.assertNotIn('体积重',prompt)
|
|
|
|
def test_destination_only_replies_continue_the_existing_quote(self):
|
|
self.app.config['DEEPSEEK_API_KEY']='test-model-key'
|
|
cid=self.conversation()
|
|
self.assertIn('目的',frames(self.client.post('/stream_chat',json={'msg':'30kg海运报价','conv_id':cid}))[0]['t'])
|
|
with patch('logistics_chat.ai_client.post_completion',return_value=ModelResponse()) as model, patch('logistics_chat.chat_service.gen_followups',return_value=[]):
|
|
frames(self.client.post('/stream_chat',json={'msg':'英国','conv_id':cid}))
|
|
prompt=model.call_args.args[0]['messages'][0]['content']
|
|
self.assertIn('英国海运',prompt)
|
|
self.assertIn('384',prompt)
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events=frames(self.client.post('/stream_chat',json={'msg':'US仓','conv_id':cid}))
|
|
self.assertIn('尚未包含',events[0]['t'])
|
|
model.assert_not_called()
|
|
|
|
def test_uk_volume_quotes_do_not_reuse_continental_warehouse_prices(self):
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat',json={'msg':'2方发英国海运多少钱'}))
|
|
self.assertIn('未提供',events[0]['t'])
|
|
self.assertIn('按方费率',events[0]['t'])
|
|
model.assert_not_called()
|
|
|
|
def test_closed_human_sessions_cannot_be_reopened_by_a_late_accept(self):
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
from datetime import datetime
|
|
clock.now.return_value = datetime(2026,10,29,12)
|
|
sid = self.client.post('/api/human/request',json={'conv_id':self.conversation()}).json['session_id']
|
|
self.client.post('/api/human/close',json={'sid':sid})
|
|
self.assertEqual(self.client.post('/api/human/accept',json={'sid':sid}).status_code,409)
|
|
self.assertEqual(self.client.get(f'/api/human/poll?sid={sid}').json['status'],'closed')
|
|
|
|
def test_human_lifecycle_typing_and_read_receipts(self):
|
|
self.assertEqual(self.client.post('/api/human/request', json={'conv_id': 0}).status_code, 400)
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
from datetime import datetime
|
|
clock.now.return_value = datetime(2026, 10, 29, 12)
|
|
sid = self.client.post('/api/human/request', json={'conv_id': self.conversation()}).json['session_id']
|
|
self.assertEqual(self.client.get('/api/human/queue').json[0]['id'], sid)
|
|
self.client.post('/api/human/accept', json={'sid': sid})
|
|
self.client.post('/api/human/send', json={'sid': sid, 'side': 'customer', 'content': 'hello'})
|
|
self.client.post('/api/human/typing', json={'sid': sid})
|
|
state = self.client.get(f'/api/human/poll?sid={sid}&after=0').json
|
|
self.assertEqual(state['status'], 'active')
|
|
self.assertTrue(state['typing'])
|
|
self.assertFalse(state['messages'][0]['read'])
|
|
self.client.post('/api/human/read', json={'sid': sid})
|
|
self.client.post('/api/human/stop_typing', json={'sid': sid})
|
|
state = self.client.get(f'/api/human/poll?sid={sid}').json
|
|
self.assertTrue(state['messages'][0]['read'])
|
|
self.assertFalse(state['typing'])
|
|
self.client.post('/api/human/close', json={'sid': sid})
|
|
self.assertEqual(self.client.get('/api/human/active').json, [])
|
|
self.assertEqual(self.client.get(f'/api/human/poll?sid={sid}').json['status'], 'closed')
|
|
|
|
def test_application_state_is_not_shared_between_factories(self):
|
|
self.app.extensions['human_typing']['1'] = 123
|
|
self.app.extensions['rate_hits']['127.0.0.1'].append(123)
|
|
other = self.application(self.root/'other')
|
|
self.assertEqual(other.extensions['human_typing'], {})
|
|
self.assertEqual(dict(other.extensions['rate_hits']), {})
|
|
|
|
def test_factories_with_different_log_directories_do_not_cross_write(self):
|
|
applications = []
|
|
try:
|
|
for name in ('first', 'second'):
|
|
root = self.root/name
|
|
applications.append(create_app({'DATABASE': root/'history.db', 'UPLOAD_DIR': root/'uploads',
|
|
'BACKUP_DIR': root/'backups', 'LOG_DIR': root/'logs', 'DEEPSEEK_API_KEY': ''}))
|
|
applications[0].logger.warning('first application only')
|
|
for app in applications:
|
|
for handler in app.logger.handlers:
|
|
handler.flush()
|
|
self.assertIn('first application only', (self.root/'first/logs/app.log').read_text())
|
|
self.assertNotIn('first application only', (self.root/'second/logs/app.log').read_text())
|
|
finally:
|
|
for app in applications:
|
|
for handler in list(app.logger.handlers):
|
|
if isinstance(handler, logging.FileHandler):
|
|
handler.close()
|
|
app.logger.removeHandler(handler)
|
|
|
|
def test_rate_limit_and_exemptions(self):
|
|
self.app.config['RATE_LIMIT_PER_MIN'] = 2
|
|
self.assertEqual(self.client.get('/api/get_history').status_code, 200)
|
|
self.assertEqual(self.client.get('/api/get_history').status_code, 200)
|
|
self.assertEqual(self.client.get('/api/get_history').status_code, 429)
|
|
self.assertEqual(self.client.get('/api/human/queue').status_code, 200)
|
|
with self.client.get('/static/fa.jpg') as image:
|
|
self.assertEqual(image.status_code, 200)
|
|
|
|
def test_attachment_roundtrip_and_rejection(self):
|
|
response = self.client.post('/api/upload', data={'file': (io.BytesIO(b'hello'), 'hello.txt')})
|
|
self.assertEqual(response.status_code, 200)
|
|
downloaded = self.client.get(response.json['url'])
|
|
self.assertEqual(downloaded.data, b'hello')
|
|
downloaded.close()
|
|
self.assertEqual(self.client.post('/api/upload', data={'file': (io.BytesIO(b'x'), 'x.exe')}).status_code, 400)
|
|
|
|
def test_health_check_remains_available_when_business_requests_are_rate_limited(self):
|
|
self.app.config['RATE_LIMIT_PER_MIN'] = 1
|
|
self.assertEqual(self.client.get('/').status_code, 200)
|
|
self.assertEqual(self.client.get('/api/get_history').status_code, 429)
|
|
response = self.client.get('/healthz')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json, {'status': 'ok'})
|
|
|
|
def test_feedback_stats_and_csv(self):
|
|
cid = self.conversation()
|
|
with self.app.app_context():
|
|
db.save_message(cid, 'user', 'hello, "quoted"\nworld')
|
|
self.client.post('/api/feedback', json={'vote': 1, 'question': 'hello', 'answer': 'world'})
|
|
self.assertEqual(self.client.get('/api/admin/stats').json['good'], 1)
|
|
response = self.client.get('/api/export_chat?format=csv')
|
|
self.assertEqual(response.mimetype, 'text/csv')
|
|
self.assertIn('chat_export.csv', response.headers['Content-Disposition'])
|
|
self.assertIn('"hello, ""quoted""\nworld"', response.get_data(as_text=True))
|
|
self.client.delete('/api/clear_all')
|
|
self.assertEqual(self.client.get('/api/get_history').json, [])
|
|
|
|
def test_business_routes_with_initialized_tables_and_legacy_auth(self):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
tables = {row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
self.assertTrue({'agents', 'orders', 'tickets'}.issubset(tables))
|
|
connection.executescript(
|
|
"INSERT INTO agents VALUES('demo', 'test-password', 'Agent');"
|
|
"INSERT INTO orders VALUES(1, '123', '15252125', 'active', '2026-10-28');")
|
|
connection.commit()
|
|
connection.close()
|
|
self.assertEqual(self.client.post('/admin/login', json={'username': 'demo', 'password': 'wrong'}).status_code, 401)
|
|
self.assertEqual(self.client.post('/admin/login', json={'username': 'demo', 'password': 'test-password'}).json['name'], 'Agent')
|
|
self.assertEqual(self.client.get('/api/my_orders?phone=123').json['orders'][0]['track_no'], '15252125')
|
|
self.assertEqual(self.client.post('/api/ticket', json={'content': 'help', 'phone': '123'}).status_code, 200)
|
|
self.assertEqual(self.client.get('/api/tickets').json[0]['content'], 'help')
|
|
|
|
def test_stream_compression_followups_and_deadlines(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
streamed = ModelResponse('这是用于覆盖长回答压缩的说明。'*100)
|
|
responses = [streamed, ModelResponse('基础运费2050元，时效38-45天。'), ModelResponse('["费用呢","时效呢","能带电吗"]')]
|
|
with patch('logistics_chat.ai_client.requests.post', side_effect=responses) as post:
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '200kg 德国海运', 'conv_id': self.conversation()}))
|
|
self.assertTrue(any('t' in event for event in events))
|
|
self.assertTrue(any('replace' in event for event in events))
|
|
self.assertTrue(any('suggest' in event for event in events))
|
|
self.assertEqual(events[-1], {'d': 1})
|
|
self.assertEqual([call.kwargs['timeout'] for call in post.call_args_list], [(5, 90), (5, 30), (5, 15)])
|
|
self.assertTrue(streamed.closed)
|
|
self.assertTrue(all(response.closed for response in responses))
|
|
|
|
def test_incomplete_model_streams_are_reported_and_saved_as_partial(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
for reason in (None, 'length'):
|
|
with self.subTest(reason=reason):
|
|
cid = self.conversation()
|
|
response = ModelResponse('这是已经生成的部分内容，需要保留以便用户了解中断前的信息。')
|
|
def lines():
|
|
yield ('data: '+json.dumps({'choices':[{'delta':{'content':response.content}, 'finish_reason':reason}]})).encode()
|
|
if reason:
|
|
yield b'data: [DONE]'
|
|
response.iter_lines = lines
|
|
with patch('logistics_chat.ai_client.post_completion', return_value=response):
|
|
events = frames(self.client.post('/stream_chat', json={'msg':'物流咨询', 'conv_id':cid}))
|
|
self.assertTrue(any('e' in event for event in events))
|
|
self.assertFalse(any(event.get('d') for event in events))
|
|
self.assertTrue(response.closed)
|
|
messages = self.client.get(f'/api/get_messages?conv_id={cid}').json
|
|
self.assertIn('回答中断', messages[-1]['content'])
|
|
|
|
def test_stream_network_and_model_errors(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
for error, expected in ((requests.exceptions.Timeout, '超时'),
|
|
(requests.exceptions.ConnectionError, '无法连接')):
|
|
with self.subTest(error=error):
|
|
with patch('logistics_chat.ai_client.requests.post', side_effect=error):
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '物流咨询'}))
|
|
self.assertIn(expected, events[0]['e'])
|
|
denied = ModelResponse(status=401)
|
|
with patch('logistics_chat.ai_client.requests.post', return_value=denied):
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '物流咨询'}))
|
|
self.assertEqual(events[0]['e'], '系统繁忙，请稍后再试')
|
|
self.assertTrue(denied.closed)
|
|
empty = ModelResponse(text='')
|
|
with patch('logistics_chat.ai_client.requests.post', return_value=empty):
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '物流咨询'}))
|
|
self.assertIn('没有返回内容', events[0]['e'])
|
|
self.assertTrue(empty.closed)
|
|
|
|
def test_cancellation_saves_partial_answer_and_closes_http_response(self):
|
|
self.app.config['DEEPSEEK_API_KEY'] = 'test-model-key'
|
|
cid = self.conversation()
|
|
model = ModelResponse('这是长度超过二十字的回答内容，用于验证中断保存行为。')
|
|
with patch('logistics_chat.ai_client.requests.post', return_value=model):
|
|
response = self.client.post('/stream_chat', json={'msg': '海运咨询', 'conv_id': cid}, buffered=False)
|
|
next(response.response)
|
|
response.close()
|
|
messages = self.client.get(f'/api/get_messages?conv_id={cid}').json
|
|
self.assertIn('回答中断', messages[-1]['content'])
|
|
self.assertTrue(model.closed)
|
|
|
|
def test_entry_and_package_imports_do_not_create_runtime_files(self):
|
|
root = str(Path(__file__).resolve().parents[1])
|
|
code = 'import sys; sys.path.insert(0, sys.argv[1]); import am; from logistics_chat import create_app'
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
result = subprocess.run([sys.executable, '-c', code, root], cwd=directory,
|
|
capture_output=True, text=True, timeout=15)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(list(Path(directory).iterdir()), [])
|
|
|
|
def test_wsgi_compatibility_creates_one_application_on_explicit_access(self):
|
|
import am
|
|
try:
|
|
with patch.object(am, 'create_app', side_effect=lambda: self.application(self.root/'wsgi')) as factory:
|
|
application = getattr(am, 'app')
|
|
self.assertIs(getattr(am, 'app'), application)
|
|
self.assertEqual(application.test_client().get('/').status_code, 200)
|
|
self.assertEqual(factory.call_count, 1)
|
|
finally:
|
|
am.__dict__.pop('app', None)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|