Method TestCreateTopic

flow/e2e/pubsub_test.go:125–180

()

}

func (s PubSubSuite) TestCreateTopic() {
	srcTableName := AttachSchema(s, "pscreate")

	_, err := s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
		CREATE TABLE IF NOT EXISTS %s (
			id SERIAL PRIMARY KEY,
			val text
		);
	`, srcTableName))
	require.NoError(s.t, err)

	sa, err := ServiceAccount()
	require.NoError(s.t, err)

	_, err = s.Conn().Exec(s.t.Context(), `insert into public.scripts (name, lang, source) values
	('e2e_pscreate', 'lua', 'function onRecord(r) return r.row and r.row.val end') on conflict do nothing`)
	require.NoError(s.t, err)

	flowName := AddSuffix(s, "e2epscreate")
	connectionGen := FlowConnectionGenerationConfig{
		FlowJobName:      flowName,
		TableNameMapping: map[string]string{srcTableName: flowName},
		Destination:      s.Peer(sa).Name,
	}
	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
	flowConnConfig.Script = "e2e_pscreate"

	tc := NewTemporalClient(s.t)
	env := ExecutePeerflow(s.t, tc, flowConnConfig)
	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)

	_, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(`
		INSERT INTO %s (id, val) VALUES (1, 'testval')
	`, srcTableName))
	require.NoError(s.t, err)

	EnvWaitFor(s.t, env, 3*time.Minute, "create topic", func() bool {
		psclient, err := sa.CreatePubSubClient(s.t.Context())
		defer func() {
			_ = psclient.Close()
		}()
		require.NoError(s.t, err)
		topic := fmt.Sprintf("projects/%s/topics/%s", psclient.Project(), flowName)
		if _, err := psclient.TopicAdminClient.GetTopic(s.t.Context(), &pubsubpb.GetTopicRequest{
			Topic: topic,
		}); err == nil {
			return true
		} else if status.Code(err) != codes.NotFound {
			require.NoError(s.t, err)
		}
		return false
	})

	env.Cancel(s.t.Context())
	RequireEnvCanceled(s.t, env)
}

func (s PubSubSuite) TestSimple() {