"""Remote dev behavior on the modular backend and the light workspace.""" from datetime import datetime, timezone 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() self.client.post('/api/visitor/session', json={'key': '1' * 64}) 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_service_hours_use_china_time_even_when_the_host_clock_is_utc(self): cid = self.client.post('/api/new_chat').json['id'] for utc_hour, expected in ((0, True),(13, True),(14, False),(23, False)): instant = datetime(2026,10,31,utc_hour,tzinfo=timezone.utc) class HostUtcClock(datetime): @classmethod def now(cls, tz=None): return instant.astimezone(tz) if tz else instant.replace(tzinfo=None) with self.subTest(utc_hour=utc_hour), patch('logistics_chat.human_service.datetime', HostUtcClock): result = self.client.post('/api/visitor/human/request',json={'conv_id':cid}).json self.assertEqual(result['ok'],expected) if not expected: self.assertTrue(result['closed']) 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()