"""显式设置 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 api_v2


@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(api_v2.v2_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/v2/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/v2/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/v2/goals/{ids[0]}', json=changed).status_code, 200)
            imported = dict(goal, id=ids[1])
            for count in (1, 0):
                response = client.post('/api/v2/goals/import', json={'goals': [goal, imported, imported]})
                self.assertEqual(response.json, {'imported': count})
            saved = {g['id']: g for g in client.get('/api/v2/goals').json['goals']}
            self.assertEqual(saved[ids[0]]['name'], '修改后的测试')
            self.assertEqual(saved[ids[0]]['ids'], ['missing-book'])

            insert = api_v2.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('api_v2.insert_goal', side_effect=fail_second):
                response = client.post('/api/v2/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/v2/goals').json['goals']}
            self.assertNotIn(ids[2], saved)
            self.assertNotIn(ids[3], saved)
            self.assertEqual(client.delete(f'/api/v2/goals/{ids[0]}').status_code, 200)
            with pymysql.connect(**api_v2.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(**api_v2.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()