nexarch/internal/parallelrun/parallelrun_test.go

package parallelrun

import (
	"context"
	"errors"
	"os"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/internal/cutoverorchestrator"
	"gitea.perlbach24.de/scripte/nexarch/internal/testdb"
)

const testTenantID int64 = 202

func testCluster(t *testing.T) *cutoverorchestrator.Cluster {
	t.Helper()
	adminDSN := os.Getenv("TEST_ADMIN_DSN")
	if adminDSN == "" {
		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	c, err := cutoverorchestrator.NewCluster(adminDSN)
	if err != nil {
		t.Fatalf("NewCluster: %v", err)
	}
	return c
}

func setupSchema(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	if _, err := pool.Exec(ctx, `
		CREATE TABLE parallel_run_period (
			id          BIGSERIAL PRIMARY KEY,
			tenant_id   BIGINT NOT NULL,
			db_name     TEXT NOT NULL,
			cutover_at  TIMESTAMPTZ NOT NULL,
			deadline    TIMESTAMPTZ NOT NULL CHECK (deadline > cutover_at),
			created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE parallel_run_discrepancy (
			id            BIGSERIAL PRIMARY KEY,
			period_id     BIGINT NOT NULL REFERENCES parallel_run_period(id),
			reported_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
			description   TEXT NOT NULL,
			status        TEXT NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'resolved', 'deferred')),
			resolved_at   TIMESTAMPTZ,
			resolved_note TEXT
		);
		CREATE TABLE parallel_run_shutdown_approval (
			id          BIGSERIAL PRIMARY KEY,
			period_id   BIGINT NOT NULL REFERENCES parallel_run_period(id),
			approved_by TEXT NOT NULL,
			approved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			note        TEXT
		);
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}
}

func testStore(t *testing.T) *Store {
	t.Helper()
	adminDSN := os.Getenv("TEST_ADMIN_DSN")
	if adminDSN == "" {
		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	pool := testdb.Pool(t, adminDSN)
	setupSchema(t, context.Background(), pool)
	return NewStore(pool)
}

func setupLegacyDB(t *testing.T, cluster *cutoverorchestrator.Cluster) string {
	t.Helper()
	ctx := context.Background()
	dbName := "cut06_" + testdb.Suffix()
	if err := cluster.CreateDatabase(ctx, dbName); err != nil {
		t.Fatalf("CreateDatabase: %v", err)
	}
	t.Cleanup(func() {
		conn, err := pgx.Connect(context.Background(), cluster.AdminDSN())
		if err != nil {
			t.Logf("aufraeumen %s: verbindung fehlgeschlagen: %v", dbName, err)
			return
		}
		defer conn.Close(context.Background())
		_, _ = conn.Exec(context.Background(), "ALTER DATABASE "+pgx.Identifier{dbName}.Sanitize()+" WITH ALLOW_CONNECTIONS true")
		_, _ = conn.Exec(context.Background(), `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1`, dbName)
		_, err = conn.Exec(context.Background(), "DROP DATABASE IF EXISTS "+pgx.Identifier{dbName}.Sanitize()+" WITH (FORCE)")
		if err != nil {
			t.Logf("aufraeumen %s fehlgeschlagen: %v", dbName, err)
		}
	})
	pool, err := cluster.Connect(ctx, dbName)
	if err != nil {
		t.Fatalf("Connect: %v", err)
	}
	if _, err := pool.Exec(ctx, `CREATE TABLE legacy_writeprobe (id BIGSERIAL PRIMARY KEY, note TEXT NOT NULL)`); err != nil {
		pool.Close()
		t.Fatalf("legacy_writeprobe anlegen: %v", err)
	}
	pool.Close()
	return dbName
}

func connectLegacyDB(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string) (*pgxpool.Pool, error) {
	pool, err := cluster.Connect(ctx, dbName)
	if err != nil {
		return nil, err
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, err
	}
	return pool, nil
}

// AK1 (Pruefung 1): Schreibzugriff auf das Altsystem waehrend des
// Parallelbetriebs wird nachweislich abgewiesen, Lesezugriff bleibt moeglich.
func TestStartPeriod_LegacySystemReadOnlyNotWritable(t *testing.T) {
	cluster := testCluster(t)
	store := testStore(t)
	ctx := context.Background()
	dbName := setupLegacyDB(t, cluster)

	cutoverAt := time.Now().Add(-time.Hour)
	_, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
	if err != nil {
		t.Fatalf("StartPeriod: %v", err)
	}

	pool, err := connectLegacyDB(ctx, cluster, dbName)
	if err != nil {
		t.Fatalf("Connect nach StartPeriod: %v", err)
	}
	defer pool.Close()

	var n int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM legacy_writeprobe`).Scan(&n); err != nil {
		t.Fatalf("lesezugriff haette funktionieren muessen: %v", err)
	}

	_, err = pool.Exec(ctx, `INSERT INTO legacy_writeprobe (note) VALUES ('sollte scheitern')`)
	if err == nil {
		t.Fatal("schreibzugriff waehrend parallelbetrieb haette abgewiesen werden muessen")
	}
}

// AK2 (Pruefung 2): Abschaltung vor Fristende wird real verweigert, auch
// mit einer vorliegenden (zu frueh erteilten) Freigabe nicht moeglich.
func TestShutdown_RefusedBeforeDeadline(t *testing.T) {
	cluster := testCluster(t)
	store := testStore(t)
	ctx := context.Background()
	dbName := setupLegacyDB(t, cluster)

	cutoverAt := time.Now()
	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
	if err != nil {
		t.Fatalf("StartPeriod: %v", err)
	}

	_, err = store.ApproveShutdown(ctx, periodID, "qa-leitung", "vorzeitiger versuch", time.Now())
	if !errors.Is(err, ErrDeadlineNotReached) {
		t.Fatalf("erwarte ErrDeadlineNotReached bei freigabe vor fristende, habe %v", err)
	}

	err = Shutdown(ctx, store, cluster, periodID, time.Now())
	if !errors.Is(err, ErrDeadlineNotReached) {
		t.Fatalf("erwarte ErrDeadlineNotReached bei abschaltung vor fristende, habe %v", err)
	}

	pool, err := connectLegacyDB(ctx, cluster, dbName)
	if err != nil {
		t.Fatalf("altsystem haette nach verweigerter abschaltung weiter lesbar sein muessen: %v", err)
	}
	pool.Close()
}

// AK2/AK3 (Pruefung 2): Abschaltung nach Fristende wird ohne zweite
// Freigabe verweigert, und solange offene (unbearbeitete) Diskrepanzen
// vorliegen, ist die zweite Freigabe selbst nicht moeglich.
func TestApproveShutdown_RefusedWithOpenDiscrepancies(t *testing.T) {
	cluster := testCluster(t)
	store := testStore(t)
	ctx := context.Background()
	dbName := setupLegacyDB(t, cluster)

	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
	if err != nil {
		t.Fatalf("StartPeriod: %v", err)
	}

	discrepancyID, err := store.ReportDiscrepancy(ctx, periodID, "zaehldifferenz tenant 202 in migrationsbericht")
	if err != nil {
		t.Fatalf("ReportDiscrepancy: %v", err)
	}

	after := time.Now()
	_, err = store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after)
	if !errors.Is(err, ErrOpenDiscrepancies) {
		t.Fatalf("erwarte ErrOpenDiscrepancies bei offener diskrepanz, habe %v", err)
	}

	err = Shutdown(ctx, store, cluster, periodID, after)
	if !errors.Is(err, ErrNoApproval) {
		t.Fatalf("erwarte ErrNoApproval ohne vorliegende freigabe, habe %v", err)
	}

	if err := store.ResolveDiscrepancy(ctx, discrepancyID, StatusDeferred, "bewusst zurueckgestellt, kein datenverlust"); err != nil {
		t.Fatalf("ResolveDiscrepancy: %v", err)
	}

	approvalID, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe nach rueckstellung", after)
	if err != nil {
		t.Fatalf("ApproveShutdown nach rueckstellung: %v", err)
	}
	if approvalID == 0 {
		t.Fatal("erwarte gueltige freigabe-id")
	}
}

