package backup import ( "context" "os" "sort" "strings" "testing" "time" "github.com/itworx/pulse/internal/database" "github.com/jackc/pgx/v5/pgxpool" ) func TestPostgreSQLBackupRestoreCleanRoom(t *testing.T) { sourceDSN := os.Getenv("PULSE_TEST_DATABASE_URL") targetDSN := os.Getenv("PULSE_TEST_RESTORE_DATABASE_URL") if sourceDSN == "" || targetDSN == "" { if os.Getenv("PULSE_REQUIRE_BACKUP_INTEGRATION") == "true" { t.Fatal("backup integration is required but both PostgreSQL DSNs are not configured") } t.Skip("PULSE_TEST_DATABASE_URL and PULSE_TEST_RESTORE_DATABASE_URL are required") } ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() source, err := database.NewPool(ctx, database.Config{URL: sourceDSN}) if err != nil { t.Fatal(err) } defer source.Close() target, err := database.NewPool(ctx, database.Config{URL: targetDSN}) if err != nil { t.Fatal(err) } defer target.Close() if err := database.Migrate(ctx, source); err != nil { t.Fatal(err) } if err := database.Migrate(ctx, target); err != nil { t.Fatal(err) } assertBackupSchemaCoverage(t, ctx, source) ids := testIDs() now := time.Now().UTC().Truncate(time.Microsecond) seed := []struct { query string args []any }{ {`INSERT INTO roles (id,name) VALUES ($1,'administrator')`, []any{ids.role}}, {`INSERT INTO users (id,external_subject,display_name,email) VALUES ($1,$2,'Backup Test','backup@example.invalid')`, []any{ids.user, ids.user}}, {`INSERT INTO user_roles (user_id,role_id) VALUES ($1,$2)`, []any{ids.user, ids.role}}, {`INSERT INTO data_sources (id,type,name,configuration_ref) VALUES ($1,'exporter','Backup source','source/ref')`, []any{ids.source}}, {`INSERT INTO entities (id,entity_type,canonical_name,display_name,first_seen_at) VALUES ($1,'container','backup-test','Backup test',$2)`, []any{ids.entity, now}}, {`INSERT INTO container_aliases (source_id,runtime_id,entity_id,name,project,service,image_digest,observed_state,observed_health,restart_count,first_seen_at,last_seen_at) VALUES ($1,'runtime-backup-test',$2,'backup-test','pulse','api','sha256:test','running','healthy',2,$3,$3)`, []any{ids.source, ids.entity, now}}, {`INSERT INTO dashboards (id,slug,name,description,owner_user_id,scope,current_version_id) VALUES ($1,'backup-test','Backup test dashboard','portable',$2,'shared',NULL)`, []any{ids.dashboard, ids.user}}, {`INSERT INTO dashboard_versions (id,dashboard_id,version_number,schema_version,document,created_by) VALUES ($1,$2,1,1,'{}',$3)`, []any{ids.dashboardVersion, ids.dashboard, ids.user}}, {`UPDATE dashboards SET current_version_id=$1 WHERE id=$2`, []any{ids.dashboardVersion, ids.dashboard}}, {`INSERT INTO alert_rules (id,schema_version,name,severity,condition,evaluation_interval_seconds,pending_seconds,resolve_seconds,unknown_behavior,group_by,suppress_when,message,current_version_id,created_by) VALUES ($1,1,'Backup test rule','critical','{}',60,0,0,'become-unknown','[]','[]','{}',NULL,$2)`, []any{ids.rule, ids.user}}, {`INSERT INTO alert_rule_versions (id,rule_id,version_number,document,created_by) VALUES ($1,$2,1,'{}',$3)`, []any{ids.ruleVersion, ids.rule, ids.user}}, {`UPDATE alert_rules SET current_version_id=$1 WHERE id=$2`, []any{ids.ruleVersion, ids.rule}}, {`INSERT INTO alert_instances (id,rule_id,rule_version_id,fingerprint,entity_id,last_evaluated_at) VALUES ($1,$2,$3,'backup-fingerprint',$4,$5)`, []any{ids.instance, ids.rule, ids.ruleVersion, ids.entity, now}}, {`INSERT INTO incidents (id,correlation_key,title,severity,started_at,owner_user_id,correlation_method,confidence) VALUES ($1,'backup-correlation','Backup test incident','critical',$2,$3,'deterministic',0.900)`, []any{ids.incident, now, ids.user}}, {`INSERT INTO incident_alerts (incident_id,alert_id,rationale,confidence,correlation_method,added_by) VALUES ($1,$2,'backup test rationale',0.900,'deterministic','test')`, []any{ids.incident, ids.instance}}, {`INSERT INTO incident_entities (incident_id,entity_id,rationale,confidence) VALUES ($1,$2,'backup test entity',0.900)`, []any{ids.incident, ids.entity}}, {`INSERT INTO incident_notes (id,incident_id,author,body) VALUES ($1,$2,'test','backup note')`, []any{ids.note, ids.incident}}, {`INSERT INTO audit_events (id,actor,action,resource_type,resource_id,result,after_diff) VALUES ($1,'backup-test','backup.seed','dashboard',$2,'success','{}')`, []any{ids.audit, ids.dashboard}}, } for _, statement := range seed { if _, err := source.Exec(ctx, statement.query, statement.args...); err != nil { t.Fatal(err) } } directory := t.TempDir() manager := Manager{Pool: source, Directory: directory, Retention: 2, Now: func() time.Time { return now }} created, err := manager.Create(ctx) if err != nil { t.Fatal(err) } if created.Rows <= 0 || created.SHA256 == "" { t.Fatalf("unexpected backup result: %#v", created) } verified, err := manager.Verify(ctx, created.Path) if err != nil { t.Fatal(err) } if verified.FormatVersion != formatVersion { t.Fatalf("backup format = %d, want %d", verified.FormatVersion, formatVersion) } containerAliasManifest := false for _, table := range verified.Tables { if table.Name == "container_aliases" && table.Rows == 1 && table.SHA256 != "" { containerAliasManifest = true } } if !containerAliasManifest { t.Fatal("container_aliases is missing from the checksummed manifest") } if _, err := manager.List(ctx); err != nil { t.Fatal(err) } restored, err := (Manager{Pool: target}).Restore(ctx, created.Path) if err != nil { t.Fatal(err) } if restored.BackupID != created.BackupID { t.Fatalf("restored manifest = %s, want %s", restored.BackupID, created.BackupID) } for _, check := range []struct { table string want int }{ {"container_aliases", 1}, {"dashboards", 1}, {"dashboard_versions", 1}, {"alert_rules", 1}, {"alert_rule_versions", 1}, {"incidents", 1}, {"incident_alerts", 1}, {"incident_entities", 1}, {"incident_notes", 1}, {"audit_events", 1}, } { var got int if err := target.QueryRow(ctx, "SELECT count(*) FROM "+check.table).Scan(&got); err != nil { t.Fatal(err) } if got != check.want { t.Fatalf("%s count = %d, want %d", check.table, got, check.want) } } var dashboardVersion, ruleVersion string if err := target.QueryRow(ctx, `SELECT current_version_id::text FROM dashboards WHERE id=$1`, ids.dashboard).Scan(&dashboardVersion); err != nil { t.Fatal(err) } if err := target.QueryRow(ctx, `SELECT current_version_id::text FROM alert_rules WHERE id=$1`, ids.rule).Scan(&ruleVersion); err != nil { t.Fatal(err) } if dashboardVersion != ids.dashboardVersion || ruleVersion != ids.ruleVersion { t.Fatalf("deferred links = %s/%s", dashboardVersion, ruleVersion) } var runtimeEntity string if err := target.QueryRow(ctx, `SELECT entity_id::text FROM container_aliases WHERE source_id=$1 AND runtime_id='runtime-backup-test'`, ids.source).Scan(&runtimeEntity); err != nil { t.Fatal(err) } if runtimeEntity != ids.entity { t.Fatalf("restored container alias entity = %s, want %s", runtimeEntity, ids.entity) } } func assertBackupSchemaCoverage(t *testing.T, ctx context.Context, pool *pgxpool.Pool) { t.Helper() rows, err := pool.Query(ctx, `SELECT c.table_name, c.column_name FROM information_schema.columns c JOIN information_schema.tables t ON t.table_schema = c.table_schema AND t.table_name = c.table_name WHERE c.table_schema = 'public' AND t.table_type = 'BASE TABLE' ORDER BY c.table_name, c.ordinal_position`) if err != nil { t.Fatal(err) } defer rows.Close() columnsByTable := map[string]map[string]bool{} for rows.Next() { var table, column string if err := rows.Scan(&table, &column); err != nil { t.Fatal(err) } if columnsByTable[table] == nil { columnsByTable[table] = map[string]bool{} } columnsByTable[table][column] = true } if err := rows.Err(); err != nil { t.Fatal(err) } classified := map[string]string{"schema_migrations": "migration metadata"} for name, reason := range backupExcludedTables { classified[name] = reason } for _, spec := range tableSpecs { if previous, duplicate := classified[spec.name]; duplicate { t.Fatalf("backup table %q is classified more than once (previous: %s)", spec.name, previous) } classified[spec.name] = "portable backup" tableColumns, exists := columnsByTable[spec.name] if !exists { t.Fatalf("backup table %q does not exist after migrations", spec.name) } for _, list := range []string{spec.columns, spec.restoreCols} { for _, column := range strings.Split(list, ",") { column = strings.TrimSpace(column) if column == "" || !tableColumns[column] { t.Fatalf("backup table %q references missing column %q", spec.name, column) } } } } var unclassified, missing []string for table := range columnsByTable { if _, ok := classified[table]; !ok { unclassified = append(unclassified, table) } } for table := range classified { if _, ok := columnsByTable[table]; !ok { missing = append(missing, table) } } sort.Strings(unclassified) sort.Strings(missing) if len(unclassified) > 0 || len(missing) > 0 { t.Fatalf("backup schema drift: unclassified=%v missing=%v", unclassified, missing) } } type testIDSet struct { role, user, source, entity, dashboard, dashboardVersion, rule, ruleVersion, instance, incident, note, audit string } func testIDs() testIDSet { id := func() string { value, err := newID() if err != nil { panic(err) } return value } return testIDSet{ role: id(), user: id(), source: id(), entity: id(), dashboard: id(), dashboardVersion: id(), rule: id(), ruleVersion: id(), instance: id(), incident: id(), note: id(), audit: id(), } }