Function TestStatefulFunction

server/server_test.go:237–291

(t *testing.T)

}

func TestStatefulFunction(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	s, httpAddr := startStandaloneSvr(t, ctx,
		WithRuntimeFactoryBuilder("mock", func(configMap common.ConfigMap) (api.FunctionRuntimeFactory, error) {
			return &MockRuntimeFactory{}, nil
		}))

	input := "input"
	output := "output"
	funcConf := &model.Function{
		Name: "test-func",
		Runtime: model.RuntimeConfig{
			Type: "mock",
		},
		Sources: []model.TubeConfig{
			{
				Type: common.MemoryTubeType,
				Config: (&contube.SourceQueueConfig{
					Topics:  []string{input},
					SubName: "test",
				}).ToConfigMap(),
			},
		},
		Sink: model.TubeConfig{
			Type: common.MemoryTubeType,
			Config: (&contube.SinkQueueConfig{
				Topic: "output",
			}).ToConfigMap(),
		},
		Replicas: 1,
	}
	err := s.Manager.StartFunction(funcConf)
	if err != nil {
		t.Fatal(err)
	}

	cfg := adminclient.NewConfiguration()
	cfg.Host = httpAddr
	cli := adminclient.NewAPIClient(cfg)

	_, err = cli.StateAPI.SetState(ctx, "key").Body("hello").Execute()
	assert.Nil(t, err)

	err = s.Manager.ProduceEvent(input, contube.NewRecordImpl(nil, func() {
	}))
	assert.Nil(t, err)

	_, err = s.Manager.ConsumeEvent(output)
	assert.Nil(t, err)

	result, _, err := cli.StateAPI.GetState(ctx, "key").Execute()
	assert.Nil(t, err)
	assert.Equal(t, "hello!", result)
}