You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
87 lines
5.1 KiB
87 lines
5.1 KiB
"""V21 MySQL 实测。只允许显式指定 qinghe_integrity_test_ 前缀的独立测试库。"""
|
|
import os
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("QINGHE_INTEGRITY_TEST_URL"), "未配置独立 MySQL 测试库")
|
|
class OrphanMigrationMySqlTest(unittest.TestCase):
|
|
def setUp(self):
|
|
import pymysql
|
|
endpoint = urlsplit(os.environ["QINGHE_INTEGRITY_TEST_URL"].removeprefix("jdbc:"))
|
|
database = endpoint.path.strip("/")
|
|
if not re.fullmatch(r"qinghe_integrity_test_[a-z0-9_]+", database):
|
|
raise ValueError("禁止使用业务数据库运行迁移测试")
|
|
self.connection = pymysql.connect(host=endpoint.hostname, port=endpoint.port or 3306,
|
|
database=database, user=os.environ["QINGHE_INTEGRITY_TEST_USER"],
|
|
password=os.environ["QINGHE_INTEGRITY_TEST_PASSWORD"], charset="utf8mb4")
|
|
self.addCleanup(self.connection.close)
|
|
self.tables = ["moment", "moment_comment", "moment_comment_like", "video_play_record",
|
|
"moment_comment_bak_v21_orphan", "moment_comment_like_bak_v21_orphan",
|
|
"video_play_record_bak_v21_orphan"]
|
|
with self.connection.cursor() as cursor:
|
|
for table in reversed(self.tables):
|
|
cursor.execute("DROP TABLE IF EXISTS `" + table + "`")
|
|
cursor.execute("CREATE TABLE moment (id VARCHAR(36) PRIMARY KEY,deleted TINYINT NOT NULL) ENGINE=InnoDB")
|
|
cursor.execute("CREATE TABLE moment_comment (id VARCHAR(36) PRIMARY KEY,moment_id VARCHAR(36),content TEXT) ENGINE=InnoDB")
|
|
cursor.execute("CREATE TABLE moment_comment_like (id VARCHAR(36) PRIMARY KEY,comment_id VARCHAR(36)) ENGINE=InnoDB")
|
|
cursor.execute("CREATE TABLE video_play_record (id VARCHAR(36) PRIMARY KEY,video_id VARCHAR(36),play_position INT) ENGINE=InnoDB")
|
|
cursor.execute("INSERT INTO moment VALUES ('active',0),('soft-deleted',1)")
|
|
cursor.execute("INSERT INTO moment_comment VALUES ('valid','active','保留'),('soft','soft-deleted','软删仍保留'),('orphan','absent','隔离')")
|
|
cursor.execute("INSERT INTO moment_comment_like VALUES ('valid-like','valid'),('orphan-like','orphan'),('missing-comment','absent')")
|
|
cursor.execute("INSERT INTO video_play_record VALUES ('valid-progress','active',10),('soft-progress','soft-deleted',20),('orphan-progress','absent',30)")
|
|
self.connection.commit()
|
|
|
|
def run_migration(self, apply=False):
|
|
script = (Path(__file__).resolve().parents[1] /
|
|
"src/main/resources/db/migration/V21__quarantine_orphan_moment_dependencies.sql").read_text(encoding="utf-8")
|
|
# 本脚本不含存储过程或字符串内分号；先移除整行注释。
|
|
script = "\n".join(line for line in script.splitlines() if not line.lstrip().startswith("--"))
|
|
try:
|
|
with self.connection.cursor() as cursor:
|
|
cursor.execute("SET @qinghe_v21_apply = %s", (1 if apply else 0,))
|
|
for statement in script.split(";"):
|
|
if statement.strip():
|
|
cursor.execute(statement)
|
|
except Exception:
|
|
self.connection.rollback()
|
|
raise
|
|
|
|
def count(self, table):
|
|
with self.connection.cursor() as cursor:
|
|
cursor.execute("SELECT COUNT(*) FROM `" + table + "`")
|
|
return cursor.fetchone()[0]
|
|
|
|
def test_dry_run_does_not_change_business_data_or_create_backups(self):
|
|
self.run_migration()
|
|
self.assertEqual(3, self.count("moment_comment"))
|
|
self.assertEqual(3, self.count("video_play_record"))
|
|
with self.connection.cursor() as cursor:
|
|
cursor.execute("SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE '%bak_v21%'")
|
|
self.assertEqual(0, cursor.fetchone()[0])
|
|
|
|
def test_apply_backs_up_orphans_preserves_soft_deleted_and_can_repeat(self):
|
|
self.run_migration(True)
|
|
for table, count in [("moment_comment",2),("moment_comment_like",1),("video_play_record",2),
|
|
("moment_comment_bak_v21_orphan",1),("moment_comment_like_bak_v21_orphan",2),
|
|
("video_play_record_bak_v21_orphan",1)]:
|
|
self.assertEqual(count, self.count(table), table)
|
|
self.run_migration(True)
|
|
self.assertEqual(1, self.count("moment_comment_bak_v21_orphan"))
|
|
self.assertEqual(2, self.count("moment_comment"))
|
|
|
|
def test_backup_failure_stops_before_deletion(self):
|
|
with self.connection.cursor() as cursor:
|
|
cursor.execute("CREATE TABLE video_play_record_bak_v21_orphan (id VARCHAR(36) PRIMARY KEY) ENGINE=InnoDB")
|
|
with self.assertRaises(Exception):
|
|
self.run_migration(True)
|
|
self.assertEqual(3, self.count("moment_comment"))
|
|
self.assertEqual(3, self.count("video_play_record"))
|
|
self.assertEqual(0, self.count("moment_comment_bak_v21_orphan"))
|
|
self.assertEqual(0, self.count("moment_comment_like_bak_v21_orphan"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|