test/apimachinery/watchlist: server supports sending resources in Table format

test/e2e/apimachinery/watchlist.go

@@ -33,7 +33,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/features"
@@ -229,9 +231,34 @@ var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithFeatureGate(
 		framework.ExpectNoError(err)
 		gomega.Expect(hasPreparedOptions).To(gomega.BeTrueBecause("it should be possible to prepare watchlist opts from an empty ListOptions"))
 
-		w, err := dynamicClient.Resource(v1.SchemeGroupVersion.WithResource("secrets")).Namespace("default").Watch(ctx, opts)
+		ginkgo.By(fmt.Sprintf("Adding 5 secrets to %s namespace", f.Namespace.Name))
+		wellKnownSecrets := addWellKnownUnstructuredSecrets(ctx, f)
+
+		ginkgo.By(fmt.Sprintf("Retrieving the secrets from %s namespace in table format", f.Namespace.Name))
+		expectedSecrets := []*unstructured.Unstructured{}
+		for i, wellKnownSecret := range wellKnownSecrets {
+			actualSecret, err := dynamicClient.Resource(v1.SchemeGroupVersion.WithResource("secrets")).Namespace(f.Namespace.Name).Get(ctx, wellKnownSecret.GetName(), metav1.GetOptions{})
+			framework.ExpectNoError(err)
+			if i != 0 {
+				// only the first obj has
+				// the column definition
+				actualSecret = removeColumnDefinitionsFromTable(actualSecret)
+			}
+			expectedSecrets = append(expectedSecrets, actualSecret)
+		}
+
+		ginkgo.By("Verifying if the secrets can be streamed in table format")
+		w, err := dynamicClient.Resource(v1.SchemeGroupVersion.WithResource("secrets")).Namespace(f.Namespace.Name).Watch(ctx, opts)
 		framework.ExpectNoError(err)
 		defer w.Stop()
+
+		for _, expectedSecret := range expectedSecrets {
+			expectEvent(w, watch.Added, expectedSecret)
+		}
+		rawBookmark := retrieveEventOfType(w, watch.Bookmark)
+		if !hasTableObjectInitialEventsAnnotationInBookmarkObj(rawBookmark) {
+			framework.Failf("expected the bookmark object to contain the required annotation, obj: %v", rawBookmark)
+		}
 	})
 
 	ginkgo.It("reflector doesn't support receiving resources as Tables", func(ctx context.Context) {
@@ -424,3 +451,87 @@ func toPartialObjectMetadata(rawItems []interface{}) ([]*metav1.PartialObjectMet
 	}
 	return ret, nil
 }
+
+func retrieveEventOfType(watch watch.Interface, expectedType watch.EventType) runtime.Object {
+	select {
+	case event, ok := <-watch.ResultChan():
+		if !ok {
+			framework.Failf("watch closed unexpectedly")
+		}
+		if event.Type != expectedType {
+			framework.Failf("unexpected watch event type: %v, expected: %v", event.Type, expectedType)
+		}
+		return event.Object
+	case <-time.After(wait.ForeverTestTimeout):
+		framework.Failf("timed out waiting for watch event")
+	}
+	return nil
+}
+
+func hasTableObjectInitialEventsAnnotationInBookmarkObj(rawObject runtime.Object) bool {
+	table, err := decodeIntoTable(rawObject)
+	framework.ExpectNoError(err)
+	if len(table.Rows) == 0 {
+		framework.Failf("table has no rows")
+	}
+	if len(table.Rows) != 1 {
+		framework.Failf("expected 1 row in the Table, got %d", len(table.Rows))
+	}
+
+	internalObjMeta, err := extractMetadataFromTableRowObject(table.Rows[0])
+	framework.ExpectNoError(err)
+	return internalObjMeta.GetAnnotations()[metav1.InitialEventsAnnotationKey] == "true"
+}
+
+var (
+	supportedTableVersions = map[schema.GroupVersionKind]bool{
+		metav1beta1.SchemeGroupVersion.WithKind("Table"): true,
+		metav1.SchemeGroupVersion.WithKind("Table"):      true,
+	}
+
+	_ metav1.Table      = metav1beta1.Table{}
+	_ metav1beta1.Table = metav1.Table{}
+)
+
+// extractMetadataFromTableRowObject retrieves the metav1.Object
+// from a single TableRow. It handles two scenarios:
+//  1. If row.Object.Object is already populated, it simply returns its metadata.
+//  2. Otherwise, it decodes row.Object.Raw into a runtime.Object and then returns its metadata.
+func extractMetadataFromTableRowObject(row metav1.TableRow) (metav1.Object, error) {
+	if row.Object.Raw == nil && row.Object.Object == nil {
+		return nil, fmt.Errorf("no object was found in the table")
+	}
+	if row.Object.Object != nil {
+		return meta.Accessor(row.Object.Object)
+	}
+
+	internalObj, err := runtime.Decode(unstructured.UnstructuredJSONScheme, row.Object.Raw)
+	framework.ExpectNoError(err)
+	return meta.Accessor(internalObj)
+}
+
+// decodeIntoTable converts a runtime.Object into a *metav1.Table.
+func decodeIntoTable(rawObject runtime.Object) (*metav1.Table, error) {
+	unstructuredObj, ok := rawObject.(*unstructured.Unstructured)
+	if !ok {
+		return nil, fmt.Errorf("expected *unstructured.Unstructured got %T", rawObject)
+	}
+	if !supportedTableVersions[rawObject.GetObjectKind().GroupVersionKind()] {
+		return nil, fmt.Errorf("unsupported Table GVK: %v", rawObject.GetObjectKind().GroupVersionKind())
+	}
+
+	var table metav1.Table
+	err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredObj.Object, &table)
+	framework.ExpectNoError(err)
+	return &table, nil
+}
+
+func removeColumnDefinitionsFromTable(rawObject *unstructured.Unstructured) *unstructured.Unstructured {
+	table, err := decodeIntoTable(rawObject)
+	framework.ExpectNoError(err)
+	table.ColumnDefinitions = nil
+
+	rawTable, err := runtime.DefaultUnstructuredConverter.ToUnstructured(table)
+	framework.ExpectNoError(err)
+	return &unstructured.Unstructured{Object: rawTable}
+}