Function TestHttpTube

server/server_test.go:136–198

(t *testing.T)

}

func TestHttpTube(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	s, httpAddr := startStandaloneSvr(t, ctx, nil, nil)

	endpoint := "test-endpoint"
	funcConf := &model.Function{
		Runtime: model.RuntimeConfig{
			Type: common.WASMRuntime,
			Config: map[string]interface{}{
				common.RuntimeArchiveConfigKey: "../bin/example_basic.wasm",
			},
		},
		Sources: []model.TubeConfig{{
			Type: common.HttpTubeType,
			Config: map[string]interface{}{
				contube.EndpointKey: endpoint,
			},
		}},
		Sink: model.TubeConfig{
			Type: common.MemoryTubeType,
			Config: (&contube.SinkQueueConfig{
				Topic: "output",
			}).ToConfigMap(),
		},
		Name:     "test-func",
		Replicas: 1,
	}

	err := s.Manager.StartFunction(funcConf)
	assert.Nil(t, err)

	p := &tests.Person{
		Name:  "rbt",
		Money: 0,
	}
	jsonBytes, err := json.Marshal(p)
	if err != nil {
		t.Fatal(err)
	}

	cfg := adminclient.NewConfiguration()
	cfg.Host = httpAddr
	cli := adminclient.NewAPIClient(cfg)
	_, err = cli.HttpTubeAPI.TriggerHttpTubeEndpoint(ctx, endpoint).Body(string(jsonBytes)).Execute()
	assert.Nil(t, err)

	event, err := s.Manager.ConsumeEvent("output")
	if err != nil {
		t.Error(err)
		return
	}
	var out tests.Person
	err = json.Unmarshal(event.GetPayload(), &out)
	if err != nil {
		t.Error(err)
		return
	}