289 lines
11 KiB
Go
289 lines
11 KiB
Go
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)
|
||
}
|
||
}
|