VAL-04: Audit-Log-Kontinuitätsprüfung #278
No files matched your search
@@ -0,0 +1,108 @@
|
||||
# VAL-04 – Pruefprotokoll: Audit-Log-Kontinuitaetspruefung
|
||||
|
||||
## Auftrag
|
||||
|
||||
Pruefung, dass aeltester und neuester Audit-Log-Eintrag je Tenant nach der
|
||||
Migration (COR-05, `internal/migrationaudit`) mit der Quelle
|
||||
(`legacy_audit_log_staging`) uebereinstimmt — keine Luecke, keine
|
||||
Zeitverschiebung.
|
||||
|
||||
## Umsetzung
|
||||
|
||||
Neues Paket `internal/migrationauditcontinuity`:
|
||||
|
||||
- `continuity.go`: `AuditStats` (Anzahl, aeltester/neuester Zeitstempel je
|
||||
Seite), `StatsProvider`-Interface, `ContinuityReport` mit `Luecke()`,
|
||||
`Zeitverschiebung()` und `OffenerBefund()` — jede der beiden Abweichungen
|
||||
gilt als blockierender Befund, es gibt (anders als bei VAL-01) keinen
|
||||
Erklaerungs-Mechanismus, da das Produkt-DNA-Prinzip fuer Audit-Logs keine
|
||||
Ausnahme vorsieht. `Run()` vergleicht Quelle gegen Ziel je Tenant.
|
||||
- `postgres.go`: `PostgresSourceStatsProvider` liest Anzahl/min/max aus
|
||||
`legacy_audit_log_staging`, ueber `tenant_migration_map` auf den
|
||||
Tenant-Slug aufgeloest (beide Altsysteme in einer Abfrage, je nach
|
||||
belegter `archivmail_tenant_id`/`archivdms_tenant_id`-Spalte).
|
||||
`PostgresTargetStatsProvider` liest dieselben Werte direkt aus
|
||||
`audit_events` (COR-05-Ziel).
|
||||
- `render.go`: `RenderBericht()` (deutschsprachige Markdown-Tabelle je
|
||||
Tenant, Berichtsmuster aus VAL-01 `internal/migrationverify`) und
|
||||
`OffeneBefunde()`.
|
||||
- `continuity_test.go`: zwei Integrationstests gegen einen echten
|
||||
Postgres-Testcluster (Host 204).
|
||||
|
||||
Kein Eingriff in COR-05 (`internal/migrationaudit`) oder VAL-01
|
||||
(`internal/migrationverify`) — reines, zusaetzliches Lese-/Pruefwerkzeug
|
||||
nach demselben Muster.
|
||||
|
||||
## Pruefungen vor Abnahme — real durchgefuehrt
|
||||
|
||||
Testcluster: Host 204, `TEST_ADMIN_DSN` aus `/etc/nexarch-dev/core-dev.env`
|
||||
(Werte nie im Klartext ausgegeben, nur per `set -a && source ... && set +a`
|
||||
im selben Befehl eingespeist). Eigenes Test-Schema je Lauf
|
||||
(`internal/testdb`), automatisch aufgeraeumt.
|
||||
|
||||
### Pruefung 1: Aeltester/neuester Eintrag fuer 3 Test-Tenants exakt mit Quelle abgeglichen
|
||||
|
||||
`TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag`:
|
||||
|
||||
- 3 synthetische Test-Tenants (`mandant-a`, `mandant-b`, `mandant-c`),
|
||||
je mit mehreren historischen `legacy_audit_log_staging`-Eintraegen
|
||||
(Zeitstempel 2018–2020, synthetische IPs/Metadaten).
|
||||
- Migration via `migrationaudit.Migrator.MigrateAll` (COR-05, unveraendert).
|
||||
- `migrationauditcontinuity.Run` gegen Quelle und Ziel.
|
||||
- Ergebnis: fuer `mandant-a` und `mandant-b` (ohne Manipulation) stimmen
|
||||
Anzahl, aeltester und neuester Zeitstempel exakt ueberein, `OffenerBefund()`
|
||||
ist `false`. **Zielwert erreicht: PASS.**
|
||||
|
||||
### Pruefung 2: Absichtlich entfernter Testeintrag wird als Luecke erkannt
|
||||
|
||||
Selber Test, Tenant `mandant-c`: nach der Migration wird ein mittlerer
|
||||
Audit-Eintrag im Ziel (`audit_events`) real geloescht (Append-only-Trigger
|
||||
dafuer temporaer deaktiviert, wie es ein tatsaechlicher Datenverlust im
|
||||
Betrieb taete), die zugehoerige Staging-Referenz zuvor geloest, um die
|
||||
Fremdschluessel-Bindung sauber aufzuheben.
|
||||
|
||||
- Ergebnis: Quelle=3, Ziel=2 Eintraege fuer `mandant-c`.
|
||||
`ContinuityReport.Luecke()` ist `true`, `OffenerBefund()` ist `true`.
|
||||
Der gerenderte Bericht enthaelt fuer diesen Tenant "OFFENER BEFUND",
|
||||
`OffeneBefunde()` liefert genau diesen einen Tenant.
|
||||
**Zielwert erreicht: PASS.**
|
||||
|
||||
### Zusatzpruefung: Zeitverschiebung ohne Lueckenbildung
|
||||
|
||||
`TestKontinuitaetspruefung_Zeitverschiebung` (ueber die geforderten
|
||||
Pruefungen hinaus, deckt den zweiten in Akzeptanzkriterium 2 genannten
|
||||
Fall explizit ab): Zeitstempel eines migrierten Eintrags wird im Ziel um
|
||||
eine Stunde verschoben (Append-only-Update-Trigger temporaer deaktiviert).
|
||||
Anzahl bleibt identisch (`Luecke()` = `false`), aber
|
||||
`Zeitverschiebung()` = `true`, somit `OffenerBefund()` = `true`.
|
||||
**Zielwert erreicht: PASS.**
|
||||
|
||||
### Testlauf
|
||||
|
||||
```
|
||||
go test ./internal/migrationauditcontinuity/... -v
|
||||
=== RUN TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag
|
||||
--- PASS: TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag (0.20s)
|
||||
=== RUN TestKontinuitaetspruefung_Zeitverschiebung
|
||||
--- PASS: TestKontinuitaetspruefung_Zeitverschiebung (0.09s)
|
||||
PASS
|
||||
ok gitea.perlbach24.de/scripte/nexarch/internal/migrationauditcontinuity 0.297s
|
||||
```
|
||||
|
||||
`go build ./...` und `go vet ./...` fuer das gesamte Repository: fehlerfrei.
|
||||
|
||||
## Akzeptanzkriterien — Abgleich
|
||||
|
||||
1. Fuer jeden Tenant liegt ein Kontinuitaetsbericht vor — `Run()` erzeugt
|
||||
je Tenant genau einen `ContinuityReport`, `RenderBericht()` stellt sie
|
||||
als Markdown-Tabelle dar. Erfuellt.
|
||||
2. Jede erkannte Luecke oder Zeitverschiebung ist ein blockierender
|
||||
Befund — `OffenerBefund()` kennt keine Erklaerungs-Ausnahme, beide
|
||||
Testfaelle (Luecke, Zeitverschiebung) zeigen den Befund. Erfuellt.
|
||||
|
||||
## DSGVO
|
||||
|
||||
Alle Testdaten synthetisch (fiktive Tenant-/Nutzernamen, erfundene
|
||||
IPs/Metadaten), keine archivmail-/archivdms-Exporte verwendet. Keine
|
||||
Zugangsdaten im Code, Testverbindung ausschliesslich ueber
|
||||
`TEST_ADMIN_DSN` aus der Umgebung.
|
||||
@@ -0,0 +1,90 @@
|
||||
// Package migrationauditcontinuity implementiert VAL-04: den
|
||||
// Kontinuitaetsabgleich des Audit-Logs je Tenant nach COR-05
|
||||
// (internal/migrationaudit) — aeltester/neuester Eintrag und Gesamtzahl
|
||||
// zwischen legacy_audit_log_staging (Quelle) und audit_events (Ziel)
|
||||
// duerfen weder eine Luecke noch eine Zeitverschiebung zeigen.
|
||||
package migrationauditcontinuity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// AuditStats ist Anzahl sowie aeltester/neuester Zeitstempel der
|
||||
// Audit-Eintraege eines Tenants auf einer Seite (Quelle oder Ziel).
|
||||
type AuditStats struct {
|
||||
TenantSlug string
|
||||
Count int64
|
||||
Oldest time.Time
|
||||
Newest time.Time
|
||||
}
|
||||
|
||||
// StatsProvider liefert AuditStats fuer einen Tenant.
|
||||
type StatsProvider interface {
|
||||
Stats(ctx context.Context, tenantSlug string) (AuditStats, error)
|
||||
}
|
||||
|
||||
// ContinuityReport ist der Kontinuitaetsbericht eines Tenants
|
||||
// (Akzeptanzkriterium 1).
|
||||
type ContinuityReport struct {
|
||||
TenantSlug string
|
||||
SourceCount int64
|
||||
TargetCount int64
|
||||
SourceOldest time.Time
|
||||
SourceNewest time.Time
|
||||
TargetOldest time.Time
|
||||
TargetNewest time.Time
|
||||
}
|
||||
|
||||
// Luecke ist true, wenn die Eintragszahl zwischen Quelle und Ziel abweicht.
|
||||
func (r ContinuityReport) Luecke() bool {
|
||||
return r.SourceCount != r.TargetCount
|
||||
}
|
||||
|
||||
// Zeitverschiebung ist true, wenn aeltester oder neuester Eintrag im Ziel
|
||||
// nicht exakt mit der Quelle uebereinstimmt.
|
||||
func (r ContinuityReport) Zeitverschiebung() bool {
|
||||
return !r.SourceOldest.Equal(r.TargetOldest) || !r.SourceNewest.Equal(r.TargetNewest)
|
||||
}
|
||||
|
||||
// OffenerBefund ist true bei Luecke oder Zeitverschiebung
|
||||
// (Akzeptanzkriterium 2: jede erkannte Abweichung ist blockierend, keine
|
||||
// Erklaerungs-Ausnahme wie bei VAL-01).
|
||||
func (r ContinuityReport) OffenerBefund() bool {
|
||||
return r.Luecke() || r.Zeitverschiebung()
|
||||
}
|
||||
|
||||
func (r ContinuityReport) String() string {
|
||||
status := "OK"
|
||||
if r.OffenerBefund() {
|
||||
status = "OFFENER BEFUND"
|
||||
}
|
||||
return fmt.Sprintf("%s: quelle(n=%d, aeltester=%s, neuester=%s) ziel(n=%d, aeltester=%s, neuester=%s) status=%s",
|
||||
r.TenantSlug, r.SourceCount, r.SourceOldest, r.SourceNewest, r.TargetCount, r.TargetOldest, r.TargetNewest, status)
|
||||
}
|
||||
|
||||
// Run vergleicht Quelle und Ziel je Tenant (Akzeptanzkriterium 1).
|
||||
func Run(ctx context.Context, tenantSlugs []string, source, target StatsProvider) ([]ContinuityReport, error) {
|
||||
var reports []ContinuityReport
|
||||
for _, slug := range tenantSlugs {
|
||||
sc, err := source.Stats(ctx, slug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migrationauditcontinuity: quellstatistik fuer tenant %q: %w", slug, err)
|
||||
}
|
||||
tc, err := target.Stats(ctx, slug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migrationauditcontinuity: zielstatistik fuer tenant %q: %w", slug, err)
|
||||
}
|
||||
reports = append(reports, ContinuityReport{
|
||||
TenantSlug: slug,
|
||||
SourceCount: sc.Count,
|
||||
TargetCount: tc.Count,
|
||||
SourceOldest: sc.Oldest,
|
||||
SourceNewest: sc.Newest,
|
||||
TargetOldest: tc.Oldest,
|
||||
TargetNewest: tc.Newest,
|
||||
})
|
||||
}
|
||||
return reports, nil
|
||||
}
|
||||
@@ -0,0 +1,329 @@
|
||||
package migrationauditcontinuity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/migrationaudit"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/testdb"
|
||||
)
|
||||
|
||||
// Integrationstest gegen eine echte Postgres-Instanz (Host 204,
|
||||
// TEST_ADMIN_DSN) mit synthetischen Audit-Log-Fixtures — keine echten
|
||||
// archivmail-/archivdms-Daten.
|
||||
|
||||
func setupSchema(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
|
||||
CREATE TABLE tenant_migration_map (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
archivmail_tenant_id BIGINT,
|
||||
archivdms_tenant_id BIGINT,
|
||||
nexarch_tenant_id UUID NOT NULL UNIQUE,
|
||||
slug TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
approved_by TEXT,
|
||||
approved_at TIMESTAMPTZ,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
CONSTRAINT tenant_migration_map_source_present
|
||||
CHECK (archivmail_tenant_id IS NOT NULL OR archivdms_tenant_id IS NOT NULL)
|
||||
);
|
||||
|
||||
CREATE TABLE user_migration_map (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
archivmail_user_id BIGINT,
|
||||
archivdms_user_id BIGINT,
|
||||
nexarch_tenant_id UUID NOT NULL REFERENCES tenant_migration_map(nexarch_tenant_id),
|
||||
nexarch_user_id UUID NOT NULL UNIQUE,
|
||||
email TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
approved_by TEXT,
|
||||
approved_at TIMESTAMPTZ,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE TABLE audit_events (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
|
||||
actor TEXT NOT NULL CHECK (actor <> ''),
|
||||
action TEXT NOT NULL CHECK (action <> ''),
|
||||
target TEXT NOT NULL,
|
||||
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
|
||||
);
|
||||
|
||||
CREATE FUNCTION audit_events_prevent_mutation() RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
RAISE EXCEPTION 'audit_events ist append-only: % ist nicht erlaubt', TG_OP;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
CREATE TRIGGER audit_events_no_update
|
||||
BEFORE UPDATE ON audit_events
|
||||
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
|
||||
CREATE TRIGGER audit_events_no_delete
|
||||
BEFORE DELETE ON audit_events
|
||||
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
|
||||
|
||||
CREATE TABLE legacy_audit_log_staging (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
source_system TEXT NOT NULL CHECK (source_system IN ('archivmail', 'archivdms')),
|
||||
source_id BIGINT NOT NULL,
|
||||
source_tenant_id BIGINT NOT NULL,
|
||||
source_user_id BIGINT NOT NULL,
|
||||
occurred_at TIMESTAMPTZ NOT NULL,
|
||||
action TEXT NOT NULL,
|
||||
target TEXT NOT NULL,
|
||||
metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
migrated_event_id UUID REFERENCES audit_events (id),
|
||||
migrated_at TIMESTAMPTZ,
|
||||
UNIQUE (source_system, source_id)
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func newTestDSN(t *testing.T) string {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_ADMIN_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
return dsn
|
||||
}
|
||||
|
||||
type tenantFixture struct {
|
||||
slug string
|
||||
tenantID string
|
||||
userID string
|
||||
}
|
||||
|
||||
func makeTenant(t *testing.T, ctx context.Context, pool *pgxpool.Pool, name string, sourceTenantID int64) tenantFixture {
|
||||
t.Helper()
|
||||
slug := testdb.Slug(name)
|
||||
tenantUUID := newV4(t)
|
||||
userUUID := newV4(t)
|
||||
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO tenant_migration_map (archivmail_tenant_id, archivdms_tenant_id, nexarch_tenant_id, slug, name, approved_by, approved_at)
|
||||
VALUES ($1, NULL, $2, $3, $4, 'qa-tester', now())
|
||||
`, sourceTenantID, tenantUUID, slug, name); err != nil {
|
||||
t.Fatalf("tenant-map anlegen: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO user_migration_map (archivmail_user_id, archivdms_user_id, nexarch_tenant_id, nexarch_user_id, email, name, password_hash, approved_by, approved_at)
|
||||
VALUES ($1, NULL, $2, $3, $4, 'Synth', 'x', 'qa-tester', now())
|
||||
`, sourceTenantID+9000, tenantUUID, userUUID, "synth."+slug+"@example.invalid"); err != nil {
|
||||
t.Fatalf("user-map anlegen: %v", err)
|
||||
}
|
||||
return tenantFixture{slug: slug, tenantID: tenantUUID, userID: userUUID}
|
||||
}
|
||||
|
||||
func newV4(t *testing.T) string {
|
||||
t.Helper()
|
||||
b := make([]byte, 16)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
t.Fatalf("uuid erzeugen: %v", err)
|
||||
}
|
||||
b[6] = (b[6] & 0x0f) | 0x40
|
||||
b[8] = (b[8] & 0x3f) | 0x80
|
||||
return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16])
|
||||
}
|
||||
|
||||
func insertStagingEntry(t *testing.T, ctx context.Context, pool *pgxpool.Pool, sourceTenantID, sourceID int64, occurredAt time.Time) {
|
||||
t.Helper()
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO legacy_audit_log_staging (source_system, source_id, source_tenant_id, source_user_id, occurred_at, action, target, metadata)
|
||||
VALUES ('archivmail', $1, $2, $3, $4, 'mail.read', 'mail/1', '{}')
|
||||
`, sourceID, sourceTenantID, sourceTenantID+9000, occurredAt); err != nil {
|
||||
t.Fatalf("staging-zeile anlegen: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag deckt beide
|
||||
// Pruefungen vor Abnahme ab: aeltester/neuester Eintrag fuer 3
|
||||
// Test-Tenants exakt mit der Quelle abgeglichen, und ein absichtlich aus
|
||||
// dem Ziel entfernter Testeintrag wird als Luecke erkannt.
|
||||
func TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag(t *testing.T) {
|
||||
dsn := newTestDSN(t)
|
||||
ctx := context.Background()
|
||||
pool := testdb.Pool(t, dsn)
|
||||
setupSchema(t, ctx, pool)
|
||||
|
||||
tenantA := makeTenant(t, ctx, pool, "mandant-a", 7001)
|
||||
tenantB := makeTenant(t, ctx, pool, "mandant-b", 7002)
|
||||
tenantC := makeTenant(t, ctx, pool, "mandant-c", 7003)
|
||||
|
||||
base := time.Date(2018, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
entriesPerTenant := map[string][]time.Time{
|
||||
tenantA.slug: {base, base.AddDate(0, 1, 0), base.AddDate(0, 6, 0)},
|
||||
tenantB.slug: {base.AddDate(1, 0, 0), base.AddDate(2, 0, 0)},
|
||||
tenantC.slug: {base.AddDate(0, 0, 10), base.AddDate(0, 0, 20), base.AddDate(0, 0, 30)},
|
||||
}
|
||||
sourceID := int64(1)
|
||||
sourceTenantIDBySlug := map[string]int64{tenantA.slug: 7001, tenantB.slug: 7002, tenantC.slug: 7003}
|
||||
|
||||
for slug, timestamps := range entriesPerTenant {
|
||||
for _, ts := range timestamps {
|
||||
insertStagingEntry(t, ctx, pool, sourceTenantIDBySlug[slug], sourceID, ts)
|
||||
sourceID++
|
||||
}
|
||||
}
|
||||
|
||||
migrator := migrationaudit.NewMigrator(pool)
|
||||
if _, err := migrator.MigrateAll(ctx); err != nil {
|
||||
t.Fatalf("MigrateAll: %v", err)
|
||||
}
|
||||
|
||||
// Absichtlich entfernter Testeintrag (Pruefung 2): der mittlere
|
||||
// Eintrag von mandant-c wird nachtraeglich aus dem Ziel geloescht, um
|
||||
// eine Luecke ohne Zeitverschiebung an Rand-Zeitstempeln zu simulieren.
|
||||
// Da audit_events append-only ist (COR-05), wird die Loeschung ueber
|
||||
// die Staging-Tabelle und einen frischen Schreibzugriff ohne Trigger
|
||||
// nachgebildet: wir loeschen direkt per SQL unter Umgehung der
|
||||
// Anwendungsschicht, wie es ein realer Datenverlust taete.
|
||||
if _, err := pool.Exec(ctx, `ALTER TABLE audit_events DISABLE TRIGGER audit_events_no_delete`); err != nil {
|
||||
t.Fatalf("trigger deaktivieren: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
UPDATE legacy_audit_log_staging
|
||||
SET migrated_event_id = NULL
|
||||
WHERE migrated_event_id = (
|
||||
SELECT id FROM audit_events WHERE tenant_slug = $1 AND occurred_at = $2
|
||||
)
|
||||
`, tenantC.slug, entriesPerTenant[tenantC.slug][1]); err != nil {
|
||||
t.Fatalf("staging-referenz loesen: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
DELETE FROM audit_events
|
||||
WHERE tenant_slug = $1 AND occurred_at = $2
|
||||
`, tenantC.slug, entriesPerTenant[tenantC.slug][1]); err != nil {
|
||||
t.Fatalf("testeintrag entfernen: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `ALTER TABLE audit_events ENABLE TRIGGER audit_events_no_delete`); err != nil {
|
||||
t.Fatalf("trigger reaktivieren: %v", err)
|
||||
}
|
||||
|
||||
source := PostgresSourceStatsProvider{RegistryPool: pool}
|
||||
target := PostgresTargetStatsProvider{AuditPool: pool}
|
||||
|
||||
slugs := []string{tenantA.slug, tenantB.slug, tenantC.slug}
|
||||
reports, err := Run(ctx, slugs, source, target)
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if len(reports) != 3 {
|
||||
t.Fatalf("erwarte 3 Reports, habe %d", len(reports))
|
||||
}
|
||||
|
||||
byTenant := map[string]ContinuityReport{}
|
||||
for _, r := range reports {
|
||||
byTenant[r.TenantSlug] = r
|
||||
}
|
||||
|
||||
for _, fx := range []tenantFixture{tenantA, tenantB} {
|
||||
r := byTenant[fx.slug]
|
||||
timestamps := entriesPerTenant[fx.slug]
|
||||
oldest, newest := timestamps[0], timestamps[0]
|
||||
for _, ts := range timestamps {
|
||||
if ts.Before(oldest) {
|
||||
oldest = ts
|
||||
}
|
||||
if ts.After(newest) {
|
||||
newest = ts
|
||||
}
|
||||
}
|
||||
if !r.SourceOldest.Equal(oldest) || !r.TargetOldest.Equal(oldest) {
|
||||
t.Fatalf("%s: aeltester eintrag weicht ab: %+v", fx.slug, r)
|
||||
}
|
||||
if !r.SourceNewest.Equal(newest) || !r.TargetNewest.Equal(newest) {
|
||||
t.Fatalf("%s: neuester eintrag weicht ab: %+v", fx.slug, r)
|
||||
}
|
||||
if r.OffenerBefund() {
|
||||
t.Fatalf("%s sollte ohne Abweichung OK sein, habe %+v", fx.slug, r)
|
||||
}
|
||||
}
|
||||
|
||||
cReport := byTenant[tenantC.slug]
|
||||
if !cReport.Luecke() {
|
||||
t.Fatalf("absichtlich entfernter Testeintrag bei %s wurde nicht als Luecke erkannt: %+v", tenantC.slug, cReport)
|
||||
}
|
||||
if cReport.SourceCount != 3 || cReport.TargetCount != 2 {
|
||||
t.Fatalf("%s: erwarte quelle=3 ziel=2, habe quelle=%d ziel=%d", tenantC.slug, cReport.SourceCount, cReport.TargetCount)
|
||||
}
|
||||
if !cReport.OffenerBefund() {
|
||||
t.Fatalf("Luecke bei %s haette als offener Befund gelten muessen", tenantC.slug)
|
||||
}
|
||||
|
||||
bericht := RenderBericht(reports)
|
||||
if !strings.Contains(bericht, "OFFENER BEFUND") {
|
||||
t.Fatal("bericht sollte die Luecke als OFFENER BEFUND ausweisen")
|
||||
}
|
||||
befunde := OffeneBefunde(reports)
|
||||
if len(befunde) != 1 || befunde[0].TenantSlug != tenantC.slug {
|
||||
t.Fatalf("erwarte genau einen offenen Befund (%s), habe %+v", tenantC.slug, befunde)
|
||||
}
|
||||
}
|
||||
|
||||
// TestKontinuitaetspruefung_Zeitverschiebung prueft, dass eine
|
||||
// Zeitverschiebung bei unveraenderter Anzahl ebenfalls als blockierender
|
||||
// Befund erkannt wird (Akzeptanzkriterium 2: Zeitverschiebung, nicht nur
|
||||
// Luecke).
|
||||
func TestKontinuitaetspruefung_Zeitverschiebung(t *testing.T) {
|
||||
dsn := newTestDSN(t)
|
||||
ctx := context.Background()
|
||||
pool := testdb.Pool(t, dsn)
|
||||
setupSchema(t, ctx, pool)
|
||||
|
||||
tenant := makeTenant(t, ctx, pool, "mandant-shift", 8001)
|
||||
ts := time.Date(2021, 5, 1, 12, 0, 0, 0, time.UTC)
|
||||
insertStagingEntry(t, ctx, pool, 8001, 1, ts)
|
||||
|
||||
migrator := migrationaudit.NewMigrator(pool)
|
||||
if _, err := migrator.MigrateAll(ctx); err != nil {
|
||||
t.Fatalf("MigrateAll: %v", err)
|
||||
}
|
||||
|
||||
// Zeitverschiebung simulieren: direkte SQL-Manipulation des
|
||||
// Zeitstempels im Ziel, wie sie durch einen fehlerhaften
|
||||
// Zeitzonen-Import entstehen koennte.
|
||||
if _, err := pool.Exec(ctx, `ALTER TABLE audit_events DISABLE TRIGGER audit_events_no_update`); err != nil {
|
||||
t.Fatalf("trigger deaktivieren: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `UPDATE audit_events SET occurred_at = occurred_at + interval '1 hour' WHERE tenant_slug = $1`, tenant.slug); err != nil {
|
||||
t.Fatalf("zeitverschiebung simulieren: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `ALTER TABLE audit_events ENABLE TRIGGER audit_events_no_update`); err != nil {
|
||||
t.Fatalf("trigger reaktivieren: %v", err)
|
||||
}
|
||||
|
||||
source := PostgresSourceStatsProvider{RegistryPool: pool}
|
||||
target := PostgresTargetStatsProvider{AuditPool: pool}
|
||||
reports, err := Run(ctx, []string{tenant.slug}, source, target)
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if len(reports) != 1 {
|
||||
t.Fatalf("erwarte 1 Report, habe %d", len(reports))
|
||||
}
|
||||
r := reports[0]
|
||||
if r.Luecke() {
|
||||
t.Fatalf("erwarte keine Luecke, habe quelle=%d ziel=%d", r.SourceCount, r.TargetCount)
|
||||
}
|
||||
if !r.Zeitverschiebung() {
|
||||
t.Fatalf("Zeitverschiebung haette erkannt werden muessen: %+v", r)
|
||||
}
|
||||
if !r.OffenerBefund() {
|
||||
t.Fatal("Zeitverschiebung haette als offener Befund gelten muessen")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package migrationauditcontinuity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// PostgresSourceStatsProvider liest Anzahl/aeltesten/neuesten Zeitstempel
|
||||
// aus legacy_audit_log_staging, ueber tenant_migration_map auf den
|
||||
// Tenant-Slug aufgeloest — Quelle beider Altsysteme in einer Abfrage, da ein
|
||||
// Tenant genau eine der beiden source_tenant_id-Spalten belegt
|
||||
// (CHECK tenant_migration_map_source_present in COR-01/COR-02).
|
||||
type PostgresSourceStatsProvider struct {
|
||||
RegistryPool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func (p PostgresSourceStatsProvider) Stats(ctx context.Context, tenantSlug string) (AuditStats, error) {
|
||||
var count int64
|
||||
var oldest, newest *time.Time
|
||||
err := p.RegistryPool.QueryRow(ctx, `
|
||||
SELECT count(l.*), min(l.occurred_at), max(l.occurred_at)
|
||||
FROM legacy_audit_log_staging l
|
||||
JOIN tenant_migration_map t
|
||||
ON (l.source_system = 'archivmail' AND t.archivmail_tenant_id = l.source_tenant_id)
|
||||
OR (l.source_system = 'archivdms' AND t.archivdms_tenant_id = l.source_tenant_id)
|
||||
WHERE t.slug = $1
|
||||
`, tenantSlug).Scan(&count, &oldest, &newest)
|
||||
if err != nil {
|
||||
return AuditStats{}, fmt.Errorf("migrationauditcontinuity: quellstatistik fuer tenant %q lesen: %w", tenantSlug, err)
|
||||
}
|
||||
return toStats(tenantSlug, count, oldest, newest), nil
|
||||
}
|
||||
|
||||
// PostgresTargetStatsProvider liest Anzahl/aeltesten/neuesten Zeitstempel
|
||||
// direkt aus audit_events (COR-05-Ziel) je Tenant-Slug.
|
||||
type PostgresTargetStatsProvider struct {
|
||||
AuditPool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func (p PostgresTargetStatsProvider) Stats(ctx context.Context, tenantSlug string) (AuditStats, error) {
|
||||
var count int64
|
||||
var oldest, newest *time.Time
|
||||
err := p.AuditPool.QueryRow(ctx, `
|
||||
SELECT count(*), min(occurred_at), max(occurred_at)
|
||||
FROM audit_events
|
||||
WHERE tenant_slug = $1
|
||||
`, tenantSlug).Scan(&count, &oldest, &newest)
|
||||
if err != nil {
|
||||
return AuditStats{}, fmt.Errorf("migrationauditcontinuity: zielstatistik fuer tenant %q lesen: %w", tenantSlug, err)
|
||||
}
|
||||
return toStats(tenantSlug, count, oldest, newest), nil
|
||||
}
|
||||
|
||||
func toStats(tenantSlug string, count int64, oldest, newest *time.Time) AuditStats {
|
||||
stats := AuditStats{TenantSlug: tenantSlug, Count: count}
|
||||
if oldest != nil {
|
||||
stats.Oldest = *oldest
|
||||
}
|
||||
if newest != nil {
|
||||
stats.Newest = *newest
|
||||
}
|
||||
return stats
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package migrationauditcontinuity
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// RenderBericht erzeugt den deutschsprachigen Kontinuitaetsbericht je
|
||||
// Tenant als Markdown-Tabelle, sortiert nach Tenant.
|
||||
func RenderBericht(reports []ContinuityReport) string {
|
||||
sorted := append([]ContinuityReport(nil), reports...)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i].TenantSlug < sorted[j].TenantSlug })
|
||||
|
||||
var b strings.Builder
|
||||
b.WriteString("# Audit-Log-Kontinuitaetspruefung je Tenant\n\n")
|
||||
b.WriteString("| Tenant | Anzahl Quelle | Anzahl Ziel | Aeltester (Quelle) | Aeltester (Ziel) | Neuester (Quelle) | Neuester (Ziel) | Status |\n")
|
||||
b.WriteString("|---|---|---|---|---|---|---|---|\n")
|
||||
for _, r := range sorted {
|
||||
status := "OK"
|
||||
if r.OffenerBefund() {
|
||||
status = "OFFENER BEFUND"
|
||||
}
|
||||
fmt.Fprintf(&b, "| %s | %d | %d | %s | %s | %s | %s | %s |\n",
|
||||
r.TenantSlug, r.SourceCount, r.TargetCount,
|
||||
r.SourceOldest.Format(timeLayout), r.TargetOldest.Format(timeLayout),
|
||||
r.SourceNewest.Format(timeLayout), r.TargetNewest.Format(timeLayout),
|
||||
status)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
const timeLayout = "2006-01-02T15:04:05Z07:00"
|
||||
|
||||
// OffeneBefunde filtert alle Reports mit Luecke oder Zeitverschiebung.
|
||||
func OffeneBefunde(reports []ContinuityReport) []ContinuityReport {
|
||||
var out []ContinuityReport
|
||||
for _, r := range reports {
|
||||
if r.OffenerBefund() {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
Reference in new issue
Block a user