"""阅读目标接口回归测试；数据库故障通过连接边界模拟，不访问线上库。""" 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()