diff --git a/backend/internal/database/migrate.go b/backend/internal/database/migrate.go
index a08c2df..cec56fc 100644
--- a/backend/internal/database/migrate.go
+++ b/backend/internal/database/migrate.go
@@ -168,7 +168,9 @@ func migrationName(version int) string {
 
 func migrationStatements(content []byte) []string {
 	statements := make([]string, 0)
-	for _, statement := range strings.Split(string(content), ";") {
+	// 只移除文件头的编码标记，不能替换 SQL 字面量中的相同字符。
+	script := strings.TrimPrefix(string(content), "\ufeff")
+	for _, statement := range strings.Split(script, ";") {
 		statement = strings.TrimSpace(statement)
 		if statement != "" {
 			statements = append(statements, statement)
diff --git a/backend/internal/database/migration_encoding_test.go b/backend/internal/database/migration_encoding_test.go
new file mode 100644
index 0000000..43d5bcb
--- /dev/null
+++ b/backend/internal/database/migration_encoding_test.go
@@ -0,0 +1,32 @@
+package database
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestMigrationStatementsIgnoreOnlyLeadingBOM(t *testing.T) {
+	want := []string{"ALTER TABLE servers ADD COLUMN auth_method VARCHAR(32)", "SELECT '\ufeff'"}
+	for _, prefix := range []string{"", "\ufeff"} {
+		got := migrationStatements([]byte(prefix + want[0] + ";\n" + want[1] + ";\n"))
+		if !reflect.DeepEqual(got, want) {
+			t.Fatalf("编码标记处理改变了 SQL：%q", got)
+		}
+	}
+}
+
+func TestEmbeddedMySQLPasswordMigrationSupportsResume(t *testing.T) {
+	content, err := migrationFiles.ReadFile("migrations/mysql/0007_server_password_auth.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	statements := migrationStatements(content)
+	if len(statements) != 2 {
+		t.Fatalf("密码认证迁移语句数量不正确：%d", len(statements))
+	}
+	for _, statement := range statements {
+		if !mysqlAddColumnPattern.MatchString(statement) {
+			t.Fatalf("迁移无法被字段重入检查识别：%q", statement)
+		}
+	}
+}
diff --git a/backend/internal/database/mysql_integration_helpers_test.go b/backend/internal/database/mysql_integration_helpers_test.go
new file mode 100644
index 0000000..95d73be
--- /dev/null
+++ b/backend/internal/database/mysql_integration_helpers_test.go
@@ -0,0 +1,265 @@
+package database
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"database/sql"
+	"encoding/hex"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+
+	"git.pcux.cn/xuans/Scert/backend/internal/config"
+)
+
+type realMySQL struct {
+	root     string
+	binary   string
+	version  string
+	port     string
+	password string
+	tlsName  string
+	admin    *sql.DB
+	command  *exec.Cmd
+	exit     chan error
+	sequence int
+}
+
+// 只接受程序目录与产物目录；不接受现有实例连接串，也不读取机器上的默认配置。
+func startRealMySQL(t *testing.T) *realMySQL {
+	t.Helper()
+	bin := os.Getenv("SCERT_MYSQL_TEST_BIN")
+	if bin == "" {
+		t.Skip("未启用 MySQL 隔离实库测试")
+	}
+	root := os.Getenv("SCERT_MYSQL_TEST_ROOT")
+	if !filepath.IsAbs(bin) || !filepath.IsAbs(root) {
+		t.Fatal("MySQL 实库测试要求程序与隔离产物目录使用绝对路径")
+	}
+	binary := filepath.Join(bin, "mysqld")
+	if runtime.GOOS == "windows" {
+		binary += ".exe"
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	output, err := exec.CommandContext(ctx, binary, "--no-defaults", "--version").Output()
+	if err != nil {
+		t.Fatal("无法读取 MySQL 服务端版本")
+	}
+	version := string(output)
+	if !strings.Contains(version, "Ver 5.7.") && !strings.Contains(version, "Ver 8.0.") {
+		t.Fatal("隔离测试仅接受明确支持目标 MySQL 5.7 或 8.0")
+	}
+	if err := os.MkdirAll(root, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	root, err = os.MkdirTemp(root, "scert-mysql-integration-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	password := make([]byte, 32)
+	if _, err := rand.Read(password); err != nil {
+		t.Fatal(err)
+	}
+	f := &realMySQL{root: root, binary: binary, version: version, password: hex.EncodeToString(password)}
+	t.Logf("独立 MySQL 程序：%s；产物保留于：%s", strings.TrimSpace(version), root)
+	f.tlsName = "scert-mysql-" + filepath.Base(root)
+	f.createTLS(t)
+	t.Cleanup(func() { mysql.DeregisterTLSConfig(f.tlsName) })
+	listener, err := net.Listen("tcp4", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.port = strconv.Itoa(listener.Addr().(*net.TCPAddr).Port)
+	if err := listener.Close(); err != nil {
+		t.Fatal(err)
+	}
+	// 初始化阶段不启动监听；首次启动在开放连接前为所有测试管理账户设随机密码。
+	initCtx, initCancel := context.WithTimeout(t.Context(), 2*time.Minute)
+	defer initCancel()
+	args := append(f.baseArgs(), "--initialize-insecure")
+	if err := exec.CommandContext(initCtx, binary, args...).Run(); err != nil {
+		t.Fatal("初始化独立 MySQL 失败，请检查保留目录中的服务端日志")
+	}
+	bootstrap := "ALTER USER 'root'@'localhost' IDENTIFIED BY '" + f.password + "';\n" +
+		"CREATE USER 'scert_harness'@'127.0.0.1' IDENTIFIED BY '" + f.password + "' REQUIRE SSL;\n" +
+		"GRANT ALL PRIVILEGES ON *.* TO 'scert_harness'@'127.0.0.1' WITH GRANT OPTION;\n"
+	f.write(t, "bootstrap.sql", []byte(bootstrap))
+	defer f.write(t, "bootstrap.sql", nil)
+	t.Cleanup(func() { f.stop(t) })
+	f.start(t, true)
+	// 初始口令文件不进入证据；保留空文件而非删除测试产物。
+	f.write(t, "bootstrap.sql", nil)
+	return f
+}
+
+func (f *realMySQL) baseArgs() []string {
+	return []string{"--no-defaults", "--basedir=" + filepath.Dir(filepath.Dir(f.binary)),
+		"--datadir=" + filepath.Join(f.root, "data"), "--log-error=" + filepath.Join(f.root, "mysql.log"),
+		"--innodb-buffer-pool-size=32M", "--innodb-log-file-size=8M", "--performance-schema=OFF"}
+}
+
+func (f *realMySQL) start(t *testing.T, bootstrap bool) {
+	t.Helper()
+	args := append(f.baseArgs(), "--bind-address=127.0.0.1", "--port="+f.port,
+		"--skip-name-resolve", "--require-secure-transport=ON", "--max-connections=20",
+		"--ssl-ca="+filepath.Join(f.root, "server.crt"), "--ssl-cert="+filepath.Join(f.root, "server.crt"),
+		"--ssl-key="+filepath.Join(f.root, "server.key"), "--local-infile=OFF", "--secure-file-priv=NULL",
+		"--character-set-server=utf8mb4", "--collation-server=utf8mb4_unicode_ci")
+	if strings.Contains(f.version, "Ver 8.0.") {
+		args = append(args, "--mysqlx=OFF")
+	}
+	if runtime.GOOS != "windows" {
+		args = append(args, "--socket="+filepath.Join(f.root, "mysql.sock"))
+	}
+	if bootstrap {
+		args = append(args, "--init-file="+filepath.Join(f.root, "bootstrap.sql"))
+	}
+	f.command = exec.Command(f.binary, args...)
+	if err := f.command.Start(); err != nil {
+		f.command = nil
+		t.Fatal("启动独立 MySQL 失败")
+	}
+	f.exit = make(chan error, 1)
+	command := f.command
+	go func() { f.exit <- command.Wait() }()
+	var err error
+	f.admin, err = sql.Open("mysql", f.dsn("scert_harness", ""))
+	if err != nil {
+		t.Fatal("创建独立 MySQL 管理连接失败")
+	}
+	deadline := time.Now().Add(time.Minute)
+	for time.Now().Before(deadline) {
+		select {
+		case <-f.exit:
+			f.command = nil
+			t.Fatal("独立 MySQL 已退出，请检查保留日志")
+		default:
+		}
+		ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+		err := f.admin.PingContext(ctx)
+		cancel()
+		if err == nil {
+			var version string
+			if err := f.admin.QueryRowContext(t.Context(), "SELECT VERSION()").Scan(&version); err != nil {
+				t.Fatal(err)
+			}
+			t.Logf("实连版本：%s；仅监听 127.0.0.1:%s；严格 TLS", version, f.port)
+			return
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	t.Fatal("独立 MySQL 未在时限内就绪，请检查保留日志；不降级 TLS")
+}
+
+func (f *realMySQL) stop(t *testing.T) {
+	t.Helper()
+	if f.command == nil {
+		if f.admin != nil {
+			_ = f.admin.Close()
+		}
+		return
+	}
+	if f.admin != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		_, _ = f.admin.ExecContext(ctx, "SHUTDOWN")
+		cancel()
+		_ = f.admin.Close()
+	}
+	select {
+	case err := <-f.exit:
+		if err != nil {
+			t.Errorf("独立 MySQL 未正常退出：%v", err)
+		}
+	case <-time.After(15 * time.Second):
+		_ = f.command.Process.Kill()
+		<-f.exit
+		t.Error("独立 MySQL 停止超时，已终止本测试启动的进程")
+	}
+	f.command = nil
+}
+
+func (f *realMySQL) dsn(user, name string) string {
+	cfg := mysql.NewConfig()
+	cfg.User, cfg.Passwd = user, f.password
+	cfg.Net, cfg.Addr = "tcp", net.JoinHostPort("127.0.0.1", f.port)
+	cfg.DBName, cfg.TLSConfig = name, f.tlsName
+	cfg.Timeout, cfg.ReadTimeout, cfg.WriteTimeout = 3*time.Second, 10*time.Second, 10*time.Second
+	cfg.ParseTime = true
+	return cfg.FormatDSN()
+}
+
+func (f *realMySQL) database(t *testing.T) *Database {
+	t.Helper()
+	f.sequence++
+	name := fmt.Sprintf("scert_test_%d", f.sequence)
+	user := fmt.Sprintf("scert_user_%d", f.sequence)
+	for _, statement := range []string{
+		"CREATE DATABASE " + name + " CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci",
+		"CREATE USER '" + user + "'@'127.0.0.1' IDENTIFIED BY '" + f.password + "' REQUIRE SSL",
+		"GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP, REFERENCES ON " + name + ".* TO '" + user + "'@'127.0.0.1'",
+	} {
+		if _, err := f.admin.ExecContext(t.Context(), statement); err != nil {
+			t.Fatal("创建隔离数据库及库级账户失败")
+		}
+	}
+	return f.open(t, user, name)
+}
+
+func (f *realMySQL) open(t *testing.T, user, name string) *Database {
+	t.Helper()
+	db, err := Open(t.Context(), config.DatabaseConfig{Driver: "mysql", DSN: f.dsn(user, name), MaxOpenConns: 2, MaxIdleConns: 1})
+	if err != nil {
+		t.Fatal("Scert 打开隔离数据库失败")
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return db
+}
+
+func (f *realMySQL) write(t *testing.T, name string, data []byte) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(f.root, name), data, 0o600); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *realMySQL) createTLS(t *testing.T) {
+	t.Helper()
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+	certificate := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "Scert isolated MySQL"},
+		NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), IsCA: true,
+		BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}}
+	der, err := x509.CreateCertificate(rand.Reader, certificate, certificate, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	encoded := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+	f.write(t, "server.crt", encoded)
+	f.write(t, "server.key", pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)}))
+	roots := x509.NewCertPool()
+	if !roots.AppendCertsFromPEM(encoded) {
+		t.Fatal("解析隔离测试证书失败")
+	}
+	if err := mysql.RegisterTLSConfig(f.tlsName, &tls.Config{RootCAs: roots, ServerName: "127.0.0.1", MinVersion: tls.VersionTLS12}); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/backend/internal/database/mysql_integration_test.go b/backend/internal/database/mysql_integration_test.go
new file mode 100644
index 0000000..14dd59e
--- /dev/null
+++ b/backend/internal/database/mysql_integration_test.go
@@ -0,0 +1,288 @@
+package database
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+func TestMySQLRealMigrations(t *testing.T) {
+	f := startRealMySQL(t)
+	baseline := f.database(t)
+	if err := Migrate(t.Context(), baseline.DB, "mysql"); err != nil {
+		t.Fatal(err)
+	}
+	expectedSchema := mysqlSchemaSnapshot(t, baseline.DB)
+	if err := baseline.Close(); err != nil {
+		t.Fatal(err)
+	}
+	t.Run("transport_security", func(t *testing.T) {
+		var required bool
+		if err := f.admin.QueryRowContext(t.Context(), "SELECT @@require_secure_transport").Scan(&required); err != nil || !required {
+			t.Fatal("服务端必须强制安全传输", err)
+		}
+		var name, cipher string
+		if err := f.admin.QueryRowContext(t.Context(), "SHOW SESSION STATUS LIKE 'Ssl_cipher'").Scan(&name, &cipher); err != nil || cipher == "" {
+			t.Fatal("实连必须使用 TLS", err)
+		}
+		encoded, err := os.ReadFile(filepath.Join(f.root, "server.crt"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		roots := x509.NewCertPool()
+		if !roots.AppendCertsFromPEM(encoded) {
+			t.Fatal("解析测试证书失败")
+		}
+		wrongName := f.tlsName + "-wrong-host"
+		if err := mysql.RegisterTLSConfig(wrongName, &tls.Config{RootCAs: roots, ServerName: "wrong.invalid", MinVersion: tls.VersionTLS12}); err != nil {
+			t.Fatal(err)
+		}
+		defer mysql.DeregisterTLSConfig(wrongName)
+		for _, mode := range []string{"false", "true", wrongName} {
+			cfg, err := mysql.ParseDSN(f.dsn("scert_harness", ""))
+			if err != nil {
+				t.Fatal("解析隔离连接参数失败")
+			}
+			cfg.TLSConfig = mode
+			conn, err := sql.Open("mysql", cfg.FormatDSN())
+			if err != nil {
+				t.Fatal("创建反向校验连接失败")
+			}
+			err = conn.PingContext(t.Context())
+			_ = conn.Close()
+			if err == nil {
+				t.Fatal("非加密、不可信证书或错误主机名连接被接受")
+			}
+			var serverError *mysql.MySQLError
+			var authorityError x509.UnknownAuthorityError
+			var hostnameError x509.HostnameError
+			switch mode {
+			case "false":
+				// 账户 REQUIRE SSL 可先于全局传输检查拒绝认证。
+				if !errors.As(err, &serverError) || (serverError.Number != 3159 && serverError.Number != 1045) {
+					t.Fatalf("应由服务端安全传输策略拒绝：%T", err)
+				}
+			case "true":
+				if !errors.As(err, &authorityError) {
+					t.Fatalf("应因证书不可信拒绝：%T", err)
+				}
+			default:
+				if !errors.As(err, &hostnameError) {
+					t.Fatalf("应因主机名不匹配拒绝：%T", err)
+				}
+			}
+		}
+	})
+	for version := 0; version <= latestMigrationVersion; version++ {
+		t.Run(fmt.Sprintf("upgrade_from_%d", version), func(t *testing.T) {
+			db := f.database(t)
+			mysqlPrepareVersion(t, db.DB, version)
+			mysqlAssertUpgrade(t, db.DB, expectedSchema)
+		})
+	}
+	// 遍历每个 DDL 边界，包括全部 DDL 已落盘但版本号未写入的情况。
+	for version := 1; version <= latestMigrationVersion; version++ {
+		content, err := migrationFiles.ReadFile(fmt.Sprintf("migrations/mysql/%04d_%s.sql", version, migrationName(version)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		statements := migrationStatements(content)
+		for count := 1; count <= len(statements); count++ {
+			t.Run(fmt.Sprintf("interrupted_version_%d_after_%d", version, count), func(t *testing.T) {
+				db := f.database(t)
+				mysqlPrepareVersion(t, db.DB, version-1)
+				for _, statement := range statements[:count] {
+					if _, err := db.DB.ExecContext(t.Context(), statement); err != nil {
+						t.Fatal(err)
+					}
+				}
+				mysqlAssertUpgrade(t, db.DB, expectedSchema)
+			})
+		}
+	}
+	t.Run("existing_business_rows", func(t *testing.T) {
+		db := f.database(t)
+		mysqlPrepareVersion(t, db.DB, 6)
+		for _, statement := range []string{
+			"INSERT INTO domains(id,name,certificate_type,sans_json,status) VALUES ('upgrade-domain','upgrade.invalid','single','[]','pending')",
+			"INSERT INTO users(id,username,email,password_hash,role) VALUES ('upgrade-user','upgrade-user','upgrade@example.invalid','hash-placeholder','admin')",
+			"INSERT INTO servers(id,name,address,port,username,private_key_ref,host_key_fingerprint,certificate_path,private_key_path,certificate_chain_path,web_server,reload_command_json,validation_command_json) VALUES ('upgrade-server','test-server','127.0.0.1',22,'test','key-ref','fingerprint','/test/cert','/test/key','/test/chain','nginx','[]','[]')",
+			"INSERT INTO domain_servers(domain_id,server_id) VALUES ('upgrade-domain','upgrade-server')",
+		} {
+			if _, err := db.DB.ExecContext(t.Context(), statement); err != nil {
+				t.Fatal(err)
+			}
+		}
+		mysqlAssertUpgrade(t, db.DB, expectedSchema)
+		var method, status, key, domain string
+		if err := db.DB.QueryRowContext(t.Context(), "SELECT s.auth_method,s.ssh_key_status,s.private_key_ref,d.name FROM servers s JOIN domain_servers ds ON ds.server_id=s.id JOIN domains d ON d.id=ds.domain_id WHERE s.id='upgrade-server'").Scan(&method, &status, &key, &domain); err != nil {
+			t.Fatal(err)
+		}
+		if method != "private_key" || status != "not_configured" || key != "key-ref" || domain != "upgrade.invalid" {
+			t.Fatal("升级改变了原有业务关联或新增字段默认值")
+		}
+		var groups sql.NullString
+		if err := db.DB.QueryRowContext(t.Context(), "SELECT groups_json FROM users WHERE id='upgrade-user'").Scan(&groups); err != nil || groups.Valid {
+			t.Fatal("升级未保留用户及默认分组语义", err)
+		}
+	})
+	t.Run("schema_scoped_account", func(t *testing.T) {
+		db := f.database(t)
+		if _, err := db.DB.ExecContext(t.Context(), "CREATE DATABASE forbidden_database"); err == nil {
+			t.Fatal("测试业务账户不应拥有创建其他库的权限")
+		}
+		if _, err := db.DB.ExecContext(t.Context(), "SELECT * FROM mysql.user"); err == nil {
+			t.Fatal("测试业务账户不应能读取管理账户信息")
+		}
+	})
+	newerHistory := make([]int, latestMigrationVersion+1)
+	for index := range newerHistory {
+		newerHistory[index] = index + 1
+	}
+	for _, invalid := range []struct {
+		name     string
+		versions []int
+		message  string
+	}{
+		{"missing_first", []int{2}, "迁移历史不连续"},
+		{"missing_middle", []int{1, 3}, "迁移历史不连续"},
+		{"zero", []int{0}, "迁移历史不连续"},
+		{"negative", []int{-1}, "迁移历史不连续"},
+		{"newer", newerHistory, "高于当前程序支持"},
+	} {
+		t.Run(invalid.name, func(t *testing.T) {
+			db := f.database(t)
+			mysqlPrepareVersion(t, db.DB, 0)
+			for _, version := range invalid.versions {
+				if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO schema_migrations(version) VALUES (?)", version); err != nil {
+					t.Fatal(err)
+				}
+			}
+			err := Migrate(t.Context(), db.DB, "mysql")
+			if err == nil || !strings.Contains(err.Error(), invalid.message) {
+				t.Fatalf("应拒绝异常历史：%v", err)
+			}
+			var count int
+			if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM schema_migrations").Scan(&count); err != nil || count != len(invalid.versions) {
+				t.Fatalf("异常历史被改变：%d：%v", count, err)
+			}
+			if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE()").Scan(&count); err != nil || count != 2 {
+				t.Fatalf("拒绝迁移时创建了业务表：%d：%v", count, err)
+			}
+			mysqlAssertPreserved(t, db.DB)
+		})
+	}
+	t.Run("restart_persistence", func(t *testing.T) {
+		db := f.database(t)
+		mysqlPrepareVersion(t, db.DB, 0)
+		mysqlAssertUpgrade(t, db.DB, expectedSchema)
+		var name, user string
+		if err := db.DB.QueryRowContext(t.Context(), "SELECT DATABASE(), SUBSTRING_INDEX(CURRENT_USER(), '@', 1)").Scan(&name, &user); err != nil {
+			t.Fatal(err)
+		}
+		// 模拟停机重启时重新打开应用连接；不将旧连接池遇到断线等同于数据丢失。
+		if err := db.Close(); err != nil {
+			t.Fatal(err)
+		}
+		f.stop(t)
+		f.start(t, false)
+		reopened := f.open(t, user, name)
+		mysqlAssertUpgrade(t, reopened.DB, expectedSchema)
+	})
+}
+
+func mysqlPrepareVersion(t *testing.T, db *sql.DB, version int) {
+	t.Helper()
+	for _, statement := range []string{migrationTableSQL("mysql"), "CREATE TABLE preserved_data (value VARCHAR(100) NOT NULL)", "INSERT INTO preserved_data VALUES ('中文迁移数据')"} {
+		if _, err := db.ExecContext(t.Context(), statement); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for next := 1; next <= version; next++ {
+		if err := applyMigration(t.Context(), db, "mysql", next); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func mysqlAssertUpgrade(t *testing.T, db *sql.DB, expectedSchema []string) {
+	t.Helper()
+	for repeat := 0; repeat < 2; repeat++ {
+		if err := Migrate(t.Context(), db, "mysql"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	version, err := currentMigrationVersion(t.Context(), db)
+	if err != nil || version != latestMigrationVersion {
+		t.Fatalf("迁移版本不完整：%d：%v", version, err)
+	}
+	mysqlAssertPreserved(t, db)
+	if actual := mysqlSchemaSnapshot(t, db); !reflect.DeepEqual(actual, expectedSchema) {
+		t.Fatal("升级后的字段、索引、默认值或存储引擎与全新迁移不一致")
+	}
+	var column string
+	if err := db.QueryRowContext(t.Context(), "SELECT column_name FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='servers' AND column_name='ssh_key_status'").Scan(&column); err != nil {
+		t.Fatal("最终业务表结构缺失", err)
+	}
+}
+
+func mysqlSchemaSnapshot(t *testing.T, db *sql.DB) []string {
+	t.Helper()
+	queries := []string{
+		"SELECT table_name,engine,table_collation FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name<>'preserved_data' ORDER BY table_name",
+		"SELECT table_name,column_name,ordinal_position,column_type,is_nullable,column_default,extra,character_set_name,collation_name FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name<>'preserved_data' ORDER BY table_name,ordinal_position",
+		"SELECT table_name,index_name,non_unique,seq_in_index,column_name,sub_part,index_type FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name<>'preserved_data' ORDER BY table_name,index_name,seq_in_index",
+	}
+	var snapshot []string
+	for _, query := range queries {
+		rows, err := db.QueryContext(t.Context(), query)
+		if err != nil {
+			t.Fatal(err)
+		}
+		columns, err := rows.Columns()
+		if err != nil {
+			_ = rows.Close()
+			t.Fatal(err)
+		}
+		for rows.Next() {
+			values := make([]sql.NullString, len(columns))
+			pointers := make([]any, len(columns))
+			for index := range values {
+				pointers[index] = &values[index]
+			}
+			if err := rows.Scan(pointers...); err != nil {
+				_ = rows.Close()
+				t.Fatal(err)
+			}
+			snapshot = append(snapshot, fmt.Sprintf("%#v", values))
+		}
+		if err := rows.Err(); err != nil {
+			_ = rows.Close()
+			t.Fatal(err)
+		}
+		if err := rows.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if len(snapshot) == 0 {
+		t.Fatal("数据库结构快照不能为空")
+	}
+	return snapshot
+}
+
+func mysqlAssertPreserved(t *testing.T, db *sql.DB) {
+	t.Helper()
+	var value string
+	if err := db.QueryRowContext(t.Context(), "SELECT value FROM preserved_data").Scan(&value); err != nil || value != "中文迁移数据" {
+		t.Fatalf("原始数据未保留：%v", err)
+	}
+}