Merge pull request #121118 from alexzielenski/apiserver/apiextensions/ratcheting-factor-correlation

CRDValidationRatcheting: Factor object correlation and comparison into reusable component

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/validation/ratcheting.go

@@ -60,7 +60,10 @@ func (r *RatchetingSchemaValidator) Validate(new interface{}) *validate.Result {
 }
 
 func (r *RatchetingSchemaValidator) ValidateUpdate(new, old interface{}) *validate.Result {
-	return newRatchetingValueValidator(new, old, r.schemaArgs).Validate()
+	return newRatchetingValueValidator(
+		common.NewCorrelatedObject(new, old, &celopenapi.Schema{Schema: r.schemaArgs.schema}),
+		r.schemaArgs,
+	).Validate(new)
 }
 
 // ratchetingValueValidator represents an invocation of SchemaValidator.ValidateUpdate
@@ -80,40 +83,13 @@ type ratchetingValueValidator struct {
 	// schemaArgs provides the arguments to use in the temporary SchemaValidator
 	// that is created during a call to Validate.
 	schemaArgs
-
+	correlation *common.CorrelatedObject
-	// Currently correlated old value during traversal of the schema/object
-	oldValue interface{}
-
-	// Value being validated
-	value interface{}
-
-	// Scratch space below, may change during validation
-
-	// Cached comparison result of DeepEqual of `value` and `thunk.oldValue`
-	comparisonResult *bool
-
-	// Cached map representation of a map-type list, or nil if not map-type list
-	mapList common.MapList
-
-	// Children spawned by a call to `Validate` on this object
-	// key is either a string or an index, depending upon whether `value` is
-	// a map or a list, respectively.
-	//
-	// The list of children may be incomplete depending upon if the internal
-	// logic of kube-openapi's SchemaValidator short-circuited before
-	// reaching all of the children.
-	//
-	// It should be expected to have an entry for either all of the children, or
-	// none of them.
-	children map[interface{}]*ratchetingValueValidator
 }
 
