mirror of
https://github.com/KingHWJ/WeReadBoard.git
synced 2026-11-03 18:24:58 +08:00
67 lines
3.4 KiB
Python
67 lines
3.4 KiB
Python
"""显式设置 WEREAD_TEST_MYSQL=1 后，验证已迁移数据库；仅清理本测试生成的 ID。"""
|
|
|
|
import os
|
|
import unittest
|
|
from uuid import uuid4
|
|
from unittest.mock import patch
|
|
|
|
import pymysql
|
|
from flask import Flask
|
|
import reading_api
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('WEREAD_TEST_MYSQL') == '1', '需显式启用 MySQL 集成测试')
|
|
class GoalsMySQLTest(unittest.TestCase):
|
|
def test_persistence_import_and_atomic_rollback(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
app.register_blueprint(reading_api.reading_bp)
|
|
client = app.test_client()
|
|
ids = [f'test-{uuid4()}' for _ in range(4)]
|
|
goal = {'id': ids[0], 'name': '数据库验收测试', 'start': '2026-09-01',
|
|
'end': '2026-09-30', 'ids': ['missing-book', '123', '123'], 'sources': ['书单']}
|
|
try:
|
|
response = client.post('/api/reading/goals', json=goal)
|
|
self.assertEqual(response.status_code, 201, response.json)
|
|
self.assertEqual(len(response.json['goal']['ids']), 2)
|
|
# 每个请求使用新的连接，检验提交后的持久化，而不是连接内未提交的内容。
|
|
saved = {g['id']: g for g in client.get('/api/reading/goals').json['goals']}
|
|
self.assertEqual(set(saved[ids[0]]['ids']), {'missing-book', '123'})
|
|
changed = dict(goal, name='修改后的测试', ids=['missing-book'])
|
|
self.assertEqual(client.put(f'/api/reading/goals/{ids[0]}', json=changed).status_code, 200)
|
|
imported = dict(goal, id=ids[1])
|
|
for count in (1, 0):
|
|
response = client.post('/api/reading/goals/import', json={'goals': [goal, imported, imported]})
|
|
self.assertEqual(response.json, {'imported': count})
|
|
saved = {g['id']: g for g in client.get('/api/reading/goals').json['goals']}
|
|
self.assertEqual(saved[ids[0]]['name'], '修改后的测试')
|
|
self.assertEqual(saved[ids[0]]['ids'], ['missing-book'])
|
|
|
|
insert = reading_api.insert_goal
|
|
def fail_second(cur, item, skip_existing=False):
|
|
if item['id'] == ids[3]:
|
|
raise pymysql.OperationalError(2013, '模拟批次中断')
|
|
return insert(cur, item, skip_existing)
|
|
|
|
with patch('reading_api.insert_goal', side_effect=fail_second):
|
|
response = client.post('/api/reading/goals/import', json={
|
|
'goals': [dict(goal, id=ids[2]), dict(goal, id=ids[3])]})
|
|
self.assertEqual(response.status_code, 503)
|
|
saved = {g['id'] for g in client.get('/api/reading/goals').json['goals']}
|
|
self.assertNotIn(ids[2], saved)
|
|
self.assertNotIn(ids[3], saved)
|
|
self.assertEqual(client.delete(f'/api/reading/goals/{ids[0]}').status_code, 200)
|
|
with pymysql.connect(**reading_api.MYSQL_CONFIG) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT COUNT(*) AS n FROM reading_goal_books WHERE goal_id=%s', (ids[0],))
|
|
self.assertEqual(cur.fetchone()['n'], 0)
|
|
finally:
|
|
with pymysql.connect(**reading_api.MYSQL_CONFIG) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.executemany('DELETE FROM reading_goals WHERE id=%s', [(goal_id,) for goal_id in ids])
|
|
conn.commit()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|