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.
52 lines
3.3 KiB
52 lines
3.3 KiB
"""话题ID扩展/回填验证，复用独立测试库连接配置。"""
|
|
import os
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
@unittest.skipUnless(os.getenv("QINGHE_INTEGRITY_TEST_URL"), "未配置独立测试库")
|
|
class TopicIdMigrationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
import pymysql
|
|
e=urlsplit(os.environ["QINGHE_INTEGRITY_TEST_URL"].removeprefix("jdbc:"))
|
|
if not re.fullmatch(r"qinghe_integrity_test_[a-z0-9_]+", e.path.strip("/")):raise ValueError("禁止业务库")
|
|
self.db=pymysql.connect(host=e.hostname,port=e.port or 3306,database=e.path.strip("/"),
|
|
user=os.environ["QINGHE_INTEGRITY_TEST_USER"],password=os.environ["QINGHE_INTEGRITY_TEST_PASSWORD"],charset="utf8mb4")
|
|
self.addCleanup(self.db.close)
|
|
self.tables=["topic_follow","topic_favorite","topic_visit","topic_hot_snapshot"]
|
|
with self.db.cursor() as q:
|
|
for t in self.tables+ [t+"_bak_v24_id" for t in self.tables]+["topic"]:q.execute("DROP TABLE IF EXISTS `"+t+"`")
|
|
q.execute("CREATE TABLE topic(id VARCHAR(36) PRIMARY KEY,name VARCHAR(100) UNIQUE,created_at DATETIME,deleted TINYINT) ENGINE=InnoDB COLLATE=utf8mb4_unicode_ci")
|
|
q.execute("INSERT INTO topic VALUES('old-id','旧名_deleted_old', '2026-01-01',1),('new-id','旧名','2026-10-01',0)")
|
|
for t in self.tables:
|
|
q.execute("CREATE TABLE `"+t+"` (id VARCHAR(36) PRIMARY KEY,user_id VARCHAR(36),topic_name VARCHAR(100),created_at DATETIME,captured_at DATETIME) ENGINE=InnoDB")
|
|
q.execute("INSERT INTO `"+t+"`(id,user_id,topic_name,created_at) VALUES('valid','u1','旧名','2026-10-02'),('deleted-parent','u2','旧名_deleted_old','2026-02-01'),('ambiguous','u3','旧名','2026-02-01'),('missing','u4','不存在','2026-10-02')")
|
|
self.db.commit()
|
|
|
|
def run_script(self, file):
|
|
text=(Path(__file__).resolve().parents[1]/"src/main/resources/db/migration"/file).read_text(encoding="utf-8")
|
|
text="\n".join(line for line in text.splitlines() if not line.lstrip().startswith("--"))
|
|
try:
|
|
with self.db.cursor() as q:
|
|
for s in text.split(";"):
|
|
if s.strip():q.execute(s)
|
|
except Exception:self.db.rollback();raise
|
|
|
|
def test_expand_preview_apply_repeat_and_ambiguous_rows(self):
|
|
self.run_script("V23__expand_topic_stable_id.sql")
|
|
self.run_script("V23__expand_topic_stable_id.sql")
|
|
self.run_script("V24__backfill_topic_stable_id.sql")
|
|
with self.db.cursor() as q:
|
|
q.execute("SELECT COUNT(*) FROM topic_follow WHERE topic_id IS NOT NULL");self.assertEqual(0,q.fetchone()[0])
|
|
q.execute("SET @topic_id_apply=1")
|
|
self.run_script("V24__backfill_topic_stable_id.sql")
|
|
with self.db.cursor() as q:
|
|
for t in self.tables:
|
|
q.execute("SELECT id,topic_id FROM `"+t+"`");rows=dict(q.fetchall())
|
|
self.assertEqual("new-id",rows["valid"]);self.assertEqual("old-id",rows["deleted-parent"])
|
|
self.assertIsNone(rows["ambiguous"]);self.assertIsNone(rows["missing"])
|
|
q.execute("SET @topic_id_apply=1")
|
|
self.run_script("V24__backfill_topic_stable_id.sql")
|
|
with self.db.cursor() as q:
|
|
q.execute("SELECT COUNT(*) FROM topic_follow_bak_v24_id");self.assertEqual(2,q.fetchone()[0])
|
|
|