xuans/Scert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善 PostgreSQL 仓储兼容测试

Browse Source
This commit is contained in:
xuans 2026-09-16 13:17:44 +08:00
parent eaf474c8e7
commit e007bdab1f
5 changed files with 105 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
backend/internal/app/app.go
View File

@ -83,8 +83,8 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
}
dnsTester := dnsprovider.NewTester(dnsprovider.NewLegoFactory())
dnsVerification := setup.NewDNSVerificationManager(dnsTester, secretStore)
repo := repository.NewSQLDomainRepository(db.DB)
serverRepo := repository.NewSQLServerRepository(db.DB)
repo := repository.NewSQLDomainRepository(db.DB, cfg.Database.Driver)
serverRepo := repository.NewSQLServerRepository(db.DB, cfg.Database.Driver)
userRepo := repository.NewSQLUserRepository(db.DB, cfg.Database.Driver)
if cfg.SetupReady && cfg.Admin.Username != "" && cfg.Admin.PasswordHash != "" {
if err := userRepo.EnsureSuperAdmin(ctx, repository.User{ID: "primary-super-admin", Username: cfg.Admin.Username, Email: cfg.Admin.Email, PasswordHash: cfg.Admin.PasswordHash}); err != nil {
@ -93,7 +93,7 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
return nil, fmt.Errorf("初始化超级管理员用户失败：%w", err)
}
}
taskRepo := repository.NewSQLTaskRepository(db.DB)
taskRepo := repository.NewSQLTaskRepository(db.DB, cfg.Database.Driver)
if recovered, err := taskRepo.RecoverInterruptedTasks(ctx, time.Now().UTC()); err != nil {
_ = db.Close()
_ = logger.Close()
@ -101,7 +101,7 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string
} else if recovered > 0 {
logger.Warn("\u670d\u52a1\u91cd\u542f\u524d\u672a\u5b8c\u6210\u4efb\u52a1\u5df2\u6807\u8bb0\u4e3a\u5931\u8d25", "count", recovered)
}
auditRepo := repository.NewSQLAuditRepository(db.DB)
auditRepo := repository.NewSQLAuditRepository(db.DB, cfg.Database.Driver)
if deleted, err := auditRepo.PruneExpiredAuditLogs(ctx, time.Now().UTC().AddDate(0, 0, -auditLogRetentionDays)); err != nil {
logger.Warn("启动时清理过期审计日志失败", "error", err)
} else if deleted > 0 {

11
backend/internal/database/database_test.go
View File

@ -3,6 +3,7 @@ package database
import (
"context"
"database/sql"
"fmt"
"path/filepath"
"testing"
@ -61,6 +62,16 @@ func TestMySQLMigrationStatementsRecognizeRecoverableChanges(t *testing.T) {
t.Errorf("可恢复的 MySQL 迁移语句未被识别：%q", item.statement)
}
}
func TestPostgresMigrationFilesAreEmbedded(t *testing.T) {
for version := 1; version <= latestMigrationVersion; version++ {
path := "migrations/postgres/" + fmt.Sprintf("%04d_%s.sql", version, migrationName(version))
content, err := migrationFiles.ReadFile(path)
if err != nil || len(content) == 0 {
t.Fatalf("PostgreSQL 迁移文件不可用：%s: %v", path, err)
}
}
}
func TestSQLiteUpgradeAddsServerValidationCommand(t *testing.T) {
dir := t.TempDir()
cfg := config.Default(dir).Database

37
backend/internal/database/review_test.go
View File

@ -3,6 +3,7 @@ package database
import (
"context"
"path/filepath"
"strings"
"testing"
"time"
@ -13,13 +14,45 @@ import (
func TestOpenRejectsUnknownDriver(t *testing.T) {
cfg := config.Default(t.TempDir()).Database
cfg.Driver = "postgres"
cfg.Driver = "unknown"
cfg.DSN = "unused"
if _, err := Open(context.Background(), cfg); err == nil {
t.Fatal("未知数据库驱动未被拒绝")
}
}
func TestNormalizePostgresDSNRequiresVerifiedTLS(t *testing.T) {
cases := []string{
"postgres://user:password@db.example.test:5432/scert",
"postgres://user:password@db.example.test:5432/scert?sslmode=disable",
"host=db.example.test port=5432 user=user password=password dbname=scert sslmode=require",
}
for _, dsn := range cases {
if _, err := normalizePostgresDSN(dsn); err == nil {
t.Fatalf("未拒绝不安全 PostgreSQL TLS 配置：%s", dsn)
}
}
valid := "postgres://user:password@db.example.test:5432/scert?sslmode=verify-full"
if _, err := normalizePostgresDSN(valid); err != nil {
t.Fatalf("有效 PostgreSQL TLS 配置被拒绝：%v", err)
}
}
func TestSanitizeDSNRedactsPostgresPassword(t *testing.T) {
for _, item := range []struct {
name string
dsn string
}{
{name: "URL", dsn: "postgres://user:secret@db.example.test:5432/scert?sslmode=verify-full"},
{name: "关键字", dsn: "host=db.example.test user=user password=secret dbname=scert sslmode=verify-full"},
} {
sanitized := SanitizeDSN("postgres", item.dsn)
if sanitized == "" || strings.Contains(sanitized, "secret") {
t.Fatalf("%s PostgreSQL DSN 未正确脱敏：%s", item.name, sanitized)
}
}
}
func TestNormalizeMySQLDSNEnablesTimeParsing(t *testing.T) {
dsn, err := normalizeMySQLDSN("user:password@tcp(127.0.0.1:3306)/scert?charset=utf8mb4")
if err != nil {
@ -62,7 +95,7 @@ func TestSQLiteMigrationIsIdempotentAndCreatesBaseTables(t *testing.T) {
if tables != 7 {
t.Fatalf("基础数据表数量不正确：%d", tables)
}
if err := Migrate(context.Background(), db.DB, "postgres"); err == nil {
if err := Migrate(context.Background(), db.DB, "unknown"); err == nil {
t.Fatal("未知迁移驱动未被拒绝")
}
}

28
backend/internal/repository/rebind_test.go Normal file
View File

@ -0,0 +1,28 @@
package repository
import "testing"
func TestRebindPostgresParameters(t *testing.T) {
query := "SELECT * FROM tasks WHERE domain_id = ? AND status = ?"
want := "SELECT * FROM tasks WHERE domain_id = $1 AND status = $2"
if got := Rebind("postgres", query); got != want {
t.Fatalf("PostgreSQL 占位符转换错误：%q", got)
}
}
func TestRebindDoesNotChangeQuotedQuestionMarks(t *testing.T) {
query := "SELECT '?' AS literal, \"?\" AS identifier, `?` AS legacy_identifier WHERE id = ? AND note = 'it''s ?'"
want := "SELECT '?' AS literal, \"?\" AS identifier, `?` AS legacy_identifier WHERE id = $1 AND note = 'it''s ?'"
if got := Rebind("postgres", query); got != want {
t.Fatalf("字符串或标识符中的问号被错误转换：%q", got)
}
}
func TestRebindLeavesSQLiteAndMySQLUntouched(t *testing.T) {
query := "SELECT * FROM tasks WHERE id = ?"
for _, driver := range []string{"sqlite", "mysql", ""} {
if got := Rebind(driver, query); got != query {
t.Fatalf("%s 的占位符不应改变：%q", driver, got)
}
}
}

49
backend/internal/repository/task_repository.go
View File

@ -54,11 +54,16 @@ type TaskRepository interface {
}
type SQLTaskRepository struct {
db *sql.DB
db *sql.DB
driver string
}
func NewSQLTaskRepository(db *sql.DB) *SQLTaskRepository {
return &SQLTaskRepository{db: db}
func NewSQLTaskRepository(db *sql.DB, driver ...string) *SQLTaskRepository {
return &SQLTaskRepository{db: db, driver: repositoryDriver(driver)}
}
func (r *SQLTaskRepository) query(statement string) string {
return Rebind(r.driver, statement)
}
func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error {
@ -77,9 +82,9 @@ func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error {
if updated.IsZero() {
updated = now
}
_, err := r.db.ExecContext(ctx, `INSERT INTO tasks
_, err := r.db.ExecContext(ctx, r.query(`INSERT INTO tasks
(id, domain_id, task_type, status, stage, message, operator_id, certificate_version_id, idempotency_key, started_at, finished_at, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
item.ID, item.DomainID, item.TaskType, item.Status, item.Stage, item.Message, item.OperatorID,
item.CertificateVersionID, item.IdempotencyKey, nullableTime(item.StartedAt), nullableTime(item.FinishedAt), now, updated)
if err != nil {
@ -96,7 +101,7 @@ func (r *SQLTaskRepository) GetTask(ctx context.Context, id string) (Task, error
if id == "" {
return Task{}, errors.New("任务 ID 不能为空")
}
return scanTask(r.db.QueryRowContext(ctx, taskSelect+` WHERE id = ?`, id))
return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE id = ?`), id))
}
func (r *SQLTaskRepository) ListTasks(ctx context.Context, domainID string, limit int) ([]Task, error) {
@ -113,7 +118,7 @@ func (r *SQLTaskRepository) ListTasks(ctx context.Context, domainID string, limi
args = append(args, domainID)
}
query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT %d", limit)
rows, err := r.db.QueryContext(ctx, query, args...)
rows, err := r.db.QueryContext(ctx, r.query(query), args...)
if err != nil {
return nil, fmt.Errorf("查询任务列表失败：%w", err)
}
@ -145,7 +150,7 @@ func (r *SQLTaskRepository) UpdateTask(ctx context.Context, id, status, stage, m
if !validTaskStatus(status) || stage == "" {
return errors.New("任务状态或阶段无效")
}
result, err := r.db.ExecContext(ctx, `UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`,
result, err := r.db.ExecContext(ctx, r.query(`UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`),
status, stage, message, nullableTime(startedAt), nullableTime(finishedAt), time.Now().UTC(), id)
if err != nil {
return fmt.Errorf("更新任务失败：%w", err)
@ -175,7 +180,7 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string,
rollback := func() {
_ = tx.Rollback()
}
result, err := tx.ExecContext(ctx, `UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), time.Now().UTC(), id)
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`), message, nullableTime(finishedAt), time.Now().UTC(), id)
if err != nil {
rollback()
return fmt.Errorf("取消任务失败：%w", err)
@ -186,7 +191,7 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string,
return fmt.Errorf("读取任务取消结果失败：%w", err)
}
if count == 0 {
item, getErr := scanTask(tx.QueryRowContext(ctx, taskSelect+` WHERE id = ?`, id))
item, getErr := scanTask(tx.QueryRowContext(ctx, r.query(taskSelect+` WHERE id = ?`), id))
if getErr != nil {
rollback()
return getErr
@ -200,7 +205,7 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string,
rollback()
return errors.New("任务当前无法取消")
}
if _, err := tx.ExecContext(ctx, `UPDATE task_attempts SET status = 'cancelled', message = ?, finished_at = ? WHERE task_id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), id); err != nil {
if _, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts SET status = 'cancelled', message = ?, finished_at = ? WHERE task_id = ? AND status IN ('queued', 'running')`), message, nullableTime(finishedAt), id); err != nil {
rollback()
return fmt.Errorf("取消任务尝试失败：%w", err)
}
@ -227,9 +232,9 @@ func (r *SQLTaskRepository) RecoverInterruptedTasks(ctx context.Context, finishe
rollback := func() {
_ = tx.Rollback()
}
result, err := tx.ExecContext(ctx, `UPDATE tasks
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks
SET status = 'failed', stage = 'service_restart', message = ?, finished_at = ?, updated_at = ?
WHERE status IN ('queued', 'running')`, "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt, finishedAt)
WHERE status IN ('queued', 'running')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt, finishedAt)
if err != nil {
rollback()
return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5931\u8d25\uff1a%w", err)
@ -239,9 +244,9 @@ func (r *SQLTaskRepository) RecoverInterruptedTasks(ctx context.Context, finishe
rollback()
return 0, fmt.Errorf("\u8bfb\u53d6\u4e2d\u65ad\u4efb\u52a1\u6570\u91cf\u5931\u8d25\uff1a%w", err)
}
if _, err := tx.ExecContext(ctx, `UPDATE task_attempts
if _, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts
SET status = 'failed', message = ?, finished_at = ?
WHERE status IN ('queued', 'running')`, "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt); err != nil {
WHERE status IN ('queued', 'running')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt); err != nil {
rollback()
return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5c1d\u8bd5\u5931\u8d25\uff1a%w", err)
}
@ -260,7 +265,7 @@ func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID
if domainID == "" || taskType == "" {
return Task{}, errors.New("域名 ID 和任务类型不能为空")
}
return scanTask(r.db.QueryRowContext(ctx, taskSelect+` WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running') ORDER BY created_at DESC LIMIT 1`, domainID, taskType))
return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running') ORDER BY created_at DESC LIMIT 1`), domainID, taskType))
}
func (r *SQLTaskRepository) FindTaskByIdempotencyKey(ctx context.Context, domainID, key string) (Task, error) {
@ -272,7 +277,7 @@ func (r *SQLTaskRepository) FindTaskByIdempotencyKey(ctx context.Context, domain
if domainID == "" || key == "" {
return Task{}, errors.New("域名 ID 和幂等键不能为空")
}
return scanTask(r.db.QueryRowContext(ctx, taskSelect+` WHERE domain_id = ? AND idempotency_key = ? ORDER BY created_at DESC LIMIT 1`, domainID, key))
return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE domain_id = ? AND idempotency_key = ? ORDER BY created_at DESC LIMIT 1`), domainID, key))
}
func (r *SQLTaskRepository) CreateAttempt(ctx context.Context, item TaskAttempt) error {
@ -292,9 +297,9 @@ func (r *SQLTaskRepository) CreateAttempt(ctx context.Context, item TaskAttempt)
if item.CreatedAt.IsZero() {
item.CreatedAt = time.Now().UTC()
}
_, err := r.db.ExecContext(ctx, `INSERT INTO task_attempts
_, err := r.db.ExecContext(ctx, r.query(`INSERT INTO task_attempts
(id, task_id, attempt_number, stage, status, message, started_at, finished_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, item.ID, item.TaskID, item.AttemptNumber, item.Stage, item.Status, item.Message,
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), item.ID, item.TaskID, item.AttemptNumber, item.Stage, item.Status, item.Message,
nullableTime(item.StartedAt), nullableTime(item.FinishedAt), item.CreatedAt.UTC())
if err != nil {
return fmt.Errorf("创建任务尝试失败：%w", err)
@ -311,7 +316,7 @@ func (r *SQLTaskRepository) FinishAttempt(ctx context.Context, id, status, messa
if id == "" || !validTaskStatus(status) {
return errors.New("任务尝试 ID 或状态无效")
}
result, err := r.db.ExecContext(ctx, `UPDATE task_attempts SET status = ?, message = ?, finished_at = ? WHERE id = ? AND (status IN ('queued', 'running') OR status = ?)`, status, message, nullableTime(finishedAt), id, status)
result, err := r.db.ExecContext(ctx, r.query(`UPDATE task_attempts SET status = ?, message = ?, finished_at = ? WHERE id = ? AND (status IN ('queued', 'running') OR status = ?)`), status, message, nullableTime(finishedAt), id, status)
if err != nil {
return fmt.Errorf("完成任务尝试失败：%w", err)
}
@ -321,7 +326,7 @@ func (r *SQLTaskRepository) FinishAttempt(ctx context.Context, id, status, messa
}
if count == 0 {
var currentStatus string
if err := r.db.QueryRowContext(ctx, `SELECT status FROM task_attempts WHERE id = ?`, id).Scan(&currentStatus); err != nil {
if err := r.db.QueryRowContext(ctx, r.query(`SELECT status FROM task_attempts WHERE id = ?`), id).Scan(&currentStatus); err != nil {
return err
}
if currentStatus == "cancelled" {
@ -340,7 +345,7 @@ func (r *SQLTaskRepository) ListAttempts(ctx context.Context, taskID string) ([]
if taskID == "" {
return nil, errors.New("任务 ID 不能为空")
}
rows, err := r.db.QueryContext(ctx, `SELECT id, task_id, attempt_number, stage, status, message, started_at, finished_at, created_at FROM task_attempts WHERE task_id = ? ORDER BY attempt_number`, taskID)
rows, err := r.db.QueryContext(ctx, r.query(`SELECT id, task_id, attempt_number, stage, status, message, started_at, finished_at, created_at FROM task_attempts WHERE task_id = ? ORDER BY attempt_number`), taskID)
if err != nil {
return nil, fmt.Errorf("查询任务尝试失败：%w", err)
}