Prevent unnecessary list queries in the SQL Proxy.

pkg/stores/sqlproxy/proxy_store.go

@@ -10,6 +10,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -613,6 +614,11 @@ func (s *Store) Delete(apiOp *types.APIRequest, schema *types.APISchema, id stri
 //   - a continue token, if there are more pages after the returned one
 //   - an error instead of all of the above if anything went wrong
 func (s *Store) ListByPartitions(apiOp *types.APIRequest, schema *types.APISchema, partitions []partition.Partition) ([]unstructured.Unstructured, int, string, error) {
+	if !schemaAllowsVerb("list", schema) {
+		logrus.Debugf("list method is not allowed for %s", apiOp.Type)
+		return nil, 0, "", nil
+	}
+
 	opts, err := listprocessor.ParseQuery(apiOp, s.namespaceCache)
 	if err != nil {
 		return nil, 0, "", err
@@ -689,3 +695,14 @@ func (s *Store) watchByPartition(partition partition.Partition, apiOp *types.API
 	}
 	return s.WatchNames(apiOp, schema, wr, partition.Names)
 }
+
+func schemaAllowsVerb(verb string, schema *types.APISchema) bool {
+	verbs := attributes.Verbs(schema)
+	if len(verbs) == 0 {
+		// Again, if we can't determine whether nor not the verb is allowed, we might
+		// as well allow K8s to decide.
+		return true
+	}
+
+	return slices.Contains(verbs, verb)
+}

pkg/stores/sqlproxy/proxy_store_test.go

@@ -503,6 +503,43 @@ func TestListByPartitions(t *testing.T) {
 			assert.NotNil(t, err)
 		},
 	})
+	tests = append(tests, testCase{
+		description: "client ListByPartitions() with schema that doesn't allow listing should get an empty response",
+		test: func(t *testing.T) {
+			nsi := NewMockCache(gomock.NewController(t))
+			cg := NewMockClientGetter(gomock.NewController(t))
+			cf := NewMockCacheFactory(gomock.NewController(t))
+
+			s := &Store{
+				namespaceCache: nsi,
+				clientGetter:   cg,
+				cacheFactory:   cf,
+			}
+			var partitions []partition.Partition
+			req := &types.APIRequest{
+				Request: &http.Request{
+					URL: &url.URL{},
+				},
+			}
+			schema := &types.APISchema{
+				Schema: &schemas.Schema{Attributes: map[string]interface{}{
+					"columns": []common.ColumnDefinition{
+						{
+							Field: "some.field",
+						},
+					},
+					"verbs": []string{"create"},
+				}},
+			}
+
+			items, count, token, err := s.ListByPartitions(req, schema, partitions)
+			assert.Nil(t, items)
+			assert.Equal(t, count, 0)
+			assert.Empty(t, token)
+			assert.Nil(t, err)
+		},
+	})
+
 	t.Parallel()
 	for _, test := range tests {
 		t.Run(test.description, func(t *testing.T) { test.test(t) })