WeReadBoard/tests/test_reading_goals.py

"""阅读目标接口回归测试；数据库故障通过连接边界模拟，不访问线上库。"""

import unittest
from datetime import date
from unittest.mock import MagicMock, patch

import pymysql
from flask import Flask

from reading_api import reading_bp


class ReadingGoalsTest(unittest.TestCase):
    def setUp(self):
        app = Flask(__name__)
        app.config["TESTING"] = True
        app.register_blueprint(reading_bp)
        self.client = app.test_client()
        self.conn = MagicMock()
        self.cur = self.conn.cursor.return_value.__enter__.return_value
        self.cur.fetchone.return_value = {"id": "goal-1"}
        self.cur.fetchall.return_value = []
        self.cur.rowcount = 1
        self.connect = patch("reading_api.pymysql.connect", return_value=self.conn).start()
        self.addCleanup(patch.stopall)
        self.goal = {
            "id": "goal-1", "name": "  九月阅读  ",
            "start": "2026-09-01", "end": "2026-09-30",
            "ids": ["unknown-book", "unknown-book", "123"],
            "sources": ["科幻书单", "科幻书单"],
        }

    def test_create_normalizes_and_commits_unknown_members(self):
        response = self.client.post("/api/reading/goals", json=self.goal)
        self.assertEqual(response.status_code, 201)
        expected = dict(self.goal, name="九月阅读", ids=["unknown-book", "123"], sources=["科幻书单"])
        self.assertEqual(response.json, {"goal": expected})
        self.conn.commit.assert_called_once()
        self.conn.rollback.assert_not_called()
        self.conn.close.assert_called_once()

    def test_legacy_goal_path_remains_available(self):
        response = self.client.post("/api/" + "v" + "2" + "/goals", json=self.goal)
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json["goal"]["id"], "goal-1")

    def test_rejects_invalid_fields_without_writes(self):
        invalid = [
            {"id": 123}, {"id": ""}, {"id": "x" * 65},
            {"name": "   "}, {"name": "字" * 61},
            {"start": "2026-02-29"}, {"start": "2026-9-01"},
            {"start": "2026-09-31"}, {"end": "2026-08-31"},
            {"ids": []}, {"ids": "123"}, {"ids": [123]},
            {"ids": [""]}, {"ids": ["x" * 65]},
            {"sources": "书单"}, {"sources": [123]},
        ]
        for fields in invalid:
            with self.subTest(fields=fields):
                response = self.client.post("/api/reading/goals", json=dict(self.goal, **fields))
                self.assertEqual(response.status_code, 400)
                self.assertTrue(response.json["error"])
        self.cur.execute.assert_not_called()
        self.conn.commit.assert_not_called()

    def test_invalid_json_and_body_shape_are_chinese_errors(self):
        for body in ("{", "null", "[]"):
            response = self.client.post("/api/reading/goals", data=body, content_type="application/json")
            self.assertEqual(response.status_code, 400)
            self.assertTrue(any("\u4e00" <= ch <= "\u9fff" for ch in response.json["error"]))

    def test_list_preserves_unknown_members_and_formats_dates(self):
        self.cur.fetchall.return_value = [
            {"id": "goal-1", "name": "九月阅读", "start_date": date(2026, 9, 1),
             "end_date": date(2026, 9, 30), "sources": '["科幻书单"]', "book_id": "unknown-book"},
            {"id": "goal-1", "name": "九月阅读", "start_date": date(2026, 9, 1),
             "end_date": date(2026, 9, 30), "sources": '["科幻书单"]', "book_id": "123"},
        ]
        response = self.client.get("/api/reading/goals")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json, {"goals": [dict(
            self.goal, name="九月阅读", ids=["unknown-book", "123"], sources=["科幻书单"]
        )]})

    def test_create_duplicate_returns_conflict(self):
        self.cur.execute.side_effect = pymysql.err.IntegrityError(1062, "Duplicate entry")
        response = self.client.post("/api/reading/goals", json=self.goal)
        self.assertEqual(response.status_code, 409)
        self.conn.commit.assert_not_called()
        self.conn.rollback.assert_called_once()

    def test_update_and_delete_missing_goal_return_not_found(self):
        self.cur.fetchone.return_value = None
        self.cur.rowcount = 0
        for method in (self.client.put, self.client.delete):
            response = method("/api/reading/goals/missing", json=dict(self.goal, id="missing"))
            self.assertEqual(response.status_code, 404)
        self.conn.commit.assert_not_called()

    def test_update_rejects_mismatched_id(self):
        response = self.client.put("/api/reading/goals/other", json=self.goal)
        self.assertEqual(response.status_code, 400)
        self.cur.execute.assert_not_called()

    def test_member_write_failure_rolls_back_parent_change(self):
        self.cur.executemany.side_effect = pymysql.err.OperationalError(2013, "Lost connection")
        for method, path in ((self.client.post, "/api/reading/goals"),
                             (self.client.put, "/api/reading/goals/goal-1")):
            with self.subTest(path=path):
                self.conn.reset_mock()
                response = method(path, json=self.goal)
                self.assertEqual(response.status_code, 503)
                self.conn.rollback.assert_called_once()
                self.conn.commit.assert_not_called()

    def test_import_validates_entire_batch_before_writes(self):
        response = self.client.post("/api/reading/goals/import", json={
            "goals": [self.goal, dict(self.goal, id="bad", ids=[])]
        })
        self.assertEqual(response.status_code, 400)
        self.cur.execute.assert_not_called()

    def test_import_skips_existing_and_duplicate_batch_ids(self):
        # 第一个 ID 已存在；第二个 ID 首次插入；批内重复项不再写入。
        counts = iter([0, 1])
        def execute(sql, params=None):
            self.cur.rowcount = next(counts)
            return self.cur.rowcount
        self.cur.execute.side_effect = execute
        response = self.client.post("/api/reading/goals/import", json={"goals": [
            self.goal, dict(self.goal, id="new"), dict(self.goal, id="new")
        ]})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json, {"imported": 1})
        self.cur.executemany.assert_called_once()
        self.conn.commit.assert_called_once()

    def test_import_second_member_failure_rolls_back_whole_batch(self):
        self.cur.executemany.side_effect = [None, pymysql.err.OperationalError(2013, "Lost connection")]
        response = self.client.post("/api/reading/goals/import", json={
            "goals": [self.goal, dict(self.goal, id="second")]
        })
        self.assertEqual(response.status_code, 503)
        self.conn.rollback.assert_called_once()
        self.conn.commit.assert_not_called()

    def test_database_failures_never_return_successful_empty_data(self):
        self.connect.side_effect = pymysql.err.OperationalError(2003, "Unavailable")
        for method, path, body in (
            (self.client.get, "/api/reading/goals", None),
            (self.client.post, "/api/reading/goals", self.goal),
            (self.client.put, "/api/reading/goals/goal-1", self.goal),
            (self.client.delete, "/api/reading/goals/goal-1", None),
            (self.client.post, "/api/reading/goals/import", {"goals": [self.goal]}),
        ):
            with self.subTest(path=path, method=method):
                response = method(path, json=body)
                self.assertEqual(response.status_code, 503)
                self.assertIn("数据库", response.json["error"])

    def test_commit_and_rollback_connection_failures_still_return_503(self):
        self.conn.commit.side_effect = pymysql.err.OperationalError(2013, "Lost connection")
        self.conn.rollback.side_effect = pymysql.err.InterfaceError(0, "Connection closed")
        response = self.client.post("/api/reading/goals", json=self.goal)
        self.assertEqual(response.status_code, 503)
        self.conn.close.assert_called_once()


if __name__ == "__main__":
    unittest.main()