// AK2 (Pruefung 2): Mit Fristende und schriftlich vorliegender zweiter
// Freigabe sperrt Shutdown das Altsystem vollstaendig -- weder Lese- noch
// Schreibzugriff mehr moeglich.
func TestShutdown_LocksSystemCompletelyAfterApproval(t *testing.T) {
	cluster := testCluster(t)
	store := testStore(t)
	ctx := context.Background()
	dbName := setupLegacyDB(t, cluster)

	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
	if err != nil {
		t.Fatalf("StartPeriod: %v", err)
	}

	after := time.Now()
	if _, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after); err != nil {
		t.Fatalf("ApproveShutdown: %v", err)
	}

	if err := Shutdown(ctx, store, cluster, periodID, after); err != nil {
		t.Fatalf("Shutdown: %v", err)
	}

	_, err = connectLegacyDB(ctx, cluster, dbName)
	if err == nil {
		t.Fatal("verbindung nach abschaltung haette fehlschlagen muessen")
	}
}

func TestShutdown_RefusedWhenDiscrepancyReportedAfterApproval(t *testing.T) {
	cluster := testCluster(t)
	store := testStore(t)
	ctx := context.Background()
	dbName := setupLegacyDB(t, cluster)

	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
	if err != nil {
		t.Fatalf("StartPeriod: %v", err)
	}

	after := time.Now()
	if _, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after); err != nil {
		t.Fatalf("ApproveShutdown: %v", err)
	}

	if _, err := store.ReportDiscrepancy(ctx, periodID, "nach freigabe gemeldet"); err != nil {
		t.Fatalf("ReportDiscrepancy: %v", err)
	}

	if err := Shutdown(ctx, store, cluster, periodID, after); !errors.Is(err, ErrOpenDiscrepancies) {
		t.Fatalf("Shutdown nach einer erst nach der Freigabe gemeldeten Diskrepanz haette ErrOpenDiscrepancies liefern muessen, habe %v", err)
	}
}