-func newRatchetingValueValidator(newValue, oldValue interface{}, args schemaArgs) *ratchetingValueValidator {
+func newRatchetingValueValidator(correlation *common.CorrelatedObject, args schemaArgs) *ratchetingValueValidator {
 	return &ratchetingValueValidator{
-		oldValue:   oldValue,
+		schemaArgs:  args,
-		value:      newValue,
+		correlation: correlation,
-		schemaArgs: args,
-		children:   map[interface{}]*ratchetingValueValidator{},
 	}
 }
 
@@ -121,12 +97,8 @@ func newRatchetingValueValidator(newValue, oldValue interface{}, args schemaArgs
 // that injects a ratchetingValueValidator to be used for all subkeys and subindices
 func (r *ratchetingValueValidator) getValidateOption() validate.Option {
 	return func(svo *validate.SchemaValidatorOptions) {
-		svo.NewValidatorForField = func(field string, schema *spec.Schema, rootSchema interface{}, root string, formats strfmt.Registry, opts ...validate.Option) validate.ValueValidator {
+		svo.NewValidatorForField = r.SubPropertyValidator
-			return r.SubPropertyValidator(field, schema, rootSchema, root, formats, opts...)
+		svo.NewValidatorForIndex = r.SubIndexValidator
-		}
-		svo.NewValidatorForIndex = func(index int, schema *spec.Schema, rootSchema interface{}, root string, formats strfmt.Registry, opts ...validate.Option) validate.ValueValidator {
-			return r.SubIndexValidator(index, schema, rootSchema, root, formats, opts...)
-		}
 	}
 }
 
@@ -149,21 +121,21 @@ func (r *ratchetingValueValidator) getValidateOption() validate.Option {
 //
 // This call has a side-effect of populating it's `children` variable with
 // the explored nodes of the object tree.
-func (r *ratchetingValueValidator) Validate() *validate.Result {
+func (r *ratchetingValueValidator) Validate(new interface{}) *validate.Result {
 	opts := append([]validate.Option{
 		r.getValidateOption(),
 	}, r.options...)
 
 	s := validate.NewSchemaValidator(r.schema, r.root, r.path, r.knownFormats, opts...)
 
-	res := s.Validate(r.value)
+	res := s.Validate(r.correlation.Value)
 
 	if res.IsValid() {
 		return res
 	}
 
 	// Current ratcheting rule is to ratchet errors if DeepEqual(old, new) is true.
-	if r.CachedDeepEqual() {
+	if r.correlation.CachedDeepEqual() {
 		newRes := &validate.Result{}
 		newRes.MergeAsWarnings(res)
 		return newRes
@@ -180,30 +152,18 @@ func (r *ratchetingValueValidator) Validate() *validate.Result {
 //
 // If the old value cannot be correlated, then default validation is used.
 func (r *ratchetingValueValidator) SubPropertyValidator(field string, schema *spec.Schema, rootSchema interface{}, root string, formats strfmt.Registry, options ...validate.Option) validate.ValueValidator {
-	// Find correlated old value
+	childNode := r.correlation.Key(field)
-	asMap, ok := r.oldValue.(map[string]interface{})
+	if childNode == nil {
-	if !ok {
 		return validate.NewSchemaValidator(schema, rootSchema, root, formats, options...)
 	}
 
-	oldValueForField, ok := asMap[field]
+	return newRatchetingValueValidator(childNode, schemaArgs{
-	if !ok {
+		schema:       schema,
-		return validate.NewSchemaValidator(schema, rootSchema, root, formats, options...)
+		root:         rootSchema,
-	}
+		path:         root,
-
+		knownFormats: formats,
-	return inlineValidator(func(new interface{}) *validate.Result {
+		options:      options,
-		childNode := newRatchetingValueValidator(new, oldValueForField, schemaArgs{
-			schema:       schema,
-			root:         rootSchema,
-			path:         root,
-			knownFormats: formats,
-			options:      options,
-		})
-
-		r.children[field] = childNode
-		return childNode.Validate()
 	})
-
 }
 
 // SubIndexValidator overrides the standard validator constructor for sub-indicies by
@@ -214,199 +174,27 @@ func (r *ratchetingValueValidator) SubPropertyValidator(field string, schema *sp
 //
 // If the old value cannot be correlated, then default validation is used.
 func (r *ratchetingValueValidator) SubIndexValidator(index int, schema *spec.Schema, rootSchema interface{}, root string, formats strfmt.Registry, options ...validate.Option) validate.ValueValidator {
-	oldValueForIndex := r.correlateOldValueForChildAtNewIndex(index)
+	childNode := r.correlation.Index(index)
-	if oldValueForIndex == nil {
+	if childNode == nil {
-		// If correlation fails, default to non-ratcheting logic
 		return validate.NewSchemaValidator(schema, rootSchema, root, formats, options...)
 	}
 
-	return inlineValidator(func(new interface{}) *validate.Result {
+	return newRatchetingValueValidator(childNode, schemaArgs{
-		childNode := newRatchetingValueValidator(new, oldValueForIndex, schemaArgs{
+		schema:       schema,
-			schema:       schema,
+		root:         rootSchema,
-			root:         rootSchema,
+		path:         root,
-			path:         root,
+		knownFormats: formats,
-			knownFormats: formats,
+		options:      options,
-			options:      options,
-		})
-
-		r.children[index] = childNode
-		return childNode.Validate()
 	})
 }
 
-// If oldValue is not a list, returns nil
+var _ validate.ValueValidator = (&ratchetingValueValidator{})
-// If oldValue is a list takes mapType into account and attempts to find the
-// old value with the same index or key, depending upon the mapType.
-//
-// If listType is map, creates a map representation of the list using the designated
-// map-keys and caches it for future calls.
-func (r *ratchetingValueValidator) correlateOldValueForChildAtNewIndex(index int) any {
-	oldAsList, ok := r.oldValue.([]interface{})
-	if !ok {
-		return nil
-	}
 
-	asList, ok := r.value.([]interface{})
+func (r ratchetingValueValidator) SetPath(path string) {
-	if !ok {
-		return nil
-	} else if len(asList) <= index {
-		// Cannot correlate out of bounds index
-		return nil
-	}
-
-	listType, _ := r.schema.Extensions.GetString("x-kubernetes-list-type")
-	switch listType {
-	case "map":
-		// Look up keys for this index in current object
-		currentElement := asList[index]
-
-		oldList := r.mapList
-		if oldList == nil {
-			oldList = celopenapi.MakeMapList(r.schema, oldAsList)
-			r.mapList = oldList
-		}
-		return oldList.Get(currentElement)
-
-	case "set":
-		// Are sets correlatable? Only if the old value equals the current value.
-		// We might be able to support this, but do not currently see a lot
-		// of value
-		// (would allow you to add/remove items from sets with ratcheting but not change them)
-		return nil
-	case "atomic":
-		// Atomic lists are not correlatable by item
-		// Ratcheting is not available on a per-index basis
-		return nil
-	default:
-		// Correlate by-index by default.
-		//
-		// Cannot correlate an out-of-bounds index
-		if len(oldAsList) <= index {
-			return nil
-		}
-
-		return oldAsList[index]
-	}
-}
-
-// CachedDeepEqual is equivalent to reflect.DeepEqual, but caches the
-// results in the tree of ratchetInvocationScratch objects on the way:
-//
-// For objects and arrays, this function will make a best effort to make
-// use of past DeepEqual checks performed by this Node's children, if available.
-//
-// If a lazy computation could not be found for all children possibly due
-// to validation logic short circuiting and skipping the children, then
-// this function simply defers to reflect.DeepEqual.
-func (r *ratchetingValueValidator) CachedDeepEqual() (res bool) {
-	if r.comparisonResult != nil {
-		return *r.comparisonResult
-	}
-
-	defer func() {
-		r.comparisonResult = &res
-	}()
-
-	if r.value == nil && r.oldValue == nil {
-		return true
-	} else if r.value == nil || r.oldValue == nil {
-		return false
-	}
-
-	oldAsArray, oldIsArray := r.oldValue.([]interface{})
-	newAsArray, newIsArray := r.value.([]interface{})
-
-	if oldIsArray != newIsArray {
-		return false
-	} else if oldIsArray {
-		if len(oldAsArray) != len(newAsArray) {
-			return false
-		} else if len(r.children) != len(oldAsArray) {
-			// kube-openapi validator is written to always visit all
-			// children of a slice, so this case is only possible if
-			// one of the children could not be correlated. In that case,
-			// we know the objects are not equal.
-			//
-			return false
-		}
-
-		// Correctly considers map-type lists due to fact that index here
-		// is only used for numbering. The correlation is stored in the
-		// childInvocation itself
-		//
-		// NOTE: This does not consider sets, since we don't correlate them.
-		for i := range newAsArray {
-			// Query for child
-			child, ok := r.children[i]
-			if !ok {
-				// This should not happen
-				return false
-			} else if !child.CachedDeepEqual() {
-				// If one child is not equal the entire object is not equal
-				return false
-			}
-		}
-
-		return true
-	}
-
-	oldAsMap, oldIsMap := r.oldValue.(map[string]interface{})
-	newAsMap, newIsMap := r.value.(map[string]interface{})
-
-	if oldIsMap != newIsMap {
-		return false
-	} else if oldIsMap {
-		if len(oldAsMap) != len(newAsMap) {
-			return false
-		} else if len(oldAsMap) == 0 && len(newAsMap) == 0 {
-			// Both empty
-			return true
-		} else if len(r.children) != len(oldAsMap) {
-			// If we are missing a key it is because the old value could not
-			// be correlated to the new, so the objects are not equal.
-			//
-			return false
-		}
-
-		for k := range oldAsMap {
-			// Check to see if this child was explored during validation
-			child, ok := r.children[k]
-			if !ok {
-				// Child from old missing in new due to key change
-				// Objects are not equal.
-				return false
-			} else if !child.CachedDeepEqual() {
-				// If one child is not equal the entire object is not equal
-				return false
-			}
-		}
-
-		return true
-	}
-
-	return reflect.DeepEqual(r.oldValue, r.value)
-}
-
-// A validator which just calls a validate function, and advertises that it
-// validates anything
-//
-// In the future kube-openapi's ValueValidator interface can be simplified
-// to be closer to `currentValidator.Options.NewValidator(value, ...).Validate()`
-// so that a tree of "validation nodes" can be more formally encoded in the API.
-// In that case this class would not be necessary.
-type inlineValidator func(new interface{}) *validate.Result
-
-var _ validate.ValueValidator = inlineValidator(nil)
-
-func (f inlineValidator) Validate(new interface{}) *validate.Result {
-	return f(new)
-}
-
-func (f inlineValidator) SetPath(path string) {
 	// Do nothing
 	// Unused by kube-openapi
 }
 
-func (f inlineValidator) Applies(source interface{}, valueKind reflect.Kind) bool {
+func (r ratchetingValueValidator) Applies(source interface{}, valueKind reflect.Kind) bool {
 	return true
 }

staging/src/k8s.io/apiextensions-apiserver/test/integration/ratcheting_test.go

@@ -1175,7 +1175,7 @@ func TestRatchetingFunctionality(t *testing.T) {
 			},
 		},
 		{
-			Name: "ArrayItems correlate by index",
+			Name: "ArrayItems do not correlate by index",
 			Operations: []ratchetingTestOperation{
 				updateMyCRDV1Beta1Schema{&apiextensionsv1.JSONSchemaProps{
 					Type: "object",
@@ -1246,9 +1246,9 @@ func TestRatchetingFunctionality(t *testing.T) {
 						},
 						"otherField": "hello world",
 					}},
-				// (This test shows an array can be correlated by index with its old value)
+				// (This test shows an array cannpt be correlated by index with its old value)
-				applyPatchOperation{
+				expectError{applyPatchOperation{
-					"add new, valid fields to elements of the array, ratcheting unchanged old fields within the array elements by correlating by index",
+					"add new, valid fields to elements of the array, failing to ratchet unchanged old fields within the array elements by correlating by index due to atomic list",
 					myCRDV1Beta1, myCRDInstanceName, map[string]interface{}{
 						"values": []interface{}{
 							map[string]interface{}{
@@ -1261,7 +1261,7 @@ func TestRatchetingFunctionality(t *testing.T) {
 								"key2": "valid value",
 							},
 						},
-					}},
+					}}},
 				expectError{
 					applyPatchOperation{
 						"reorder the array, preventing index correlation",

staging/src/k8s.io/apiserver/pkg/cel/common/equality.go

@@ -0,0 +1,296 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"reflect"
+)
+
+// CorrelatedObject represents a node in a tree of objects that are being
+// validated. It is used to keep track of the old value of an object during
+// traversal of the new value. It is also used to cache the results of
+// DeepEqual comparisons between the old and new values of objects.
+//
+// All receiver functions support being called on `nil` to support ergonomic
+// recursive descent. The nil `CorrelatedObject` represents an uncorrelatable
+// node in the tree.
+//
+// CorrelatedObject is not thread-safe. It is the responsibility of the caller
+// to handle concurrency, if any.
+type CorrelatedObject struct {
+	// Currently correlated old value during traversal of the schema/object
+	OldValue interface{}
+
+	// Value being validated
+	Value interface{}
+
+	// Schema used for validation of this value. The schema is also used
+	// to determine how to correlate the old object.
+	Schema Schema
+
+	// Scratch space below, may change during validation
+
+	// Cached comparison result of DeepEqual of `value` and `thunk.oldValue`
+	comparisonResult *bool
+
+	// Cached map representation of a map-type list, or nil if not map-type list
+	mapList MapList
+
+	// Children spawned by a call to `Validate` on this object
+	// key is either a string or an index, depending upon whether `value` is
+	// a map or a list, respectively.
+	//
+	// The list of children may be incomplete depending upon if the internal
+	// logic of kube-openapi's SchemaValidator short-circuited before
+	// reaching all of the children.
+	//
+	// It should be expected to have an entry for either all of the children, or
+	// none of them.
+	children map[interface{}]*CorrelatedObject
+}
+
+func NewCorrelatedObject(new, old interface{}, schema Schema) *CorrelatedObject {
+	return &CorrelatedObject{
+		OldValue: old,
+		Value:    new,
+		Schema:   schema,
+	}
+}
+
+// If OldValue or Value is not a list, or the index is out of bounds of the
+// Value list, returns nil
+// If oldValue is a list, this considers the x-list-type to decide how to
+// correlate old values:
+//
+// If listType is map, creates a map representation of the list using the designated
+// map-keys, caches it for future calls, and returns the map value, or nil if
+// the correlated key is not in the old map
+//
+// Otherwise, if the list type is not correlatable this funcion returns nil.
+func (r *CorrelatedObject) correlateOldValueForChildAtNewIndex(index int) interface{} {
+	oldAsList, ok := r.OldValue.([]interface{})
+	if !ok {
+		return nil
+	}
+
+	asList, ok := r.Value.([]interface{})
+	if !ok {
+		return nil
+	} else if len(asList) <= index {
+		// Cannot correlate out of bounds index
+		return nil
+	}
+
+	listType := r.Schema.XListType()
+	switch listType {
+	case "map":
+		// Look up keys for this index in current object
+		currentElement := asList[index]
+
+		oldList := r.mapList
+		if oldList == nil {
+			oldList = MakeMapList(r.Schema, oldAsList)
+			r.mapList = oldList
+		}
+		return oldList.Get(currentElement)
+
+	case "set":
+		// Are sets correlatable? Only if the old value equals the current value.
+		// We might be able to support this, but do not currently see a lot
+		// of value
+		// (would allow you to add/remove items from sets with ratcheting but not change them)
+		return nil
+	case "":
+		fallthrough
+	case "atomic":
+		// Atomic lists are the default are not correlatable by item
+		// Ratcheting is not available on a per-index basis
+		return nil
+	default:
+		// Unrecognized list type. Assume non-correlatable.
+		return nil
+	}
+}
+
+// CachedDeepEqual is equivalent to reflect.DeepEqual, but caches the
+// results in the tree of ratchetInvocationScratch objects on the way:
+//
+// For objects and arrays, this function will make a best effort to make
+// use of past DeepEqual checks performed by this Node's children, if available.
+//
+// If a lazy computation could not be found for all children possibly due
+// to validation logic short circuiting and skipping the children, then
+// this function simply defers to reflect.DeepEqual.
+func (r *CorrelatedObject) CachedDeepEqual() (res bool) {
+	if r == nil {
+		// Uncorrelatable node is not considered equal to its old value
+		return false
+	} else if r.comparisonResult != nil {
+		return *r.comparisonResult
+	}
+
+	defer func() {
+		r.comparisonResult = &res
+	}()
+
+	if r.Value == nil && r.OldValue == nil {
+		return true
+	} else if r.Value == nil || r.OldValue == nil {
+		return false
+	}
+
+	oldAsArray, oldIsArray := r.OldValue.([]interface{})
+	newAsArray, newIsArray := r.Value.([]interface{})
+
+	oldAsMap, oldIsMap := r.OldValue.(map[string]interface{})
+	newAsMap, newIsMap := r.Value.(map[string]interface{})
+
+	// If old and new are not the same type, they are not equal
+	if (oldIsArray != newIsArray) || oldIsMap != newIsMap {
+		return false
+	}
+
+	// Objects are known to be same type of (map, slice, or primitive)
+	switch {
+	case oldIsArray:
+		// Both arrays case. oldIsArray == newIsArray
+		if len(oldAsArray) != len(newAsArray) {
+			return false
+		}
+
+		for i := range newAsArray {
+			child := r.Index(i)
+			if child == nil {
+				if r.mapList == nil {
+					// Treat non-correlatable array as a unit with reflect.DeepEqual
+					return reflect.DeepEqual(oldAsArray, newAsArray)
+				}
+
+				// If array is correlatable, but old not found. Just short circuit
+				// comparison
+				return false
+
+			} else if !child.CachedDeepEqual() {
+				// If one child is not equal the entire object is not equal
+				return false
+			}
+		}
+
+		return true
+	case oldIsMap:
+		// Both maps case. oldIsMap == newIsMap
+		if len(oldAsMap) != len(newAsMap) {
+			return false
+		}
+
+		for k := range newAsMap {
+			child := r.Key(k)
+			if child == nil {
+				// Un-correlatable child due to key change.
+				// Objects are not equal.
+				return false
+			} else if !child.CachedDeepEqual() {
+				// If one child is not equal the entire object is not equal
+				return false
+			}
+		}
+
+		return true
+
+	default:
+		// Primitive: use reflect.DeepEqual
+		return reflect.DeepEqual(r.OldValue, r.Value)
+	}
+}
+
+// Key returns the child of the receiver with the given name.
+// Returns nil if the given name is does not exist in the new object, or its
+// value is not correlatable to an old value.
+// If receiver is nil or if the new value is not an object/map, returns nil.
+func (r *CorrelatedObject) Key(field string) *CorrelatedObject {
+	if r == nil || r.Schema == nil {
+		return nil
+	} else if existing, exists := r.children[field]; exists {
+		return existing
+	}
+
+	// Find correlated old value
+	oldAsMap, okOld := r.OldValue.(map[string]interface{})
+	newAsMap, okNew := r.Value.(map[string]interface{})
+	if !okOld || !okNew {
+		return nil
+	}
+
+	oldValueForField, okOld := oldAsMap[field]
+	newValueForField, okNew := newAsMap[field]
+	if !okOld || !okNew {
+		return nil
+	}
+
+	var propertySchema Schema
+	if prop, exists := r.Schema.Properties()[field]; exists {
+		propertySchema = prop
+	} else if addP := r.Schema.AdditionalProperties(); addP != nil && addP.Schema() != nil {
+		propertySchema = addP.Schema()
+	} else {
+		return nil
+	}
+
+	if r.children == nil {
+		r.children = make(map[interface{}]*CorrelatedObject, len(newAsMap))
+	}
+
+	res := NewCorrelatedObject(newValueForField, oldValueForField, propertySchema)
+	r.children[field] = res
+	return res
+}
+
+// Index returns the child of the receiver at the given index.
+// Returns nil if the given index is out of bounds, or its value is not
+// correlatable to an old value.
+// If receiver is nil or if the new value is not an array, returns nil.
+func (r *CorrelatedObject) Index(i int) *CorrelatedObject {
+	if r == nil || r.Schema == nil {
+		return nil
+	} else if existing, exists := r.children[i]; exists {
+		return existing
+	}
+
+	asList, ok := r.Value.([]interface{})
+	if !ok || len(asList) <= i {
+		return nil
+	}
+
+	oldValueForIndex := r.correlateOldValueForChildAtNewIndex(i)
+	if oldValueForIndex == nil {
+		return nil
+	}
+	var itemSchema Schema
+	if i := r.Schema.Items(); i != nil {
+		itemSchema = i
+	} else {
+		return nil
+	}
+
+	if r.children == nil {
+		r.children = make(map[interface{}]*CorrelatedObject, len(asList))
+	}
+
+	res := NewCorrelatedObject(asList[i], oldValueForIndex, itemSchema)
+	r.children[i] = res
+	return res
+}

staging/src/k8s.io/apiserver/pkg/cel/common/equality_test.go

@@ -0,0 +1,756 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common_test
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/util/yaml"
+	"k8s.io/apiserver/pkg/cel/common"
+	"k8s.io/apiserver/pkg/cel/openapi"
+	"k8s.io/kube-openapi/pkg/validation/spec"
+)
+
+type TestCase struct {
+	Name string
+
+	// Expected old value after traversal. If nil, then the traversal should fail.
+	OldValue interface{}
+
+	// Expected value after traversal. If nil, then the traversal should fail.
+	NewValue interface{}
+
+	// Whether OldValue and NewValue are considered to be equal.
+	// Defaults to reflect.DeepEqual comparison of the two. Can be overridden to
+	// true here if the two values are not DeepEqual, but are considered equal
+	// for instance due to map-list reordering.
+	ExpectEqual bool
+
+	// Schema to provide to the correlated object
+	Schema common.Schema
+
+	// Array of field names and indexes to traverse to get to the value
+	KeyPath []interface{}
+
+	// Root object to traverse from
+	RootObject    interface{}
+	RootOldObject interface{}
+}
+
+func (c TestCase) Run() error {
+	// Create the correlated object
+	correlatedObject := common.NewCorrelatedObject(c.RootObject, c.RootOldObject, c.Schema)
+
+	// Traverse the correlated object
+	var err error
+	for _, key := range c.KeyPath {
+		if correlatedObject == nil {
+			break
+		}
+
+		switch k := key.(type) {
+		case string:
+			correlatedObject = correlatedObject.Key(k)
+		case int:
+			correlatedObject = correlatedObject.Index(k)
+		default:
+			return errors.New("key must be a string or int")
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	if correlatedObject == nil {
+		if c.OldValue != nil || c.NewValue != nil {
+			return fmt.Errorf("expected non-nil value, got nil")
+		}
+	} else {
+		// Check that the correlated object has the expected values
+		if !reflect.DeepEqual(correlatedObject.Value, c.NewValue) {
+			return fmt.Errorf("expected value %v, got %v", c.NewValue, correlatedObject.Value)
+		}
+		if !reflect.DeepEqual(correlatedObject.OldValue, c.OldValue) {
+			return fmt.Errorf("expected old value %v, got %v", c.OldValue, correlatedObject.OldValue)
+		}
+
+		// Check that the correlated object is considered equal to the expected value
+		if (c.ExpectEqual || reflect.DeepEqual(correlatedObject.Value, correlatedObject.OldValue)) != correlatedObject.CachedDeepEqual() {
+			return fmt.Errorf("expected equal, got not equal")
+		}
+	}
+
+	return nil
+}
+
+// Creates a *spec.Schema Schema by decoding the given YAML. Panics on error
+func mustSchema(source string) *openapi.Schema {
+	d := yaml.NewYAMLOrJSONDecoder(strings.NewReader(source), 4096)
+	res := &spec.Schema{}
+	if err := d.Decode(res); err != nil {
+		panic(err)
+	}
+	return &openapi.Schema{Schema: res}
+}
+
+// Creates an *unstructured by decoding the given YAML. Panics on error
+func mustUnstructured(source string) interface{} {
+	d := yaml.NewYAMLOrJSONDecoder(strings.NewReader(source), 4096)
+	var res interface{}
+	if err := d.Decode(&res); err != nil {
+		panic(err)
+	}
+	return res
+}
+
+func TestCorrelation(t *testing.T) {
+	// Tests ensure that the output of following keypath using the given
+	// schema and root objects yields the provided new value and old value.
+	// If new or old are nil, then ensures that the traversal failed due to
+	// uncorrelatable field path.
+	// Also confirms that CachedDeepEqual output is equal to expected result of
+	// reflect.DeepEqual of the new and old values.
+	cases := []TestCase{
+		{
+			Name:          "Basic Key",
+			RootObject:    mustUnstructured(`a: b`),
+			RootOldObject: mustUnstructured(`a: b`),
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{"a"},
+			NewValue: "b",
+			OldValue: "b",
+		},
+		{
+			Name:          "Atomic Array not correlatable",
+			RootObject:    mustUnstructured(`[a, b]`),
+			RootOldObject: mustUnstructured(`[a, b]`),
+			Schema: mustSchema(`
+                items:
+                  type: string
+            `),
+			KeyPath: []interface{}{1},
+		},
+		{
+			Name: "Added Key Not In Old Object",
+			RootObject: mustUnstructured(`
+                a: b
+                c: d
+            `),
+			RootOldObject: mustUnstructured(`
+                a: b
+            `),
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+                  c: { type: string }
+            `),
+			KeyPath: []interface{}{"c"},
+		},
+		{
+			Name: "Added Index Not In Old Object",
+			RootObject: mustUnstructured(`
+                - a
+                - b
+                - c
+            `),
+			RootOldObject: mustUnstructured(`
+                - a
+                - b
+            `),
+			Schema: mustSchema(`
+                items:
+                    type: string
+            `),
+			KeyPath: []interface{}{2},
+		},
+		{
+			Name: "Changed Index In Old Object not correlatable",
+			RootObject: []interface{}{
+				"a",
+				"b",
+			},
+			RootOldObject: []interface{}{
+				"a",
+				"oldB",
+			},
+			Schema: mustSchema(`
+                items:
+                    type: string
+            `),
+			KeyPath: []interface{}{1},
+		},
+		{
+			Name: "Changed Index In Nested Old Object",
+			RootObject: []interface{}{
+				"a",
+				"b",
+			},
+			RootOldObject: []interface{}{
+				"a",
+				"oldB",
+			},
+			Schema: mustSchema(`
+                items:
+                    type: string
+            `),
+			KeyPath:  []interface{}{},
+			NewValue: []interface{}{"a", "b"},
+			OldValue: []interface{}{"a", "oldB"},
+		},
+		{
+			Name: "Changed Key In Old Object",
+			RootObject: map[string]interface{}{
+				"a": "b",
+			},
+			RootOldObject: map[string]interface{}{
+				"a": "oldB",
+			},
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{"a"},
+			NewValue: "b",
+			OldValue: "oldB",
+		},
+		{
+			Name: "Replaced Key In Old Object",
+			RootObject: map[string]interface{}{
+				"a": "b",
+			},
+			RootOldObject: map[string]interface{}{
+				"b": "a",
+			},
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{},
+			NewValue: map[string]interface{}{"a": "b"},
+			OldValue: map[string]interface{}{"b": "a"},
+		},
+		{
+			Name: "Added Key In Old Object",
+			RootObject: map[string]interface{}{
+				"a": "b",
+			},
+			RootOldObject: map[string]interface{}{},
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{},
+			NewValue: map[string]interface{}{"a": "b"},
+			OldValue: map[string]interface{}{},
+		},
+		{
+			Name: "Changed list to map",
+			RootObject: map[string]interface{}{
+				"a": "b",
+			},
+			RootOldObject: []interface{}{"a", "b"},
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{},
+			NewValue: map[string]interface{}{"a": "b"},
+			OldValue: []interface{}{"a", "b"},
+		},
+		{
+			Name: "Changed string to map",
+			RootObject: map[string]interface{}{
+				"a": "b",
+			},
+			RootOldObject: "a string",
+			Schema: mustSchema(`
+                properties:
+                  a: { type: string }
+            `),
+			KeyPath:  []interface{}{},
+			NewValue: map[string]interface{}{"a": "b"},
+			OldValue: "a string",
+		},
+		{
+			Name: "Map list type",
+			RootObject: mustUnstructured(`
+                foo:
+                - bar: baz
+                  val: newBazValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - bar: fizz
+                  val: fizzValue
+                - bar: baz
+                  val: bazValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        bar:
+                          type: string
+                        val:
+                          type: string
+                    x-kubernetes-list-type: map
+                    x-kubernetes-list-map-keys:
+                      - bar
+            `),
+			KeyPath:  []interface{}{"foo", 0, "val"},
+			NewValue: "newBazValue",
+			OldValue: "bazValue",
+		},
+		{
+			Name: "Atomic list item should not correlate",
+			RootObject: mustUnstructured(`
+                foo:
+                - bar: baz
+                  val: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - bar: fizz
+                  val: fizzValue
+                - bar: baz
+                  val: barValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        bar:
+                          type: string
+                        val:
+                          type: string
+                    x-kubernetes-list-type: atomic
+            `),
+			KeyPath: []interface{}{"foo", 0, "val"},
+		},
+		{
+			Name: "Map used inside of map list type should correlate",
+			RootObject: mustUnstructured(`
+                foo:
+                - key: keyValue
+                  bar:
+                    baz: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - key: otherKeyValue
+                  bar:
+                    baz: otherOldValue
+                - key: altKeyValue
+                  bar:
+                    baz: altOldValue
+                - key: keyValue
+                  bar:
+                    baz: oldValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        key:
+                          type: string
+                        bar:
+                          type: object
+                          properties:
+                            baz:
+                              type: string
+                    x-kubernetes-list-type: map
+                    x-kubernetes-list-map-keys:
+                      - key
+            `),
+			KeyPath:  []interface{}{"foo", 0, "bar", "baz"},
+			NewValue: "newValue",
+			OldValue: "oldValue",
+		},
+		{
+			Name: "Map used inside another map should correlate",
+			RootObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                    key: otherKeyValue
+                    bar:
+                        baz: otherOldValue
+                altFoo:
+                    key: altKeyValue
+                    bar:
+                        baz: altOldValue
+                otherFoo:
+                    key: keyValue
+                    bar:
+                        baz: oldValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: object
+                    properties:
+                      key:
+                        type: string
+                      bar:
+                        type: object
+                        properties:
+                          baz:
+                            type: string
+            `),
+			KeyPath:  []interface{}{"foo", "bar"},
+			NewValue: map[string]interface{}{"baz": "newValue"},
+			OldValue: map[string]interface{}{"baz": "otherOldValue"},
+		},
+		{
+			Name: "Nested map equal to old",
+			RootObject: mustUnstructured(`
+                foo:
+                    key: newKeyValue
+                    bar:
+                        baz: value
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: value
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: object
+                    properties:
+                      key:
+                        type: string
+                      bar:
+                        type: object
+                        properties:
+                          baz:
+                            type: string
+            `),
+			KeyPath:  []interface{}{"foo", "bar"},
+			NewValue: map[string]interface{}{"baz": "value"},
+			OldValue: map[string]interface{}{"baz": "value"},
+		},
+		{
+			Name: "Re-ordered list considered equal to old value due to map keys",
+			RootObject: mustUnstructured(`
+                foo:
+                - key: keyValue
+                  bar:
+                    baz: value
+                - key: altKeyValue
+                  bar:
+                    baz: altValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - key: altKeyValue
+                  bar:
+                    baz: altValue
+                - key: keyValue
+                  bar:
+                    baz: value
+            `),
+			Schema: mustSchema(`
+                properties:
+                  foo:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        key:
+                          type: string
+                        bar:
+                          type: object
+                          properties:
+                            baz:
+                              type: string
+                    x-kubernetes-list-type: map
+                    x-kubernetes-list-map-keys:
+                      - key
+            `),
+			KeyPath: []interface{}{"foo"},
+			NewValue: mustUnstructured(`
+                - key: keyValue
+                  bar:
+                    baz: value
+                - key: altKeyValue
+                  bar:
+                    baz: altValue
+            `),
+			OldValue: mustUnstructured(`
+                - key: altKeyValue
+                  bar:
+                    baz: altValue
+                - key: keyValue
+                  bar:
+                    baz: value
+            `),
+			ExpectEqual: true,
+		},
+		{
+			Name: "Correlate unknown string key via additional properties",
+			RootObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                    key: otherKeyValue
+                    bar:
+                        baz: otherOldValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: object
+                        additionalProperties:
+                            properties:
+                                baz:
+                                    type: string
+            `),
+			KeyPath:  []interface{}{"foo", "bar", "baz"},
+			NewValue: "newValue",
+			OldValue: "otherOldValue",
+		},
+		{
+			Name: "Changed map value",
+			RootObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: oldValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: object
+                        properties:
+                            key:
+                                type: string
+                            bar:
+                                type: object
+                                properties:
+                                    baz:
+                                        type: string
+            `),
+			KeyPath: []interface{}{"foo", "bar"},
+			NewValue: mustUnstructured(`
+                baz: newValue
+            `),
+			OldValue: mustUnstructured(`
+                baz: oldValue
+            `),
+		},
+		{
+			Name: "Changed nested map value",
+			RootObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: newValue
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                    key: keyValue
+                    bar:
+                        baz: oldValue
+            `),
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: object
+                        properties:
+                            key:
+                                type: string
+                            bar:
+                                type: object
+                                properties:
+                                    baz:
+                                        type: string
+            `),
+			KeyPath: []interface{}{"foo"},
+			NewValue: mustUnstructured(`
+                key: keyValue    
+                bar:
+                  baz: newValue
+            `),
+			OldValue: mustUnstructured(`
+                key: keyValue    
+                bar:
+                  baz: oldValue
+            `),
+		},
+		{
+			Name: "unchanged list type set with atomic map values",
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: array
+                        items:
+                            type: object
+                            x-kubernetes-map-type: atomic
+                            properties:
+                                key:
+                                    type: string
+                                bar:
+                                    type: string
+                        x-kubernetes-list-type: set
+            `),
+			RootObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+			KeyPath: []interface{}{"foo"},
+			NewValue: mustUnstructured(`
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+			OldValue: mustUnstructured(`
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+		},
+		{
+			Name: "changed list type set with atomic map values",
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: array
+                        items:
+                            type: object
+                            x-kubernetes-map-type: atomic
+                            properties:
+                                key:
+                                    type: string
+                                bar:
+                                    type: string
+                        x-kubernetes-list-type: set
+            `),
+			RootObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: newValue2
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+			KeyPath: []interface{}{"foo"},
+			NewValue: mustUnstructured(`
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: newValue2
+            `),
+			OldValue: mustUnstructured(`
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+		},
+		{
+			Name: "elements of list type set with atomic map values are not correlated",
+			Schema: mustSchema(`
+                properties:
+                    foo:
+                        type: array
+                        items:
+                            type: object
+                            x-kubernetes-map-type: atomic
+                            properties:
+                                key:
+                                    type: string
+                                bar:
+                                    type: string
+                        x-kubernetes-list-type: set
+            `),
+			RootObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: newValue2
+            `),
+			RootOldObject: mustUnstructured(`
+                foo:
+                - key: key1
+                  bar: value1
+                - key: key2
+                  bar: value2
+            `),
+			KeyPath:  []interface{}{"foo", 0, "key"},
+			NewValue: nil,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.Name, func(t *testing.T) {
+			if err := c.Run(); err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+		})
+	}
+}