"""V22 合成数据实测，只使用显式独立测试库。"""
import os
import re
import unittest
from pathlib import Path
from urllib.parse import urlsplit


@unittest.skipUnless(os.getenv("QINGHE_INTEGRITY_TEST_URL"), "未配置独立测试库")
class IndexMigrationTest(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.db = 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", cursorclass=pymysql.cursors.DictCursor)
        self.addCleanup(self.db.close)
        self.specs = [
            ("activity_payment_order","registration_id","UK_dw3dsgr7ik4xqb0p4l5lwt1t7","uk_payment_registration",True),
            ("user_session","token_hash","UK_au1tr2kouh0rhp1us6nqjpmyi","idx_session_token_hash",True),
            ("gift_record","sender_id","idx_gift_record_sender","idx_gr_sender",False),
            ("activity_participant","activity_id,user_id","idx_ap_activity_user","uk_ap_activity_user",False),
            ("moment_like","moment_id,user_id","idx_like_moment_user","uk_like_moment_user",False),
            ("moment_comment_like","comment_id,user_id","idx_clike_comment_user","uk_clike_comment_user",False)]
        with self.db.cursor() as q:
            for t in ["chat_message","notification"] + [s[0] for s in self.specs]:
                q.execute("DROP TABLE IF EXISTS `"+t+"`")
            q.execute("CREATE TABLE chat_message(seq_id BIGINT PRIMARY KEY,message_id VARCHAR(36),conversation_id VARCHAR(100),deleted TINYINT,created_at DATETIME,content TEXT) ENGINE=InnoDB")
            q.execute("CREATE TABLE notification(id BIGINT PRIMARY KEY,user_id VARCHAR(64),deleted TINYINT,is_read TINYINT,type VARCHAR(50),content TEXT) ENGINE=InnoDB")
            for t,cols,old,keep,unique in self.specs:
                fields=",".join("`"+c+"` VARCHAR(36)" for c in cols.split(","))
                q.execute("CREATE TABLE `"+t+"`(id BIGINT PRIMARY KEY,"+fields+", "+("UNIQUE " if unique else "")+"KEY `"+old+"`("+cols+"), "+("" if t=="gift_record" else "UNIQUE ")+"KEY `"+keep+"`("+cols+")) ENGINE=InnoDB")
            q.execute("ALTER TABLE gift_record ADD receiver_id VARCHAR(36),ADD KEY idx_gift_record_receiver(receiver_id),ADD KEY idx_gr_receiver(receiver_id)")
            q.executemany("INSERT INTO chat_message VALUES(%s,%s,%s,%s,%s,%s)",
                [(i,"m"+str(i),"c"+str(i%200),0,"2026-10-22 00:00:00","测试") for i in range(12000)])
            q.executemany("INSERT INTO notification VALUES(%s,%s,%s,%s,%s,%s)",
                [(i,"u"+str(i%200),0,i%3==0,"t"+str(i%7),"测试") for i in range(12000)])
        self.db.commit()

    def migrate(self):
        script=(Path(__file__).resolve().parents[1]/"src/main/resources/db/migration/V22__optimize_chat_notification_indexes.sql").read_text(encoding="utf-8")
        script="\n".join(s for s in script.splitlines() if not s.lstrip().startswith("--"))
        with self.db.cursor() as q:
            for statement in script.split(";"):
                if statement.strip(): q.execute(statement)

    def test_repeatable_migration_plans_and_uniqueness(self):
        self.migrate(); self.migrate()
        with self.db.cursor() as q:
            q.execute("ANALYZE TABLE chat_message,notification")
            plans = [
                ("SELECT * FROM chat_message WHERE conversation_id='c10' AND deleted=0 AND seq_id>0 ORDER BY seq_id LIMIT 20","idx_chat_conv_deleted_seq"),
                ("SELECT * FROM chat_message WHERE conversation_id='c10' AND deleted=0 AND seq_id>0 ORDER BY created_at,seq_id LIMIT 20","idx_chat_conv_deleted_time_seq"),
                ("SELECT * FROM chat_message WHERE message_id='m10'","idx_chat_message_id"),
                ("SELECT * FROM notification WHERE user_id='u10' AND deleted=0 AND is_read=0 LIMIT 20","idx_notification_user_deleted_read"),
                ("SELECT * FROM notification WHERE user_id='u10' AND deleted=0 AND type='t1' LIMIT 20","idx_notification_user_deleted_type")]
            for sql,index in plans:
                q.execute("EXPLAIN "+sql); plan=q.fetchone()
                if index == "idx_chat_conv_deleted_time_seq":
                    # 小会话可能由成本模型选择seq范围再排序，不能把候选索引当成必选项。
                    self.assertIn(plan["key"], (index, "idx_chat_conv_deleted_seq"), str(plan))
                    q.execute("EXPLAIN "+sql.replace("FROM chat_message", "FROM chat_message FORCE INDEX (idx_chat_conv_deleted_time_seq)"))
                    ordered=q.fetchone()
                    self.assertNotIn("filesort",ordered["Extra"] or "")
                else:
                    self.assertEqual(index,plan["key"],str(plan))
                    if "ORDER BY" in sql:self.assertNotIn("filesort",plan["Extra"] or "")
            for t,cols,old,keep,unique in self.specs:
                q.execute("SELECT INDEX_NAME,NON_UNIQUE FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s",(t,))
                indexes=q.fetchall();self.assertFalse(any(r["INDEX_NAME"]==old for r in indexes))
                self.assertTrue(any(r["INDEX_NAME"]==keep for r in indexes))
                if t!="gift_record":self.assertTrue(any(r["INDEX_NAME"]==keep and r["NON_UNIQUE"]==0 for r in indexes))
            q.execute("SELECT COUNT(*) n FROM chat_message");self.assertEqual(12000,q.fetchone()["n"])
            q.execute("SELECT COUNT(*) n FROM notification");self.assertEqual(12000,q.fetchone()["n"])

    def test_wrong_named_index_definition_stops_without_dropping_duplicates(self):
        with self.db.cursor() as q:q.execute("ALTER TABLE chat_message ADD INDEX idx_chat_conv_deleted_seq(message_id)")
        with self.assertRaises(Exception):self.migrate()
        with self.db.cursor() as q:
            q.execute("SHOW INDEX FROM activity_payment_order WHERE Key_name='UK_dw3dsgr7ik4xqb0p4l5lwt1t7'")
            self.assertEqual(1,len(q.fetchall()))