154 lines
8.8 KiB
Python
154 lines
8.8 KiB
Python
"""Remote dev behavior on the modular backend and the light workspace."""
|
|
|
|
from datetime import datetime
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
from logistics_chat import create_app, db
|
|
|
|
|
|
class DevIntegrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
root = Path(directory.name)
|
|
self.app = create_app({'TESTING': True, 'DATABASE': root/'test.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()
|
|
|
|
def session(self):
|
|
cid = self.client.post('/api/new_chat').json['id']
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
clock.now.return_value = datetime(2026, 10, 31, 8)
|
|
result = self.client.post('/api/human/request', json={'conv_id': cid}).json
|
|
return result['session_id']
|
|
|
|
def test_new_conversation_persists_one_welcome_before_user_message(self):
|
|
data = self.client.post('/api/new_chat').json
|
|
self.assertIn('欢迎', data['welcome'])
|
|
self.client.post('/stream_chat', json={'msg': '15252125', 'conv_id': data['id']}, buffered=True)
|
|
messages = self.client.get('/api/get_messages?conv_id='+str(data['id'])).json
|
|
self.assertEqual([message['role'] for message in messages], ['assistant', 'user', 'assistant'])
|
|
self.assertEqual(messages[0]['content'], data['welcome'])
|
|
|
|
def test_daily_service_hours_and_offline_ticket_on_fresh_database(self):
|
|
cid = self.client.post('/api/new_chat').json['id']
|
|
for hour, open_now in ((7, False), (8, True), (21, True), (22, False)):
|
|
with self.subTest(hour=hour), patch('logistics_chat.human_service.datetime') as clock:
|
|
clock.now.return_value = datetime(2026, 10, 31, hour)
|
|
result = self.client.post('/api/human/request', json={'conv_id': cid}).json
|
|
self.assertEqual(result['ok'], open_now)
|
|
if not open_now:
|
|
self.assertTrue(result['closed'])
|
|
self.assertEqual(self.client.post('/api/ticket', json={'content': '周末留言'}).json, {'ok': True})
|
|
self.assertEqual(self.client.get('/api/tickets').json[0]['content'], '周末留言')
|
|
|
|
def test_transfer_returns_to_queue_preserves_messages_and_note(self):
|
|
sid = self.session()
|
|
self.client.post('/api/human/accept', json={'sid': sid})
|
|
self.client.post('/api/human/note', json={'sid': sid, 'note': '请跟进派送'})
|
|
self.client.post('/api/human/send', json={'sid': sid, 'side': 'agent', 'content': '正在处理'})
|
|
self.client.post('/api/human/typing', json={'sid': sid})
|
|
self.assertEqual(self.client.post('/api/human/transfer', json={'sid': sid}).json, {'ok': True})
|
|
self.assertEqual(self.client.get('/api/human/active').json, [])
|
|
self.assertEqual(self.client.get('/api/human/queue').json[0]['id'], sid)
|
|
state = self.client.get(f'/api/human/poll?sid={sid}').json
|
|
self.assertEqual(state['note'], '请跟进派送')
|
|
self.assertFalse(state['typing'])
|
|
self.assertEqual(state['messages'][0]['content'], '正在处理')
|
|
self.assertEqual(state['messages'][-1]['side'], 'system')
|
|
self.assertEqual(self.client.post('/api/human/transfer', json={'sid': sid}).status_code, 409)
|
|
self.client.post('/api/human/accept', json={'sid': sid})
|
|
self.assertEqual(self.client.get(f'/api/human/poll?sid={sid}').json['status'], 'active')
|
|
|
|
def test_human_send_rejects_closed_missing_or_mismatched_conversations(self):
|
|
sid = self.session()
|
|
payload = {'sid': sid, 'side': 'customer', 'content': '不能串到其他会话'}
|
|
other_cid = self.client.post('/api/new_chat').json['id']
|
|
self.assertEqual(self.client.post('/api/human/send', json={**payload, 'conv_id': other_cid}).status_code, 409)
|
|
self.assertEqual(self.client.post('/api/human/send', json={**payload, 'sid': 99999}).status_code, 404)
|
|
self.assertEqual(self.client.post('/api/human/send', json={**payload, 'side': 'system'}).status_code, 400)
|
|
self.client.post('/api/human/close', json={'sid': sid})
|
|
self.assertEqual(self.client.post('/api/human/send', json=payload).status_code, 409)
|
|
self.assertEqual(self.client.get(f'/api/human/poll?sid={sid}').json['messages'], [])
|
|
|
|
def test_recall_targets_older_agent_message_and_is_idempotent(self):
|
|
sid = self.session()
|
|
for side, text in (('agent', '早先的回复'), ('customer', '客户消息'), ('agent', '最新回复')):
|
|
self.client.post('/api/human/send', json={'sid': sid, 'side': side, 'content': text})
|
|
before = self.client.get(f'/api/human/poll?sid={sid}').json['messages']
|
|
mid = before[0]['id']
|
|
for _ in range(2):
|
|
self.assertEqual(self.client.post('/api/human/recall', json={'mid': mid}).json, {'ok': True})
|
|
self.assertEqual(self.client.post('/api/human/recall', json={'mid': before[1]['id']}).status_code, 400)
|
|
self.assertEqual(self.client.post('/api/human/recall', json={'mid': 99999}).status_code, 404)
|
|
updates = self.client.get(f'/api/human/poll?sid={sid}&after={before[-1]["id"]}').json['messages']
|
|
self.assertEqual(len(updates), 1)
|
|
self.assertEqual(updates[0]['recalled_id'], mid)
|
|
all_messages = self.client.get(f'/api/human/poll?sid={sid}').json['messages']
|
|
self.assertEqual(all_messages[0]['content'], '该消息已撤回')
|
|
self.assertEqual(all_messages[2]['content'], '最新回复')
|
|
|
|
def test_ratings_validate_stars_and_feed_admin_stats(self):
|
|
sid = self.session()
|
|
for stars in (0, 6, True, '5'):
|
|
self.assertEqual(self.client.post('/api/rate', json={'sid': sid, 'stars': stars}).status_code, 400)
|
|
self.assertEqual(self.client.post('/api/rate', json={'sid': 999, 'stars': 5}).status_code, 404)
|
|
for stars in (4, 5):
|
|
self.assertEqual(self.client.post('/api/rate', json={'sid': sid, 'stars': stars}).json, {'ok': True})
|
|
stats = self.client.get('/api/admin/stats').json
|
|
self.assertEqual((stats['rating_avg'], stats['rating_count']), (4.5, 2))
|
|
|
|
def test_excel_export_search_and_legacy_agent_schema(self):
|
|
cid = self.client.post('/api/new_chat').json['id']
|
|
with self.app.app_context():
|
|
db.save_message(cid, 'user', '=SUM(1,2)\n查询专线')
|
|
connection = db.connect()
|
|
connection.executescript('DROP TABLE agents; CREATE TABLE agents '
|
|
'(id INTEGER PRIMARY KEY, username TEXT, password TEXT, name TEXT);'
|
|
"INSERT INTO agents VALUES(7, 'demo', 'fixture-password', '测试客服');")
|
|
connection.commit()
|
|
connection.close()
|
|
db.init_db()
|
|
results = self.client.get('/api/search_chat?q=专线').json
|
|
self.assertEqual(len(results), 1)
|
|
self.assertIn('SUM', results[0]['content'])
|
|
self.assertEqual(self.client.get('/api/search_chat').json, {'results': []})
|
|
self.assertEqual(self.client.get('/api/agents/list').json, [{'id': 7, 'username': 'demo', 'name': '测试客服'}])
|
|
exported = self.client.get('/api/export_chat')
|
|
self.assertIn('chat_export.xlsx', exported.headers['Content-Disposition'])
|
|
workbook = load_workbook(io.BytesIO(exported.data))
|
|
sheet = workbook.active
|
|
self.assertEqual(sheet['C3'].value, '=SUM(1,2)\n查询专线')
|
|
self.assertEqual(sheet['C3'].data_type, 's')
|
|
self.assertEqual(sheet['B3'].value, '客户')
|
|
workbook.close()
|
|
|
|
def test_existing_schema_is_upgraded_without_losing_business_data(self):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
connection.executescript('DROP TABLE human_sessions; DROP TABLE human_messages;'
|
|
'CREATE TABLE human_sessions(id INTEGER PRIMARY KEY, conv_id INTEGER, status TEXT, ctime TEXT);'
|
|
'CREATE TABLE human_messages(id INTEGER PRIMARY KEY, session_id INTEGER, side TEXT, content TEXT, ctime TEXT);'
|
|
"INSERT INTO human_sessions VALUES(1, 1, 'active', 'now');"
|
|
"INSERT INTO human_messages VALUES(1, 1, 'agent', '保留的回复', 'now');"
|
|
"INSERT INTO agents VALUES('fixture-agent', 'fixture-password', 'Fixture Agent');")
|
|
connection.commit()
|
|
connection.close()
|
|
db.init_db()
|
|
db.init_db()
|
|
state = self.client.get('/api/human/poll?sid=1').json
|
|
self.assertEqual(state['messages'][0]['content'], '保留的回复')
|
|
self.assertEqual(state['note'], '')
|
|
self.assertEqual(self.client.get('/api/agents/list').json[0]['username'], 'fixture-agent')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|