Function TestFetch

scheduler/pkg/envoy/processor/server_test.go:39–82

(t *testing.T)

var permanentClusterNames = []string{"mirror_http", "mirror_grpc"}

func TestFetch(t *testing.T) {
	g := NewGomegaWithT(t)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	logger := log.New()

	memoryStore := store.NewMemoryStore(logger, store.NewLocalSchedulerStore(), nil)
	pipelineHandler := pipeline.NewPipelineStore(logger, nil, memoryStore)

	xdsCache, err := xdscache.NewSeldonXDSCache(log.New(), &xdscache.PipelineGatewayDetails{}, nil)
	g.Expect(err).To(BeNil())
	inc := &IncrementalProcessor{
		logger:           logger,
		xdsCache:         xdsCache,
		pipelineHandler:  pipelineHandler,
		modelStore:       memoryStore,
		experimentServer: experiment.NewExperimentServer(logger, nil, memoryStore, pipelineHandler),
		nodeID:           "node_1",
		batchWait:        time.Nanosecond,
	}

	port, err := testing_utils.GetFreePortForTest()
	if err != nil {
		t.Fatal(err)
	}
	go func() {
		err := startAdsServer(inc, uint(port))
		g.Expect(err).To(BeNil())
	}()

	g.Expect(err).To(BeNil())

	conn, err := grpc.NewClient(":"+strconv.Itoa(port), grpc.WithTransportCredentials(insecure.NewCredentials()))
	g.Expect(err).To(BeNil())
	defer conn.Close()

	c := client.NewADSClient(ctx, &core.Node{Id: "node_1"}, resource.ClusterType)
	err = c.InitConnect(conn)
	g.Expect(err).To(BeNil())

	t.Run("Test initial fetch with model version 1", testInitialFetch(g, inc, c))
	t.Run("Test update model version to 2", testUpdateModelVersion(g, inc, c))
}

func testInitialFetch(g *WithT, inc *IncrementalProcessor, c client.ADSClient) func(t *testing.T) {
	firstFetch := []string{"model_1_grpc", "model_1_http"}