ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
303 lines
12 KiB
Go
303 lines
12 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestMigrationsAreEmbeddedAndOrdered(t *testing.T) {
|
|
entries, err := migrationFiles.ReadDir("migrations")
|
|
if err != nil {
|
|
t.Fatalf("read embedded migrations: %v", err)
|
|
}
|
|
expected := []string{
|
|
"0001_foundation.sql", "0002_inventory.sql", "0003_dashboard_immutability.sql", "0004_dashboard_revision.sql",
|
|
"0005_services_probes.sql", "0006_alert_rules.sql", "0007_alert_evaluator_leases.sql", "0008_alert_state.sql",
|
|
"0009_alert_hysteresis.sql", "0010_alert_controls.sql", "0011_alert_unacknowledge.sql", "0012_notifications.sql",
|
|
"0013_incidents.sql", "0014_incident_notes.sql", "0015_entity_listing_index.sql", "0016_agent_snapshots.sql",
|
|
"0017_worker_runtime.sql", "0018_inventory_read_indexes.sql", "0019_capacity_samples.sql",
|
|
"0020_service_certificate_history_index.sql",
|
|
}
|
|
if len(entries) != len(expected) {
|
|
t.Fatalf("migration count = %d, want %d: %#v", len(entries), len(expected), entries)
|
|
}
|
|
for index, name := range expected {
|
|
if entries[index].Name() != name {
|
|
t.Fatalf("migration %d = %q, want %q", index, entries[index].Name(), name)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestServiceCertificateHistoryMigrationMatchesStatusQuery(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0020_service_certificate_history_index.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"service_certificates_service_history_idx", "service_id", "observed_at DESC", "id ASC"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("service certificate history migration is missing %q", fragment)
|
|
}
|
|
}
|
|
if strings.Contains(sql, "CONCURRENTLY") {
|
|
t.Fatal("migrations run inside a transaction and cannot create indexes concurrently")
|
|
}
|
|
}
|
|
|
|
func TestWorkerRuntimeMigrationContainsAliasAndJobLookupSafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0017_worker_runtime.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE TABLE container_aliases", "PRIMARY KEY (source_id, runtime_id)", "tombstoned_at", "ON DELETE CASCADE", "container_aliases_active_idx", "CREATE INDEX job_runs_recent_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("worker runtime migration is missing %q", fragment)
|
|
}
|
|
}
|
|
if strings.Contains(sql, "CONCURRENTLY") {
|
|
t.Fatal("migrations run inside a transaction and cannot create indexes concurrently")
|
|
}
|
|
}
|
|
|
|
func TestAlertControlsMigrationContainsExpiryAndIndexes(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0010_alert_controls.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE TABLE alert_silences", "CREATE TABLE maintenance_windows", "expires_at", "ends_at", "status", "alert_silences_active_expiry_idx", "maintenance_windows_active_expiry_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("control migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestNotificationsMigrationContainsOutboxAndAuditConstraints(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0012_notifications.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE TABLE notification_channels", "CREATE TABLE notification_outbox", "CREATE TABLE notification_deliveries", "UNIQUE (outbox_id, attempt)", "ON DELETE RESTRICT", "notification_outbox_due_idx", "notification_deliveries_history_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("notification migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestIncidentsMigrationContainsCorrelationAndAssociationSafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0013_incidents.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE TABLE incidents", "incidents_active_correlation_key_uq", "CREATE TABLE incident_alerts", "CREATE TABLE incident_entities", "ON DELETE RESTRICT", "confidence", "rationale"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("incident migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestIncidentNotesMigrationContainsBoundedNotes(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0014_incident_notes.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE TABLE incident_notes", "incident_notes_history_idx", "ON DELETE CASCADE", "char_length(body) BETWEEN 1 AND 2000"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("incident notes migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestEntityListingMigrationIndexesKeysetPagination(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0015_entity_listing_index.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"CREATE INDEX IF NOT EXISTS entities_canonical_name_idx", "ON entities (canonical_name ASC, id ASC)"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("entity listing migration is missing %q", fragment)
|
|
}
|
|
}
|
|
if strings.Contains(sql, "CONCURRENTLY") {
|
|
t.Fatal("migrations run inside a transaction and cannot create indexes concurrently")
|
|
}
|
|
}
|
|
func TestPostgreSQLMigrationsAreRestartSafe(t *testing.T) {
|
|
dsn := os.Getenv("PULSE_TEST_DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("PULSE_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
pool, err := NewPool(ctx, Config{URL: dsn})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pool.Close()
|
|
if err := Ping(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := Migrate(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := Migrate(ctx, pool); err != nil {
|
|
t.Fatalf("repeated migration: %v", err)
|
|
}
|
|
var count int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations WHERE id = '0001_foundation'`).Scan(&count); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if count != 1 {
|
|
t.Fatalf("migration count = %d, want 1", count)
|
|
}
|
|
var inventoryCount int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations WHERE id = '0002_inventory'`).Scan(&inventoryCount); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if inventoryCount != 1 {
|
|
t.Fatalf("inventory migration count = %d, want 1", inventoryCount)
|
|
}
|
|
if _, err := pool.Exec(ctx, `INSERT INTO system_settings (key, value) VALUES ('test.persistence', '{"ok":true}') ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var value bool
|
|
if err := pool.QueryRow(ctx, `SELECT value->>'ok' = 'true' FROM system_settings WHERE key = 'test.persistence'`).Scan(&value); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !value {
|
|
t.Fatal("persisted setting was not retained")
|
|
}
|
|
pool.Close()
|
|
restartedPool, err := NewPool(ctx, Config{URL: dsn})
|
|
if err != nil {
|
|
t.Fatalf("reopen database pool: %v", err)
|
|
}
|
|
defer restartedPool.Close()
|
|
var afterRestart bool
|
|
if err := restartedPool.QueryRow(ctx, `SELECT value->>'ok' = 'true' FROM system_settings WHERE key = 'test.persistence'`).Scan(&afterRestart); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !afterRestart {
|
|
t.Fatal("persisted setting was not retained after pool restart")
|
|
}
|
|
}
|
|
|
|
func TestAlertRuleMigrationContainsVersionSafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0006_alert_rules.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, table := range []string{"alert_rules", "alert_rule_versions"} {
|
|
if !strings.Contains(sql, "CREATE TABLE "+table) {
|
|
t.Fatalf("migration is missing table %s", table)
|
|
}
|
|
}
|
|
for _, constraint := range []string{"UNIQUE (rule_id, version_number)", "ON DELETE RESTRICT", "DEFERRABLE INITIALLY DEFERRED", "WHERE enabled = true"} {
|
|
if !strings.Contains(sql, constraint) {
|
|
t.Fatalf("migration is missing safety constraint %q", constraint)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAlertEvaluatorLeaseMigrationContainsExpiryIndex(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0007_alert_evaluator_leases.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"ADD COLUMN lease_owner text", "ADD COLUMN lease_until timestamptz", "CREATE INDEX job_runs_lease_idx", "status IN ('queued', 'running')"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("lease migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAlertStateMigrationContainsLifecycleAndHistorySafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0008_alert_state.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, table := range []string{"alert_instances", "alert_occurrences"} {
|
|
if !strings.Contains(sql, "CREATE TABLE "+table) {
|
|
t.Fatalf("migration is missing table %s", table)
|
|
}
|
|
}
|
|
for _, fragment := range []string{"UNIQUE (rule_id, fingerprint)", "UNIQUE (instance_id, evaluation_key)", "ON DELETE RESTRICT", "current_state text NOT NULL", "CREATE INDEX alert_occurrences_history_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("state migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestAlertHysteresisMigrationContainsCooldownSafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0009_alert_hysteresis.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{"ADD COLUMN cooldown_seconds", "ADD COLUMN cooldown_until", "cooldown_seconds BETWEEN 0 AND 2592000", "CREATE INDEX alert_instances_cooldown_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("hysteresis migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestAlertUnacknowledgeMigrationContainsConstraintSafety(t *testing.T) {
|
|
sqlBytes, err := migrationFiles.ReadFile("migrations/0011_alert_unacknowledge.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(sqlBytes)
|
|
for _, fragment := range []string{"DROP CONSTRAINT alert_occurrences_event_type_check", "unacknowledge", "alert_instances_acknowledged_idx"} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("unacknowledge migration is missing %q", fragment)
|
|
}
|
|
}
|
|
}
|
|
func TestServiceProbeMigrationContainsHistoryAndAccessSafety(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0005_services_probes.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, table := range []string{"services", "service_endpoints", "probes", "probe_results", "service_certificates", "service_dependencies", "service_permissions"} {
|
|
if !strings.Contains(sql, "CREATE TABLE "+table) {
|
|
t.Fatalf("migration is missing table %s", table)
|
|
}
|
|
}
|
|
for _, constraint := range []string{"revision bigint NOT NULL DEFAULT 1 CHECK (revision > 0)", "ON DELETE RESTRICT", "WHERE archived_at IS NULL", "UNIQUE (probe_id, observed_at)", "permission IN ('view', 'operate', 'edit', 'admin')"} {
|
|
if !strings.Contains(sql, constraint) {
|
|
t.Fatalf("migration is missing safety constraint %q", constraint)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAgentSnapshotMigrationBoundsPayloadAndCapability(t *testing.T) {
|
|
content, err := migrationFiles.ReadFile("migrations/0016_agent_snapshots.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sql := string(content)
|
|
for _, fragment := range []string{
|
|
"CREATE TABLE agent_snapshots",
|
|
"observed_at timestamptz NOT NULL",
|
|
"received_at timestamptz NOT NULL",
|
|
"jsonb_typeof(payload) = 'object'",
|
|
"pg_column_size(payload) <= 2097152",
|
|
"capability IN ('host', 'processes', 'containers', 'array', 'disks', 'pools', 'shares')",
|
|
"PRIMARY KEY (agent_id, capability)",
|
|
"agent_snapshots_capability_freshness_idx",
|
|
} {
|
|
if !strings.Contains(sql, fragment) {
|
|
t.Fatalf("agent snapshot migration is missing %q", fragment)
|
|
}
|
|
}
|
|
if strings.Contains(sql, "CONCURRENTLY") {
|
|
t.Fatal("migrations run inside a transaction and cannot create indexes concurrently")
|
|
}
|
|
}
|