package workflowevent import ( "context" "fmt" "net" "net/http" "os" "os/exec" "path/filepath" "runtime" "sort" "strings" "sync" "testing" "time" "github.com/jackc/pgx/v5/pgxpool" "gitea.perlbach24.de/scripte/nexarch/mail/internal/testdb" ) // TestMain entfernt das einmalig gebaute workflow-app-Binary wieder, damit // wiederholte Testläufe nicht Temp-Verzeichnisse ansammeln. func TestMain(m *testing.M) { code := m.Run() if workflowBinaryPath != "" { _ = os.RemoveAll(filepath.Dir(workflowBinaryPath)) } os.Exit(code) } // realWorkflowInstance ist eine ECHTE, lauffähige Workflow-App // (workflow/cmd/workflow-app), gestartet als eigener Prozess gegen ein // frisches Postgres-Testschema — keine Simulation des HTTP-Vertrags, wie // von der Ticket-Prüfung "gegen eine echte Workflow-Testinstanz" // gefordert. Mail und Workflow sind physisch getrennte Go-Module // (eigene go.mod je Modul), deshalb Kopplung ausschließlich über HTTP // wie in Produktion, nicht über einen Go-Import. type realWorkflowInstance struct { BaseURL string ServiceToken string Pool *pgxpool.Pool // dasselbe Testschema, fuer Pruefungen direkt per SQL cmd *exec.Cmd } var ( workflowBinaryOnce sync.Once workflowBinaryPath string workflowBinaryErr error ) // workflowModuleRoot findet das Wurzelverzeichnis des Workflow-Go-Moduls // (nexarch/workflow) relativ zu dieser Testdatei — Mail und Workflow // liegen als Geschwisterverzeichnisse im selben Checkout. func workflowModuleRoot(t *testing.T) string { t.Helper() _, thisFile, _, ok := runtime.Caller(0) if !ok { t.Fatal("eigener dateipfad nicht ermittelbar") } dir := filepath.Dir(thisFile) for { if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { break } parent := filepath.Dir(dir) if parent == dir { t.Fatal("go.mod des mail-moduls nicht gefunden") } dir = parent } repoRoot := filepath.Dir(dir) workflowRoot := filepath.Join(repoRoot, "workflow") if _, err := os.Stat(filepath.Join(workflowRoot, "go.mod")); err != nil { t.Skipf("workflow-modul nicht neben mail-modul gefunden (%s): %v", workflowRoot, err) } return workflowRoot } func buildWorkflowApp(t *testing.T) string { t.Helper() workflowBinaryOnce.Do(func() { workflowRoot := workflowModuleRoot(t) dir, err := os.MkdirTemp("", "workflow-app-bin-*") if err != nil { workflowBinaryErr = err return } bin := filepath.Join(dir, "workflow-app") cmd := exec.Command("go", "build", "-o", bin, "./cmd/workflow-app") cmd.Dir = workflowRoot out, err := cmd.CombinedOutput() if err != nil { workflowBinaryErr = fmt.Errorf("workflow-app bauen: %w\n%s", err, out) return } workflowBinaryPath = bin }) if workflowBinaryErr != nil { t.Fatalf("%v", workflowBinaryErr) } return workflowBinaryPath } func freeLocalPort(t *testing.T) int { t.Helper() l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("freien port ermitteln: %v", err) } defer l.Close() return l.Addr().(*net.TCPAddr).Port } // applyWorkflowMigrations wendet alle Workflow-Tenant-Migrationen in // Dateireihenfolge an -- reine SQL-Dateien, kein Import von // workflow/internal/migrate (anderes Go-Modul). func applyWorkflowMigrations(ctx context.Context, t *testing.T, pool *pgxpool.Pool, workflowRoot string) { t.Helper() dir := filepath.Join(workflowRoot, "migrations", "tenant") entries, err := os.ReadDir(dir) if err != nil { t.Fatalf("migrationsverzeichnis lesen: %v", err) } var files []string for _, e := range entries { if strings.HasSuffix(e.Name(), ".up.sql") { files = append(files, e.Name()) } } sort.Strings(files) for _, f := range files { sql, err := os.ReadFile(filepath.Join(dir, f)) if err != nil { t.Fatalf("migration %q lesen: %v", f, err) } if _, err := pool.Exec(ctx, string(sql)); err != nil { t.Fatalf("migration %q anwenden: %v", f, err) } } } // startRealWorkflowInstance baut (einmal pro Testlauf) und startet die // echte workflow-app gegen ein frisches Testschema. Prüfung 1: der // Objekttyp und seine Ereignisse landen in einer tatsächlich laufenden // Workflow-Instanz, nicht in einem Fake. func startRealWorkflowInstance(t *testing.T, dsn string) *realWorkflowInstance { t.Helper() workflowRoot := workflowModuleRoot(t) bin := buildWorkflowApp(t) pool, schema := testdb.PoolWithSchema(t, dsn, nil) ctx := context.Background() applyWorkflowMigrations(ctx, t, pool, workflowRoot) port := freeLocalPort(t) addr := fmt.Sprintf("127.0.0.1:%d", port) serviceToken := "int10-test-token-" + testdb.Suffix() schemaDSN := dsn + "&search_path=" + schema + ",public" cmd := exec.Command(bin) cmd.Env = append(os.Environ(), "NEXARCH_WORKFLOW_APP_TENANT_DSN="+schemaDSN, "NEXARCH_WORKFLOW_APP_SERVICE_TOKEN="+serviceToken, "NEXARCH_WORKFLOW_APP_TENANT_SLUG=mandant-int10", "NEXARCH_TENANT_REGISTRY_URL=http://127.0.0.1:1", "NEXARCH_WORKFLOW_APP_LISTEN_ADDR="+addr, ) var stderr strings.Builder cmd.Stderr = &stderr if err := cmd.Start(); err != nil { t.Fatalf("workflow-app starten: %v", err) } baseURL := "http://" + addr if err := waitForHealthz(baseURL, 10*time.Second); err != nil { _ = cmd.Process.Kill() t.Fatalf("workflow-app wurde nicht rechtzeitig erreichbar: %v\nstderr:\n%s", err, stderr.String()) } t.Cleanup(func() { if cmd.Process != nil { _ = cmd.Process.Kill() _ = cmd.Wait() } }) return &realWorkflowInstance{BaseURL: baseURL, ServiceToken: serviceToken, Pool: pool, cmd: cmd} } func waitForHealthz(baseURL string, timeout time.Duration) error { deadline := time.Now().Add(timeout) var lastErr error for time.Now().Before(deadline) { resp, err := http.Get(baseURL + "/healthz") if err == nil { resp.Body.Close() if resp.StatusCode == http.StatusOK { return nil } lastErr = fmt.Errorf("status %d", resp.StatusCode) } else { lastErr = err } time.Sleep(100 * time.Millisecond) } return lastErr }