123 lines
5.8 KiB
Python
123 lines
5.8 KiB
Python
"""SQLite migration and attachment boundaries in isolated application contexts."""
|
|
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
from flask import Flask
|
|
from werkzeug.datastructures import FileStorage
|
|
|
|
from logistics_chat import db, file_service
|
|
from logistics_chat.config import default_config
|
|
|
|
|
|
class StorageTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.root = Path(self.directory.name)
|
|
self.app = Flask(__name__)
|
|
self.app.config.update(default_config())
|
|
self.app.config.update(DATABASE=self.root/'history.db', UPLOAD_DIR=self.root/'uploads',
|
|
BACKUP_DIR=self.root/'backups', LOG_DIR=self.root/'logs')
|
|
self.app.teardown_appcontext(db.close_connections)
|
|
self.context = self.app.app_context()
|
|
self.context.push()
|
|
self.addCleanup(self.context.pop)
|
|
db.init_db()
|
|
|
|
def test_old_schema_is_migrated_and_legacy_title_backfilled(self):
|
|
connection = db.connect()
|
|
connection.executescript('DROP TABLE chat_history; CREATE TABLE chat_history '
|
|
'(id INTEGER PRIMARY KEY, title TEXT NOT NULL, ctime TEXT NOT NULL);')
|
|
connection.execute('INSERT INTO chat_history VALUES(1, ?, ?)', ('旧会话', '2026-10-20'))
|
|
connection.commit()
|
|
connection.close()
|
|
db.init_db()
|
|
connection = db.connect()
|
|
self.assertEqual(connection.execute('SELECT utime FROM chat_history').fetchone()[0], '2026-10-20')
|
|
self.assertEqual(connection.execute('SELECT role, content FROM messages').fetchone(), ('user', '旧会话'))
|
|
connection.close()
|
|
|
|
def test_business_tables_start_empty_and_reinitialization_preserves_records(self):
|
|
connection = db.connect()
|
|
for table in ('agents', 'orders', 'tickets'):
|
|
self.assertEqual(connection.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0], 0)
|
|
connection.executescript(
|
|
"INSERT INTO agents VALUES('fixture-agent', 'fixture-password', 'Fixture Agent');"
|
|
"INSERT INTO orders(phone, track_no, status, created) VALUES('123', 'fixture-order', 'active', 'now');"
|
|
"INSERT INTO tickets(content, phone, ctime, status) VALUES('fixture-ticket', '123', 'now', 'pending');"
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
db.init_db()
|
|
connection = db.connect()
|
|
self.assertEqual(connection.execute('SELECT username, password, name FROM agents').fetchall(),
|
|
[('fixture-agent', 'fixture-password', 'Fixture Agent')])
|
|
self.assertEqual(connection.execute('SELECT track_no FROM orders').fetchall(), [('fixture-order',)])
|
|
self.assertEqual(connection.execute('SELECT content FROM tickets').fetchall(), [('fixture-ticket',)])
|
|
connection.close()
|
|
|
|
def test_user_retry_is_idempotent_and_card_context_is_readable(self):
|
|
connection = db.connect()
|
|
connection.execute('INSERT INTO chat_history(title, ctime, utime) VALUES(?,?,?)', ('新对话', 'now', 'now'))
|
|
connection.commit()
|
|
connection.close()
|
|
db.save_message(1, 'user', '海运多少钱')
|
|
self.assertEqual(db.save_message(1, 'user', '海运多少钱'), 'duplicate')
|
|
db.save_message(1, 'assistant', '__CARD__{"no":"15252125","status":"派送中"}')
|
|
context = db.load_context(1, '然后呢')
|
|
self.assertEqual(len(context), 3)
|
|
self.assertIn('派送中', context[1]['content'])
|
|
self.assertNotIn('__CARD__', context[1]['content'])
|
|
self.assertEqual(db.last_user_question(1), '海运多少钱')
|
|
|
|
def test_backup_and_retention_only_touch_configured_directories(self):
|
|
backups = self.root/'backups'
|
|
backups.mkdir()
|
|
(backups/'history_20000101.db').write_bytes(b'old')
|
|
db.backup_db()
|
|
self.assertFalse((backups/'history_20000101.db').exists())
|
|
self.assertEqual(len(list(backups.glob('history_*.db'))), 1)
|
|
uploads = self.root/'uploads'
|
|
uploads.mkdir()
|
|
old = uploads/'old.txt'
|
|
old.write_text('old')
|
|
(uploads/'new.txt').write_text('new')
|
|
os.utime(old, (time.time()-8*86400, time.time()-8*86400))
|
|
file_service.cleanup_uploads()
|
|
self.assertFalse(old.exists())
|
|
self.assertEqual((uploads/'new.txt').read_text(), 'new')
|
|
|
|
def test_upload_uses_generated_name_and_configured_directory(self):
|
|
upload = FileStorage(stream=io.BytesIO(b'hello'), filename='../report.txt')
|
|
result = file_service.upload_file({'file': upload})
|
|
name = result['url'].rsplit('/', 1)[1]
|
|
self.assertEqual((self.root/'uploads'/name).read_bytes(), b'hello')
|
|
self.assertNotIn('..', name)
|
|
self.assertFalse(result['is_image'])
|
|
|
|
def test_invalid_and_oversized_uploads_are_rejected(self):
|
|
self.assertEqual(file_service.upload_file({})[1], 400)
|
|
denied = FileStorage(stream=io.BytesIO(b'bad'), filename='test.exe')
|
|
self.assertEqual(file_service.upload_file({'file': denied})[1], 400)
|
|
self.app.config['MAX_UPLOAD_MB'] = 1
|
|
oversized = FileStorage(stream=io.BytesIO(b'x'*(1024*1024+1)), filename='test.txt')
|
|
self.assertEqual(file_service.upload_file({'file': oversized})[1], 400)
|
|
self.assertFalse((self.root/'uploads').exists())
|
|
|
|
def test_teardown_rolls_back_uncommitted_work(self):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
connection.execute('INSERT INTO chat_history(title, ctime) VALUES(?,?)', ('uncommitted', 'now'))
|
|
connection = db.connect()
|
|
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_history').fetchone()[0], 0)
|
|
connection.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|