refactor(api): 去除代码版本命名

This commit is contained in:
admin committed 2026-09-14 15:35:53 +08:00
1 parent bbc4bdd140
commit dcf4cb0719
31 files changed
+174 -154

No files matched your search

+26 -21
View File
@@ -7,21 +7,21 @@ from unittest.mock import MagicMock, patch
import pymysql
from flask import Flask
from api_v2 import v2_bp
from reading_api import reading_bp
class ReadingGoalsTest(unittest.TestCase):
def setUp(self):
app = Flask(__name__)
app.config["TESTING"] = True
app.register_blueprint(v2_bp)
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("api_v2.pymysql.connect", return_value=self.conn).start()
self.connect = patch("reading_api.pymysql.connect", return_value=self.conn).start()
self.addCleanup(patch.stopall)
self.goal = {
"id": "goal-1", "name": " 九月阅读 ",
@@ -31,7 +31,7 @@ class ReadingGoalsTest(unittest.TestCase):
}
def test_create_normalizes_and_commits_unknown_members(self):
response = self.client.post("/api/v2/goals", json=self.goal)
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})
@@ -39,6 +39,11 @@ class ReadingGoalsTest(unittest.TestCase):
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},
@@ -51,7 +56,7 @@ class ReadingGoalsTest(unittest.TestCase):
]
for fields in invalid:
with self.subTest(fields=fields):
response = self.client.post("/api/v2/goals", json=dict(self.goal, **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()
@@ -59,7 +64,7 @@ class ReadingGoalsTest(unittest.TestCase):
def test_invalid_json_and_body_shape_are_chinese_errors(self):
for body in ("{", "null", "[]"):
response = self.client.post("/api/v2/goals", data=body, content_type="application/json")
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"]))
@@ -70,7 +75,7 @@ class ReadingGoalsTest(unittest.TestCase):
{"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/v2/goals")
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=["科幻书单"]
@@ -78,7 +83,7 @@ class ReadingGoalsTest(unittest.TestCase):
def test_create_duplicate_returns_conflict(self):
self.cur.execute.side_effect = pymysql.err.IntegrityError(1062, "Duplicate entry")
response = self.client.post("/api/v2/goals", json=self.goal)
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()
@@ -87,19 +92,19 @@ class ReadingGoalsTest(unittest.TestCase):
self.cur.fetchone.return_value = None
self.cur.rowcount = 0
for method in (self.client.put, self.client.delete):
response = method("/api/v2/goals/missing", json=dict(self.goal, id="missing"))
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/v2/goals/other", json=self.goal)
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/v2/goals"),
(self.client.put, "/api/v2/goals/goal-1")):
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)
@@ -108,7 +113,7 @@ class ReadingGoalsTest(unittest.TestCase):
self.conn.commit.assert_not_called()
def test_import_validates_entire_batch_before_writes(self):
response = self.client.post("/api/v2/goals/import", json={
response = self.client.post("/api/reading/goals/import", json={
"goals": [self.goal, dict(self.goal, id="bad", ids=[])]
})
self.assertEqual(response.status_code, 400)
@@ -121,7 +126,7 @@ class ReadingGoalsTest(unittest.TestCase):
self.cur.rowcount = next(counts)
return self.cur.rowcount
self.cur.execute.side_effect = execute
response = self.client.post("/api/v2/goals/import", json={"goals": [
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)
@@ -131,7 +136,7 @@ class ReadingGoalsTest(unittest.TestCase):
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/v2/goals/import", json={
response = self.client.post("/api/reading/goals/import", json={
"goals": [self.goal, dict(self.goal, id="second")]
})
self.assertEqual(response.status_code, 503)
@@ -141,11 +146,11 @@ class ReadingGoalsTest(unittest.TestCase):
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/v2/goals", None),
(self.client.post, "/api/v2/goals", self.goal),
(self.client.put, "/api/v2/goals/goal-1", self.goal),
(self.client.delete, "/api/v2/goals/goal-1", None),
(self.client.post, "/api/v2/goals/import", {"goals": [self.goal]}),
(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)
@@ -155,7 +160,7 @@ class ReadingGoalsTest(unittest.TestCase):
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/v2/goals", json=self.goal)
response = self.client.post("/api/reading/goals", json=self.goal)
self.assertEqual(response.status_code, 503)
self.conn.close.assert_called_once()
+14 -14
View File
@@ -7,7 +7,7 @@ from unittest.mock import patch
import pymysql
from flask import Flask
import api_v2
import reading_api
@unittest.skipUnless(os.environ.get('WEREAD_TEST_MYSQL') == '1', '需显式启用 MySQL 集成测试')
@@ -15,48 +15,48 @@ 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)
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/v2/goals', json=goal)
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/v2/goals').json['goals']}
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/v2/goals/{ids[0]}', json=changed).status_code, 200)
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/v2/goals/import', json={'goals': [goal, imported, imported]})
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/v2/goals').json['goals']}
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 = api_v2.insert_goal
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('api_v2.insert_goal', side_effect=fail_second):
response = client.post('/api/v2/goals/import', json={
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/v2/goals').json['goals']}
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/v2/goals/{ids[0]}').status_code, 200)
with pymysql.connect(**api_v2.MYSQL_CONFIG) as conn:
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(**api_v2.MYSQL_CONFIG) as conn:
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()