Function SetupCDCFlowStatusQuery

flow/e2e/test_utils.go:264–292

(t *testing.T, env WorkflowRun, config *protos.FlowConnectionConfigs)

}

func SetupCDCFlowStatusQuery(t *testing.T, env WorkflowRun, config *protos.FlowConnectionConfigs) {
	t.Helper()
	pool, err := internal.GetCatalogConnectionPoolFromEnv(t.Context())
	if err != nil {
		env.Cancel(t.Context())
		t.Fatal("could not get catalog connection", err)
	}
	// errors expected while PeerFlowStatusQuery is setup
	counter := 0
	for {
		time.Sleep(time.Second)
		counter++
		status, err := internal.GetWorkflowStatus(t.Context(), pool, env.GetID())
		if err == nil {
			if status == protos.FlowStatus_STATUS_RUNNING || status == protos.FlowStatus_STATUS_COMPLETED {
				return
			} else if counter > 60 {
				env.Cancel(t.Context())
				t.Fatal("UNEXPECTED STATUS TIMEOUT", status)
			}
		} else if counter > 30 {
			env.Cancel(t.Context())
			t.Fatal("UNEXPECTED STATUS QUERY TIMEOUT", err.Error())
		} else if counter > 5 {
			// log the error for informational purposes
			t.Log(err.Error())
		}
	}
}

func CreateTableForQRep(ctx context.Context, conn *pgx.Conn, suffix string, tableName string) error {
	createMoodEnum := "CREATE TYPE mood AS ENUM ('happy', 'sad', 'angry');"