修复SQLite历史业务数据升级并增加独立程序回归
This commit is contained in:
parent
0e16d28c1b
commit
fd2810bf03
@ -31,10 +31,17 @@ type sqliteBusinessFixture struct {
|
||||
certificate certificate.SavedResource
|
||||
}
|
||||
|
||||
func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture {
|
||||
func newSQLiteBusinessAssets(t *testing.T) sqliteBusinessFixture {
|
||||
t.Helper()
|
||||
fixture := sqliteBusinessFixture{config: config.Default(t.TempDir()), password: rand.Text()}
|
||||
root := t.TempDir()
|
||||
fixture := sqliteBusinessFixture{config: config.Default(root), password: rand.Text()}
|
||||
cfg := &fixture.config
|
||||
// 明确覆盖全部路径，Linux 测试也不能读写正式安装目录。
|
||||
cfg.Paths = config.PathsConfig{
|
||||
Install: root, Config: filepath.Join(root, "conf"), Data: filepath.Join(root, "data"),
|
||||
Certificate: filepath.Join(root, "ssl"), Log: filepath.Join(root, "log"), Backup: filepath.Join(root, "backup"),
|
||||
}
|
||||
cfg.Database.DSN = filepath.Join(cfg.Paths.Data, "scert.db")
|
||||
for _, directory := range []string{cfg.Paths.Config, cfg.Paths.Data, cfg.Paths.Certificate, cfg.Paths.Backup} {
|
||||
if err := os.MkdirAll(directory, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
@ -69,6 +76,13 @@ func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture {
|
||||
}
|
||||
}
|
||||
fixture.certificate = saveSQLiteBusinessCertificate(t, *cfg)
|
||||
return fixture
|
||||
}
|
||||
|
||||
func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture {
|
||||
t.Helper()
|
||||
fixture := newSQLiteBusinessAssets(t)
|
||||
cfg := &fixture.config
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(fixture.password), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
229
backend/internal/cli/sqlite_upgrade_cli_test.go
Normal file
229
backend/internal/cli/sqlite_upgrade_cli_test.go
Normal file
@ -0,0 +1,229 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/repository"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/secretstore"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// 显式指定当批构建的产品程序；不能用测试子进程或辅助脚本冒充独立 CLI。
|
||||
func TestSQLiteBusinessUpgradeStandaloneCLI(t *testing.T) {
|
||||
binary := os.Getenv("SCERT_SQLITE_TEST_CLI")
|
||||
if binary == "" {
|
||||
t.Skip("未指定本批构建的 Scert 独立程序")
|
||||
}
|
||||
if !filepath.IsAbs(binary) || !isRegularFile(binary) {
|
||||
t.Fatal("独立 CLI 测试要求有效的产品程序绝对路径")
|
||||
}
|
||||
fixture := newSQLiteBusinessAssets(t)
|
||||
cfg := &fixture.config
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(fixture.password), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.SetupReady = true
|
||||
cfg.Admin = config.AdminConfig{Username: "upgrade-owner", Email: "owner@example.test", PasswordHash: string(hash)}
|
||||
cfg.DNS.OwnerUserID = "primary-super-admin"
|
||||
// 仅构造隔离历史配置，不把预置状态当成真实 DNS 或 Web 初始化证据。
|
||||
cfg.DNS.VerificationState = "verified"
|
||||
cfg.Web.Hostname = "example.test"
|
||||
if err := cfg.Save(fixture.path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY, applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for index, name := range []string{"0001_initial.sql", "0002_domain_options.sql", "0003_deployment.sql"} {
|
||||
content, err := os.ReadFile(filepath.Join("..", "database", "migrations", "sqlite", name))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), strings.TrimPrefix(string(content), "\ufeff")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO schema_migrations (version) VALUES (?)", index+1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, statement := range []string{
|
||||
`INSERT INTO settings (key, value) VALUES ('historical', '历史设置')`,
|
||||
`INSERT INTO domains (id, name, certificate_type, sans_json, status, dns_provider) VALUES ('domain', 'example.test', 'single', '["example.test"]', 'issued', 'alidns')`,
|
||||
`INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('task', 'domain', 'deploy', 'success', '历史任务')`,
|
||||
`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) VALUES ('server', '历史服务器', '192.0.2.10', 22, 'deploy', 'secret://ssh/business', 'test-only', '/test/cert.pem', '/test/key.pem', '/test/chain.pem', 'nginx', '["nginx","-s","reload"]')`,
|
||||
`INSERT INTO domain_servers (domain_id, server_id) VALUES ('domain', 'server')`,
|
||||
`INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('deploy', 'domain', 'domain', 'success', '历史审计')`,
|
||||
} {
|
||||
if _, err := db.DB.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
historical := snapshotSQLiteCLIState(t, *cfg)
|
||||
archive := filepath.Join(t.TempDir(), "historical-backup")
|
||||
runSQLiteProductCLI(t, binary, fixture, "backup", "--output", archive)
|
||||
runSQLiteProductCLI(t, binary, fixture, "certificate", "list")
|
||||
assertSQLiteCLIUpgradeUsable(t, fixture)
|
||||
upgraded := snapshotSQLiteCLIState(t, *cfg)
|
||||
currentArchive := filepath.Join(t.TempDir(), "upgraded-backup")
|
||||
runSQLiteProductCLI(t, binary, fixture, "backup", "--output", currentArchive)
|
||||
|
||||
for _, scenario := range []struct {
|
||||
name string
|
||||
archive string
|
||||
want map[string]string
|
||||
}{
|
||||
{"historical_backup", archive, historical},
|
||||
{"upgraded_backup", currentArchive, upgraded},
|
||||
} {
|
||||
t.Run(scenario.name, func(t *testing.T) {
|
||||
current, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := current.DB.ExecContext(t.Context(), "UPDATE tasks SET message = '不应保留'; DELETE FROM domain_servers"); err != nil {
|
||||
_ = current.Close()
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := current.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(fixture.certificate.CertificatePath, []byte("待恢复证书"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := secretstore.Open(cfg.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Put(t.Context(), cfg.DNS.CredentialRef, map[string]string{"changed": "待恢复"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runSQLiteProductCLI(t, binary, fixture, "restore", "--input", scenario.archive)
|
||||
if actual := snapshotSQLiteCLIState(t, *cfg); !reflect.DeepEqual(actual, scenario.want) {
|
||||
t.Fatal("独立 CLI 未恢复全部原始模式和表内容")
|
||||
}
|
||||
// 从旧备份恢复后再次走正式应用入口升级，不由测试直接调用 Migrate。
|
||||
runSQLiteProductCLI(t, binary, fixture, "certificate", "list")
|
||||
assertSQLiteCLIUpgradeUsable(t, fixture)
|
||||
if err := validateBackupManifest(scenario.archive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func runSQLiteProductCLI(t *testing.T, binary string, fixture sqliteBusinessFixture, args ...string) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, binary, append([]string{"--config", fixture.path}, args...)...)
|
||||
cmd.Dir = fixture.config.Paths.Install
|
||||
cmd.Env = make([]string, 0)
|
||||
for _, entry := range os.Environ() {
|
||||
if !strings.HasPrefix(strings.ToUpper(entry), "SCERT_") {
|
||||
cmd.Env = append(cmd.Env, entry)
|
||||
}
|
||||
}
|
||||
cmd.WaitDelay = 2 * time.Second
|
||||
output, err := cmd.CombinedOutput()
|
||||
for _, values := range fixture.secrets {
|
||||
for _, value := range values {
|
||||
if strings.Contains(string(output), value) {
|
||||
t.Fatal("独立 CLI 输出包含敏感凭据")
|
||||
}
|
||||
}
|
||||
}
|
||||
if strings.Contains(string(output), fixture.password) {
|
||||
t.Fatal("独立 CLI 输出包含用户密码")
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("独立 CLI %s 失败：%v：%s", args[0], err, output)
|
||||
}
|
||||
}
|
||||
|
||||
func snapshotSQLiteCLIState(t *testing.T, cfg config.Config) map[string]string {
|
||||
t.Helper()
|
||||
db, err := database.Open(t.Context(), cfg.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
result := map[string]string{"schema": snapshotSQLiteRows(t, db.DB, "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name")}
|
||||
rows, err := db.DB.QueryContext(t.Context(), "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var tables []string
|
||||
for rows.Next() {
|
||||
var name string
|
||||
if err := rows.Scan(&name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tables = append(tables, name)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, name := range tables {
|
||||
result[name] = snapshotSQLiteRows(t, db.DB, `SELECT * FROM "`+strings.ReplaceAll(name, `"`, `""`)+`" ORDER BY 1`)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func assertSQLiteCLIUpgradeUsable(t *testing.T, fixture sqliteBusinessFixture) {
|
||||
t.Helper()
|
||||
db, err := database.Open(t.Context(), fixture.config.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
var version, bindings int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT MAX(version) FROM schema_migrations").Scan(&version); err != nil || version != 9 {
|
||||
t.Fatalf("正式应用入口未完成历史升级：%d，%v", version, err)
|
||||
}
|
||||
task, err := repository.NewSQLTaskRepository(db.DB).GetTask(t.Context(), "task")
|
||||
if err != nil || task.Message != "历史任务" || task.DomainID != "domain" || task.Status != "success" {
|
||||
t.Fatalf("历史任务不可用：%v", err)
|
||||
}
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM domain_servers WHERE domain_id = 'domain' AND server_id = 'server'").Scan(&bindings); err != nil || bindings != 1 {
|
||||
t.Fatalf("历史部署绑定丢失：%v", err)
|
||||
}
|
||||
user, err := repository.NewSQLUserRepository(db.DB).FindByIdentifier(t.Context(), "upgrade-owner")
|
||||
if err != nil || bcrypt.CompareHashAndPassword([]byte(user.PasswordHash), []byte(fixture.password)) != nil {
|
||||
t.Fatalf("升级后用户密码不可用：%v", err)
|
||||
}
|
||||
store, err := secretstore.Open(fixture.config.Paths.Config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for reference, want := range fixture.secrets {
|
||||
actual, err := store.Get(t.Context(), reference)
|
||||
if err != nil || !reflect.DeepEqual(actual, want) {
|
||||
t.Fatalf("升级恢复后加密凭据不一致：%v", err)
|
||||
}
|
||||
}
|
||||
if _, err := tls.LoadX509KeyPair(fixture.certificate.CertificatePath, fixture.certificate.PrivateKeyPath); err != nil {
|
||||
t.Fatalf("升级恢复后证书与私钥不可用：%v", err)
|
||||
}
|
||||
}
|
||||
@ -71,6 +71,9 @@ func applyMigration(ctx context.Context, db *sql.DB, driver string, version int)
|
||||
return fmt.Errorf("读取迁移文件失败：%w", err)
|
||||
}
|
||||
statements := migrationStatements(content)
|
||||
if driver == "sqlite" && version == 4 {
|
||||
return applySQLiteTaskMigration(ctx, db, statements, version)
|
||||
}
|
||||
if driver == "mysql" {
|
||||
for _, statement := range statements {
|
||||
skip, err := shouldSkipMySQLStatement(ctx, db, statement)
|
||||
|
||||
@ -4,8 +4,6 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
)
|
||||
|
||||
func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) {
|
||||
@ -26,7 +24,7 @@ func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) {
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
db, err := Open(t.Context(), config.Default(t.TempDir()).Database)
|
||||
db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -73,7 +71,7 @@ func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) {
|
||||
func TestSQLiteUpgradeFromEverySupportedVersion(t *testing.T) {
|
||||
for version := 0; version <= latestMigrationVersion; version++ {
|
||||
t.Run(fmt.Sprintf("version_%d", version), func(t *testing.T) {
|
||||
db, err := Open(t.Context(), config.Default(t.TempDir()).Database)
|
||||
db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
207
backend/internal/database/sqlite_business_upgrade_test.go
Normal file
207
backend/internal/database/sqlite_business_upgrade_test.go
Normal file
@ -0,0 +1,207 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
)
|
||||
|
||||
// 历史版本必须先写入业务数据；空表迁移不能证明真实升级可用。
|
||||
func TestSQLiteBusinessUpgradeFromEveryVersion(t *testing.T) {
|
||||
for version := 1; version <= latestMigrationVersion; version++ {
|
||||
t.Run(fmt.Sprintf("version_%d", version), func(t *testing.T) {
|
||||
cfg := sqliteBusinessUpgradeConfig(t)
|
||||
db, err := Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
createSQLiteHistoricalBusiness(t, db.DB, version)
|
||||
before := snapshotSQLiteHistoricalBusiness(t, db.DB)
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err = Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
for round := 0; round < 2; round++ {
|
||||
if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil {
|
||||
t.Fatalf("历史版本 %d 第 %d 次升级失败：%v", version, round+1, err)
|
||||
}
|
||||
assertSQLiteHistoricalBusiness(t, db.DB, before)
|
||||
}
|
||||
var history int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM schema_migrations").Scan(&history); err != nil || history != latestMigrationVersion {
|
||||
t.Fatalf("升级后的迁移历史不完整：%d，%v", history, err)
|
||||
}
|
||||
var timestamp time.Time
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT updated_at FROM tasks WHERE id = 'task'").Scan(×tamp); err != nil || timestamp.IsZero() {
|
||||
t.Fatalf("历史任务更新时间不可用：%v", err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), `INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('new-task', 'domain', 'deploy', 'queued', '升级后任务')`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT updated_at FROM tasks WHERE id = 'new-task'").Scan(×tamp); err != nil || time.Since(timestamp) < -time.Second || time.Since(timestamp) > time.Minute {
|
||||
t.Fatalf("新任务必须保留当前时间默认值：%v", err)
|
||||
}
|
||||
result, err := db.DB.ExecContext(t.Context(), `INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('upgrade', 'task', 'new-task', 'success', '升级后审计')`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if id, err := result.LastInsertId(); err != nil || id != 18 {
|
||||
t.Fatalf("审计自增序列错误：%d，%v", id, err)
|
||||
}
|
||||
var check string
|
||||
if err := db.DB.QueryRowContext(t.Context(), "PRAGMA integrity_check").Scan(&check); err != nil || check != "ok" {
|
||||
t.Fatalf("升级后完整性检查失败：%s，%v", check, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func sqliteBusinessUpgradeConfig(t *testing.T) config.DatabaseConfig {
|
||||
t.Helper()
|
||||
cfg := config.Default(t.TempDir()).Database
|
||||
// Linux 默认安装路径不随 baseDir 改变，测试必须显式指定临时数据库。
|
||||
cfg.DSN = filepath.Join(t.TempDir(), "historical.db")
|
||||
return cfg
|
||||
}
|
||||
|
||||
func createSQLiteHistoricalBusiness(t *testing.T, db *sql.DB, version int) {
|
||||
t.Helper()
|
||||
if _, err := db.ExecContext(t.Context(), migrationTableSQL("sqlite")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for next := 1; next <= version; next++ {
|
||||
if err := applyMigration(t.Context(), db, "sqlite", next); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
statements := []string{
|
||||
`INSERT INTO settings (key, value) VALUES ('historical', '历史设置：测试')`,
|
||||
`INSERT INTO domains (id, name, certificate_type, sans_json, status, expires_at) VALUES ('domain', 'example.test', 'single', '["example.test"]', 'issued', '2030-01-01 00:00:00')`,
|
||||
`INSERT INTO tasks (id, domain_id, task_type, status, message, started_at, finished_at, created_at) VALUES ('task', 'domain', 'deploy', 'success', '历史部署任务', '2020-01-01 01:00:00', '2020-01-01 02:00:00', '2020-01-01 00:00:00')`,
|
||||
`INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('pending', 'domain', 'renew', 'queued', '历史待执行任务')`,
|
||||
`INSERT INTO audit_logs (id, action, resource_type, resource_id, status, message) VALUES (17, 'deploy', 'domain', 'domain', 'success', '历史审计记录')`,
|
||||
}
|
||||
if version >= 2 {
|
||||
statements = append(statements, `UPDATE domains SET auto_renew = 0, acme_environment = 'production', dns_provider = 'alidns'`)
|
||||
}
|
||||
if version >= 3 {
|
||||
statements = append(statements,
|
||||
`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) VALUES ('server', '历史部署服务器', '192.0.2.10', 22, 'deploy', 'secret://ssh/historical', 'test-only', '/test/cert.pem', '/test/key.pem', '/test/chain.pem', 'nginx', '["nginx","-s","reload"]')`,
|
||||
`INSERT INTO domain_servers (domain_id, server_id) VALUES ('domain', 'server')`)
|
||||
}
|
||||
if version >= 4 {
|
||||
statements = append(statements,
|
||||
`UPDATE tasks SET stage = 'completed', operator_id = 'owner', certificate_version_id = 'historical-certificate', idempotency_key = 'historical-operation' WHERE id = 'task'`,
|
||||
`INSERT INTO task_attempts (id, task_id, attempt_number, stage, status, message) VALUES ('attempt', 'task', 1, 'deploy', 'success', '历史部署尝试')`)
|
||||
}
|
||||
if version >= 5 {
|
||||
statements = append(statements, `INSERT INTO users (id, username, email, password_hash, role) VALUES ('owner', 'historical-owner', 'owner@example.test', 'historical-hash', 'super_admin')`)
|
||||
}
|
||||
if version >= 6 {
|
||||
statements = append(statements, `UPDATE servers SET validation_command_json = '["nginx","-t"]'`)
|
||||
}
|
||||
if version >= 7 {
|
||||
statements = append(statements, `UPDATE servers SET auth_method = 'password', password_ref = 'secret://ssh/password'`)
|
||||
}
|
||||
if version >= 8 {
|
||||
statements = append(statements, `UPDATE users SET groups_json = '["历史分组"]'`)
|
||||
}
|
||||
if version >= 9 {
|
||||
statements = append(statements, `UPDATE servers SET ssh_key_status = 'deployed', ssh_key_fingerprint = 'test-fingerprint', ssh_key_deployed_at = '2020-01-01T00:00:00Z'`)
|
||||
}
|
||||
for _, statement := range statements {
|
||||
if _, err := db.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type sqliteHistoricalTable struct {
|
||||
query string
|
||||
rows string
|
||||
}
|
||||
|
||||
func snapshotSQLiteHistoricalBusiness(t *testing.T, db *sql.DB) map[string]sqliteHistoricalTable {
|
||||
t.Helper()
|
||||
result := make(map[string]sqliteHistoricalTable)
|
||||
for _, name := range []string{"settings", "domains", "tasks", "audit_logs", "servers", "domain_servers", "task_attempts", "users", "sqlite_sequence"} {
|
||||
var exists int
|
||||
if err := db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?", name).Scan(&exists); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if exists == 0 {
|
||||
continue
|
||||
}
|
||||
rows, err := db.QueryContext(t.Context(), `SELECT * FROM "`+name+`" LIMIT 0`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
columns, err := rows.Columns()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for index, column := range columns {
|
||||
columns[index] = `"` + strings.ReplaceAll(column, `"`, `""`) + `"`
|
||||
}
|
||||
query := "SELECT " + strings.Join(columns, ",") + ` FROM "` + name + `" ORDER BY 1`
|
||||
result[name] = sqliteHistoricalTable{query: query, rows: snapshotSQLiteUpgradeQuery(t, db, query)}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func snapshotSQLiteUpgradeQuery(t *testing.T, db *sql.DB, query string) string {
|
||||
t.Helper()
|
||||
rows, err := db.QueryContext(t.Context(), query)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
columns, err := rows.Columns()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var result [][]any
|
||||
for rows.Next() {
|
||||
values := make([]any, len(columns))
|
||||
destinations := make([]any, len(columns))
|
||||
for index := range values {
|
||||
destinations[index] = &values[index]
|
||||
}
|
||||
if err := rows.Scan(destinations...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result = append(result, values)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
encoded, err := json.Marshal(result)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(encoded)
|
||||
}
|
||||
|
||||
func assertSQLiteHistoricalBusiness(t *testing.T, db *sql.DB, expected map[string]sqliteHistoricalTable) {
|
||||
t.Helper()
|
||||
for table, snapshot := range expected {
|
||||
if actual := snapshotSQLiteUpgradeQuery(t, db, snapshot.query); !reflect.DeepEqual(actual, snapshot.rows) {
|
||||
t.Errorf("历史表 %s 的原有业务字段发生变化", table)
|
||||
}
|
||||
}
|
||||
}
|
||||
183
backend/internal/database/sqlite_task_migration.go
Normal file
183
backend/internal/database/sqlite_task_migration.go
Normal file
@ -0,0 +1,183 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"database/sql/driver"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 固定物理连接后按 SQLite 表重建流程处理迁移，外键设置不能泄漏到连接池。
|
||||
func applySQLiteTaskMigration(ctx context.Context, db *sql.DB, statements []string, version int) (err error) {
|
||||
conn, err := db.Conn(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取 SQLite 迁移连接失败：%w", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
var foreignKeys int
|
||||
if err := conn.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil {
|
||||
return fmt.Errorf("读取 SQLite 外键设置失败：%w", err)
|
||||
}
|
||||
defer func() {
|
||||
cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_, resetErr := conn.ExecContext(cleanup, fmt.Sprintf("PRAGMA foreign_keys = %d", foreignKeys))
|
||||
if resetErr != nil {
|
||||
// 不能把外键状态未知的连接继续交给后续业务使用。
|
||||
_ = conn.Raw(func(any) error { return driver.ErrBadConn })
|
||||
err = errors.Join(err, fmt.Errorf("恢复 SQLite 外键设置失败：%w", resetErr))
|
||||
}
|
||||
}()
|
||||
if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil {
|
||||
return fmt.Errorf("设置 SQLite 迁移连接失败：%w", err)
|
||||
}
|
||||
tx, err := conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("开始 SQLite 迁移事务失败：%w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
for _, statement := range statements {
|
||||
if statement == "ALTER TABLE tasks ADD COLUMN updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP" {
|
||||
err = rebuildSQLiteTasks(ctx, tx)
|
||||
} else {
|
||||
_, err = tx.ExecContext(ctx, statement)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("执行 SQLite 任务迁移失败：%w", err)
|
||||
}
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, "PRAGMA foreign_key_check")
|
||||
if err != nil {
|
||||
return fmt.Errorf("检查 SQLite 迁移外键失败：%w", err)
|
||||
}
|
||||
invalid := rows.Next()
|
||||
rowErr := errors.Join(rows.Err(), rows.Close())
|
||||
if rowErr != nil {
|
||||
return fmt.Errorf("读取 SQLite 迁移外键结果失败：%w", rowErr)
|
||||
}
|
||||
if invalid {
|
||||
return errors.New("SQLite 任务迁移存在外键冲突，已拒绝提交")
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, "INSERT INTO schema_migrations (version) VALUES (?)", version); err != nil {
|
||||
return fmt.Errorf("记录 SQLite 迁移版本失败：%w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("提交 SQLite 迁移事务失败：%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
var sqliteTaskTableHeader = regexp.MustCompile(`(?is)^CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:tasks|"tasks"|` + "`tasks`" + `|\[tasks\])\s*\(`)
|
||||
|
||||
type sqliteTaskSchemaObject struct {
|
||||
kind string
|
||||
name string
|
||||
sql string
|
||||
}
|
||||
|
||||
func rebuildSQLiteTasks(ctx context.Context, tx *sql.Tx) error {
|
||||
var definition string
|
||||
if err := tx.QueryRowContext(ctx, "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'tasks'").Scan(&definition); err != nil {
|
||||
return err
|
||||
}
|
||||
header := sqliteTaskTableHeader.FindStringIndex(definition)
|
||||
end := strings.LastIndex(definition, ")")
|
||||
if header == nil || end < header[1] {
|
||||
return errors.New("无法安全识别 SQLite 任务表定义，拒绝重建")
|
||||
}
|
||||
// 保留原始列、约束和表选项，不用硬编码的旧表字段覆盖扩展结构。
|
||||
create := `CREATE TABLE "scert_tasks_migration_4" (` + definition[header[1]:end] + ", updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP" + definition[end:]
|
||||
columns, err := sqliteTaskCopyColumns(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, `SELECT type, name, sql FROM sqlite_master WHERE sql IS NOT NULL AND (type IN ('view', 'trigger') OR (type = 'index' AND tbl_name = 'tasks')) ORDER BY type, name`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var objects []sqliteTaskSchemaObject
|
||||
for rows.Next() {
|
||||
var object sqliteTaskSchemaObject
|
||||
if err := rows.Scan(&object.kind, &object.name, &object.sql); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
objects = append(objects, object)
|
||||
}
|
||||
if err := errors.Join(rows.Err(), rows.Close()); err != nil {
|
||||
return err
|
||||
}
|
||||
// 名字冲突时直接失败，不覆盖或删除已有对象。
|
||||
if _, err := tx.ExecContext(ctx, create); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `INSERT INTO "scert_tasks_migration_4" (`+columns+`) SELECT `+columns+` FROM tasks`); err != nil {
|
||||
return err
|
||||
}
|
||||
// 重命名会校验整个模式，临时移除依赖视图和触发器后按原定义恢复。
|
||||
for _, object := range objects {
|
||||
if object.kind == "view" || object.kind == "trigger" {
|
||||
if _, err := tx.ExecContext(ctx, "DROP "+strings.ToUpper(object.kind)+" "+quoteSQLiteMigrationName(object.name)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, statement := range []string{"DROP TABLE tasks", `ALTER TABLE "scert_tasks_migration_4" RENAME TO tasks`} {
|
||||
if _, err := tx.ExecContext(ctx, statement); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// 先恢复视图，最后恢复触发器，兼容视图上的 INSTEAD OF 触发器。
|
||||
for _, kind := range []string{"index", "view", "trigger"} {
|
||||
for _, object := range objects {
|
||||
if object.kind == kind {
|
||||
if _, err := tx.ExecContext(ctx, object.sql); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func sqliteTaskCopyColumns(ctx context.Context, tx *sql.Tx) (string, error) {
|
||||
rows, err := tx.QueryContext(ctx, "SELECT name FROM pragma_table_xinfo('tasks') WHERE hidden = 0 ORDER BY cid")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var columns []string
|
||||
names := make(map[string]bool)
|
||||
for rows.Next() {
|
||||
var name string
|
||||
if err := rows.Scan(&name); err != nil {
|
||||
_ = rows.Close()
|
||||
return "", err
|
||||
}
|
||||
names[strings.ToLower(name)] = true
|
||||
columns = append(columns, quoteSQLiteMigrationName(name))
|
||||
}
|
||||
if err := errors.Join(rows.Err(), rows.Close()); err != nil {
|
||||
return "", err
|
||||
}
|
||||
var withoutRowID int
|
||||
if err := tx.QueryRowContext(ctx, "SELECT wr FROM pragma_table_list WHERE schema = 'main' AND name = 'tasks'").Scan(&withoutRowID); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if withoutRowID == 0 {
|
||||
for _, alias := range []string{"rowid", "_rowid_", "oid"} {
|
||||
if !names[alias] {
|
||||
columns = append([]string{quoteSQLiteMigrationName(alias)}, columns...)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return strings.Join(columns, ","), nil
|
||||
}
|
||||
|
||||
func quoteSQLiteMigrationName(name string) string {
|
||||
return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
|
||||
}
|
||||
122
backend/internal/database/sqlite_task_migration_test.go
Normal file
122
backend/internal/database/sqlite_task_migration_test.go
Normal file
@ -0,0 +1,122 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSQLiteBusinessUpgradePreservesDependentObjects(t *testing.T) {
|
||||
for _, foreignKeys := range []int{0, 1} {
|
||||
t.Run(fmt.Sprintf("foreign_keys_%d", foreignKeys), func(t *testing.T) {
|
||||
db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
db.DB.SetMaxOpenConns(1)
|
||||
createSQLiteHistoricalBusiness(t, db.DB, 3)
|
||||
for _, statement := range []string{
|
||||
fmt.Sprintf("PRAGMA foreign_keys = %d", foreignKeys),
|
||||
`ALTER TABLE tasks ADD COLUMN extra TEXT NOT NULL DEFAULT '扩展值)'`,
|
||||
`ALTER TABLE tasks ADD COLUMN generated TEXT GENERATED ALWAYS AS (message || extra) VIRTUAL`,
|
||||
`UPDATE tasks SET rowid = 77 WHERE id = 'task'`,
|
||||
`CREATE UNIQUE INDEX "历史索引" ON tasks(extra, id) WHERE status = 'success'`,
|
||||
`CREATE TABLE task_links (task_id TEXT PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE)`,
|
||||
`INSERT INTO task_links VALUES ('task')`,
|
||||
`CREATE TABLE trigger_events (message TEXT)`,
|
||||
`CREATE TRIGGER "历史任务触发器" AFTER UPDATE ON tasks BEGIN INSERT INTO trigger_events VALUES (new.message); END`,
|
||||
`CREATE VIEW z_task_view AS SELECT id, message FROM tasks`,
|
||||
`CREATE VIEW a_task_view AS SELECT * FROM z_task_view`,
|
||||
`CREATE TRIGGER view_insert INSTEAD OF INSERT ON a_task_view BEGIN UPDATE tasks SET message = new.message WHERE id = new.id; END`,
|
||||
`CREATE TRIGGER external_insert AFTER INSERT ON settings BEGIN UPDATE tasks SET message = new.value WHERE id = 'pending'; END`,
|
||||
} {
|
||||
if _, err := db.DB.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
before := snapshotSQLiteHistoricalBusiness(t, db.DB)
|
||||
objectsQuery := `SELECT type, name, sql FROM sqlite_master WHERE name IN ('历史索引', '历史任务触发器', 'z_task_view', 'a_task_view', 'view_insert', 'external_insert', 'task_links') ORDER BY name`
|
||||
objects := snapshotSQLiteUpgradeQuery(t, db.DB, objectsQuery)
|
||||
generated := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT rowid, id, generated FROM tasks ORDER BY id")
|
||||
if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertSQLiteHistoricalBusiness(t, db.DB, before)
|
||||
if actual := snapshotSQLiteUpgradeQuery(t, db.DB, objectsQuery); actual != objects {
|
||||
t.Fatal("依赖对象定义未完整保留")
|
||||
}
|
||||
if actual := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT rowid, id, generated FROM tasks ORDER BY id"); actual != generated {
|
||||
t.Fatal("行号或生成列发生变化")
|
||||
}
|
||||
var count, actualForeignKeys int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM task_links WHERE task_id = 'task'").Scan(&count); err != nil || count != 1 {
|
||||
t.Fatalf("任务表重建触发了级联删除：%v", err)
|
||||
}
|
||||
if err := db.DB.QueryRowContext(t.Context(), "PRAGMA foreign_keys").Scan(&actualForeignKeys); err != nil || actualForeignKeys != foreignKeys {
|
||||
t.Fatalf("连接的外键设置未恢复：%d，%v", actualForeignKeys, err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), `INSERT INTO a_task_view VALUES ('task', '升级后视图写入'); INSERT INTO settings (key, value) VALUES ('after', '升级后外部触发器')`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM trigger_events").Scan(&count); err != nil || count != 2 {
|
||||
t.Fatalf("触发器执行结果错误：%d，%v", count, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLiteBusinessUpgradeFailureRollsBackAndRetries(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
prepare string
|
||||
cleanup string
|
||||
message string
|
||||
}{
|
||||
{"version_record", `CREATE TRIGGER reject_version BEFORE INSERT ON schema_migrations WHEN new.version = 4 BEGIN SELECT RAISE(ABORT, 'blocked-upgrade'); END`, "DROP TRIGGER reject_version", "blocked-upgrade"},
|
||||
{"temporary_name", `CREATE TABLE scert_tasks_migration_4 (value TEXT); INSERT INTO scert_tasks_migration_4 VALUES ('保留内容')`, "DROP TABLE scert_tasks_migration_4", "already exists"},
|
||||
{"foreign_key", `CREATE TABLE broken_links (id TEXT REFERENCES tasks(id)); INSERT INTO broken_links VALUES ('missing-task')`, "DELETE FROM broken_links", "外键冲突"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
db.DB.SetMaxOpenConns(1)
|
||||
createSQLiteHistoricalBusiness(t, db.DB, 3)
|
||||
if _, err := db.DB.ExecContext(t.Context(), tc.prepare); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), "PRAGMA foreign_keys = ON"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := snapshotSQLiteHistoricalBusiness(t, db.DB)
|
||||
schemaQuery := "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name"
|
||||
schema := snapshotSQLiteUpgradeQuery(t, db.DB, schemaQuery)
|
||||
history := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT * FROM schema_migrations ORDER BY version")
|
||||
if err := Migrate(t.Context(), db.DB, "sqlite"); err == nil || !strings.Contains(err.Error(), tc.message) {
|
||||
t.Fatalf("未拒绝有阻塞项的迁移：%v", err)
|
||||
}
|
||||
assertSQLiteHistoricalBusiness(t, db.DB, before)
|
||||
if actual := snapshotSQLiteUpgradeQuery(t, db.DB, schemaQuery); actual != schema {
|
||||
t.Fatal("迁移失败后表、索引或触发器定义被改变")
|
||||
}
|
||||
if actual := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT * FROM schema_migrations ORDER BY version"); actual != history {
|
||||
t.Fatal("迁移失败后版本被错误推进")
|
||||
}
|
||||
var foreignKeys int
|
||||
if err := db.DB.QueryRowContext(t.Context(), "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil || foreignKeys != 1 {
|
||||
t.Fatalf("失败路径未恢复外键设置：%d，%v", foreignKeys, err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), tc.cleanup); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil {
|
||||
t.Fatalf("排除阻塞后无法重新升级：%v", err)
|
||||
}
|
||||
assertSQLiteHistoricalBusiness(t, db.DB, before)
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user