fixed integration test for ads (#440)

fixed integration tests for ads
envoyproxy · May 27, 2021 · e1f5cab · e1f5cab
1 parent 15c2d00
commit e1f5cab
Show file tree

Hide file tree

Showing 3 changed files with 81 additions and 17 deletions.
diff --git a/pkg/server/v3/delta_test.go b/pkg/server/v3/delta_test.go
@@ -379,3 +379,81 @@ func TestDeltaAggregatedHandlers(t *testing.T) {
 		}
 	}
 }
+
+func TestDeltaAggregateRequestType(t *testing.T) {
+	config := makeMockConfigWatcher()
+	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
+	resp := makeMockDeltaStream(t)
+	resp.recv <- &discovery.DeltaDiscoveryRequest{Node: node}
+	if err := s.DeltaAggregatedResources(resp); err == nil {
+		t.Error("DeltaAggregatedResources() => got nil, want an error")
+	}
+}
+
+func TestDeltaCancellations(t *testing.T) {
+	config := makeMockConfigWatcher()
+	resp := makeMockDeltaStream(t)
+	for _, typ := range testTypes {
+		resp.recv <- &discovery.DeltaDiscoveryRequest{
+			Node:    node,
+			TypeUrl: typ,
+		}
+	}
+	close(resp.recv)
+	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
+	if err := s.DeltaAggregatedResources(resp); err != nil {
+		t.Errorf("DeltaAggregatedResources() => got %v, want no error", err)
+	}
+	if config.watches != 0 {
+		t.Errorf("Expect all watches cancelled, got %q", config.watches)
+	}
+}
+
+func TestDeltaOpaqueRequestsChannelMuxing(t *testing.T) {
+	config := makeMockConfigWatcher()
+	resp := makeMockDeltaStream(t)
+	for i := 0; i < 10; i++ {
+		resp.recv <- &discovery.DeltaDiscoveryRequest{
+			Node:                   node,
+			TypeUrl:                fmt.Sprintf("%s%d", opaqueType, i%2),
+			ResourceNamesSubscribe: []string{fmt.Sprintf("%d", i)},
+		}
+	}
+	close(resp.recv)
+	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
+	if err := s.DeltaAggregatedResources(resp); err != nil {
+		t.Errorf("DeltaAggregatedResources() => got %v, want no error", err)
+	}
+	if config.watches != 0 {
+		t.Errorf("Expect all watches cancelled, got %q", config.watches)
+	}
+}
+
+func TestDeltaCallbackError(t *testing.T) {
+	for _, typ := range testTypes {
+		t.Run(typ, func(t *testing.T) {
+			config := makeMockConfigWatcher()
+			config.deltaResponses = makeDeltaResponses()
+
+			s := server.NewServer(context.Background(), config, server.CallbackFuncs{
+				DeltaStreamOpenFunc: func(ctx context.Context, i int64, s string) error {
+					return errors.New("stream open error")
+				},
+			})
+
+			// make a request
+			resp := makeMockDeltaStream(t)
+			resp.recv <- &discovery.DeltaDiscoveryRequest{
+				Node:    node,
+				TypeUrl: typ,
+			}
+
+			// check that response fails since stream open returns error
+			if err := s.DeltaAggregatedResources(resp); err == nil {
+				t.Error("Stream() => got no error, want error")
+			}
+
+			close(resp.recv)
+		})
+	}
+}
diff --git a/pkg/server/v3/server.go b/pkg/server/v3/server.go
@@ -95,7 +95,7 @@ func (c CallbackFuncs) OnStreamClosed(streamID int64) {
 
 // OnDeltaStreamOpen invokes DeltaStreamOpenFunc.
 func (c CallbackFuncs) OnDeltaStreamOpen(ctx context.Context, streamID int64, typeURL string) error {
-	if c.StreamOpenFunc != nil {
+	if c.DeltaStreamOpenFunc != nil {
 		return c.DeltaStreamOpenFunc(ctx, streamID, typeURL)
 	}
 
@@ -104,7 +104,7 @@ func (c CallbackFuncs) OnDeltaStreamOpen(ctx context.Context, streamID int64, ty
 
 // OnDeltaStreamClosed invokes DeltaStreamClosedFunc.
 func (c CallbackFuncs) OnDeltaStreamClosed(streamID int64) {
-	if c.StreamClosedFunc != nil {
+	if c.DeltaStreamClosedFunc != nil {
 		c.DeltaStreamClosedFunc(streamID)
 	}
 }

diff --git a/sample/bootstrap-delta-ads.yaml b/sample/bootstrap-delta-ads.yaml
@@ -35,18 +35,4 @@ static_resources:
                 address: 127.0.0.1
                 port_value: 18000
     http2_protocol_options: {}
-    name: xds_cluster
-layered_runtime:
-  layers:
-    - name: runtime-0
-      rtds_layer:
-        rtds_config:
-          resource_api_version: V3
-          ads: {}
-        name: runtime-0
-    - name: runtime-1
-      rtds_layer:
-        rtds_config:
-          resource_api_version: V3
-          ads: {}
-        name: runtime-1
+    name: xds_cluster