tests: add version compatibility test for upstreams

internal/test/e2e/version_compatibility_test.go

@@ -37,7 +37,12 @@ type vcPlugins struct {
 	fieldUpdateChecks map[string][]update
 }
 
-func TestVersionCompatibility(t *testing.T) {
+type vcUpstreams struct {
+	upstream          *v1.Upstream
+	versionedExpected map[string]*v1.Upstream
+}
+
+func TestPluginsVersionCompatibility(t *testing.T) {
 	cleanup := run.Koko(t)
 	defer cleanup()
 
@@ -329,6 +334,94 @@ func TestVersionCompatibility(t *testing.T) {
 	})
 }
 
+func TestUpstreamsVersionCompatibility(t *testing.T) {
+	cleanup := run.Koko(t)
+	defer cleanup()
+
+	dpCleanup := run.KongDP(kong.GetKongConfForShared())
+	defer dpCleanup()
+	util.WaitForKong(t)
+	util.WaitForKongAdminAPI(t)
+
+	admin := httpexpect.New(t, "http://localhost:3000")
+
+	tests := []vcUpstreams{
+		{
+			upstream: &v1.Upstream{
+				Id:             uuid.NewString(),
+				Name:           "foo",
+				HashOn:         "ip",
+				HashOnQueryArg: "test",
+			},
+			versionedExpected: map[string]*v1.Upstream{
+				"< 3.0.0": {
+					Name:           "foo",
+					HashOn:         "ip",
+					HashOnQueryArg: "",
+				},
+				">= 3.0.0": {
+					Name:           "foo",
+					HashOn:         "ip",
+					HashOnQueryArg: "test",
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		res := admin.POST("/v1/upstreams").WithJSON(test.upstream).Expect()
+		res.Status(http.StatusCreated)
+	}
+
+	util.WaitFunc(t, func() error {
+		err := ensureUpstreams(tests)
+		t.Log("upstreams validation failed", err)
+		return err
+	})
+}
+
+func ensureUpstreams(upstreams []vcUpstreams) error {
+	kongAdmin, err := kongClient.NewClient(util.BasedKongAdminAPIAddr, nil)
+	if err != nil {
+		return fmt.Errorf("create go client for kong: %v", err)
+	}
+	ctx := context.Background()
+	info, err := kongAdmin.Root(ctx)
+	if err != nil {
+		return fmt.Errorf("fetching Kong Gateway info: %v", err)
+	}
+	dataPlaneVersion, err := kongClient.ParseSemanticVersion(kongClient.VersionFromInfo(info))
+	if err != nil {
+		return fmt.Errorf("parsing Kong Gateway version: %v", err)
+	}
+	dataPlaneUpstreams, err := kongAdmin.Upstreams.ListAll(ctx)
+	if err != nil {
+		return fmt.Errorf("fetching upstreams: %v", err)
+	}
+
+	if len(upstreams) != len(dataPlaneUpstreams) {
+		return fmt.Errorf("upstreams configured count does not match [%d != %d]", len(upstreams), len(dataPlaneUpstreams))
+	}
+
+	for _, u := range upstreams {
+		for _, dataPlaneUpstream := range dataPlaneUpstreams {
+			if u.upstream.Name == *dataPlaneUpstream.Name && u.upstream.Id == *dataPlaneUpstream.ID {
+				for version, expectedUpstream := range u.versionedExpected {
+					version := semver.MustParseRange(version)
+					if version(dataPlaneVersion) {
+						expectedConfig := &v1.TestingConfig{
+							Upstreams: []*v1.Upstream{expectedUpstream},
+						}
+						err := util.EnsureConfig(expectedConfig)
+						return err
+					}
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 func ensurePlugins(plugins []vcPlugins) error {
 	kongAdmin, err := kongClient.NewClient(util.BasedKongAdminAPIAddr, nil)
 	if err != nil {