addressed comments:

* format issues
* watcher store now saves keys with namespace/kind/name

Author: MiguelAAe

tests/integration/godog/go.mod

@@ -9,11 +9,11 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/pflag v1.0.7
 	google.golang.org/grpc v1.73.0
-	gopkg.in/yaml.v3 v3.0.1
-	k8s.io/api v0.34.2
+	k8s.io/apiextensions-apiserver v0.34.1
 	k8s.io/apimachinery v0.34.2
 	k8s.io/client-go v0.34.2
 	sigs.k8s.io/controller-runtime v0.22.4
+	sigs.k8s.io/yaml v1.6.0
 )
 
 replace (
@@ -71,13 +71,13 @@ require (
 	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
-	k8s.io/apiextensions-apiserver v0.34.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/api v0.34.2 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect
 	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect
 	knative.dev/pkg v0.0.0-20250128013458-efddeac3ec35 // indirect
 	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect
-	sigs.k8s.io/yaml v1.6.0 // indirect
 )

tests/integration/godog/k8sclient/watcher_store.go

@@ -14,31 +14,40 @@ import (
 	"fmt"
 	"sync"
 
+	mlopsscheme "github.com/seldonio/seldon-core/operator/v2/pkg/generated/clientset/versioned/scheme"
 	"github.com/seldonio/seldon-core/operator/v2/pkg/generated/clientset/versioned/typed/mlops/v1alpha1"
 	log "github.com/sirupsen/logrus"
+	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/scheme"
 	"k8s.io/apimachinery/pkg/api/meta"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
 type WatcherStorage interface {
-	Put(runtime.Object)
-	Get(runtime.Object) (runtime.Object, bool)
-	WaitForObject(ctx context.Context, obj runtime.Object, cond ConditionFunc) error
-	WaitForKey(ctx context.Context, key string, cond ConditionFunc) error
+	WaitForObjectCondition(ctx context.Context, obj runtime.Object, cond ConditionFunc) error
+	WaitForModelCondition(ctx context.Context, modelName string, cond ConditionFunc) error
+	WaitForPipelineCondition(ctx context.Context, modelName string, cond ConditionFunc) error
 	Clear()
 	Start()
 	Stop()
 }
 
+type objectKind string
+
+const (
+	model    objectKind = "Model"
+	pipeline objectKind = "Pipeline"
+)
+
 type WatcherStore struct {
 	namespace       string
 	label           string
 	mlopsClient     v1alpha1.MlopsV1alpha1Interface
 	modelWatcher    watch.Interface
 	pipelineWatcher watch.Interface
 	logger          log.FieldLogger
+	scheme          *runtime.Scheme
 
 	mu      sync.RWMutex
 	store   map[string]runtime.Object // key: "namespace/name"
@@ -71,6 +80,11 @@ func NewWatcherStore(namespace string, label string, mlopsClient v1alpha1.MlopsV
 		return nil, fmt.Errorf("failed to create pipeline watcher: %w", err)
 	}
 
+	// Base scheme + register your CRDs
+	s := runtime.NewScheme()
+	_ = scheme.AddToScheme(s)      // core k8s types (optional but fine)
+	_ = mlopsscheme.AddToScheme(s) // <-- this is the key line for your CRDs
+
 	return &WatcherStore{
 		namespace:       namespace,
 		label:           label,
@@ -80,6 +94,7 @@ func NewWatcherStore(namespace string, label string, mlopsClient v1alpha1.MlopsV
 		logger:          logger.WithField("client", "watcher_store"),
 		store:           make(map[string]runtime.Object),
 		doneChan:        make(chan struct{}),
+		scheme:          s,
 	}, nil
 }
 
@@ -107,7 +122,7 @@ func (s *WatcherStore) Start() {
 
 				switch event.Type {
 				case watch.Added, watch.Modified:
-					s.Put(event.Object)
+					s.put(event.Object)
 				case watch.Deleted:
 					s.delete(event.Object)
 				case watch.Error:
@@ -143,7 +158,7 @@ func (s *WatcherStore) Start() {
 
 				switch event.Type {
 				case watch.Added, watch.Modified:
-					s.Put(event.Object)
+					s.put(event.Object)
 				case watch.Deleted:
 					s.delete(event.Object)
 				case watch.Error:
@@ -177,14 +192,24 @@ func (s *WatcherStore) keyFor(obj runtime.Object) (string, error) {
 
 	ns := accessor.GetNamespace()
 	if ns == "" {
-		// fall back to store namespace if the object is cluster-scoped or unset
-		ns = s.namespace
+		ns = s.namespace // or "_cluster" if you prefer
+	}
+
+	// Prefer scheme-based GVK for typed objects
+	gvks, _, err := s.scheme.ObjectKinds(obj)
+	if err != nil || len(gvks) == 0 {
+		// fallback: TypeMeta if present
+		if ta, taErr := meta.TypeAccessor(obj); taErr == nil && ta.GetKind() != "" {
+			return fmt.Sprintf("%s/%s/%s", ns, ta.GetKind(), accessor.GetName()), nil
+		}
+		return "", fmt.Errorf("failed to determine kind for %T: %w", obj, err)
 	}
 
-	return fmt.Sprintf("%s/%s", ns, accessor.GetName()), nil
+	kind := gvks[0].Kind
+	return fmt.Sprintf("%s/%s/%s", ns, kind, accessor.GetName()), nil
 }
 
-func (s *WatcherStore) Put(obj runtime.Object) {
+func (s *WatcherStore) put(obj runtime.Object) {
 	if obj == nil {
 		return
 	}
@@ -201,7 +226,7 @@ func (s *WatcherStore) Put(obj runtime.Object) {
 	s.notifyWaiters(key, obj)
 }
 
-func (s *WatcherStore) Get(obj runtime.Object) (runtime.Object, bool) {
+func (s *WatcherStore) get(obj runtime.Object) (runtime.Object, bool) {
 	if obj == nil {
 		return nil, false
 	}
@@ -242,7 +267,7 @@ func (s *WatcherStore) delete(obj runtime.Object) {
 	s.notifyWaiters(key, nil)
 }
 
-func (s *WatcherStore) WaitForObject(ctx context.Context, obj runtime.Object, cond ConditionFunc) error {
+func (s *WatcherStore) WaitForObjectCondition(ctx context.Context, obj runtime.Object, cond ConditionFunc) error {
 	key, err := s.keyFor(obj)
 	if err != nil {
 		return err
@@ -283,10 +308,17 @@ func (s *WatcherStore) WaitForObject(ctx context.Context, obj runtime.Object, co
 		return err
 	}
 }
+func (s *WatcherStore) WaitForModelCondition(ctx context.Context, modelName string, cond ConditionFunc) error {
+	key := fmt.Sprintf("%s/%s/%s", s.namespace, model, modelName)
+	return s.waitForKey(ctx, key, cond)
+}
+
+func (s *WatcherStore) WaitForPipelineCondition(ctx context.Context, pipelineName string, cond ConditionFunc) error {
+	key := fmt.Sprintf("%s/%s/%s", s.namespace, pipeline, pipelineName)
+	return s.waitForKey(ctx, key, cond)
+}
 
-func (s *WatcherStore) WaitForKey(ctx context.Context, key string, cond ConditionFunc) error {
-	// build key
-	key = fmt.Sprintf("%s/%s", s.namespace, key)
+func (s *WatcherStore) waitForKey(ctx context.Context, key string, cond ConditionFunc) error {
 
 	// Fast path: check current state
 	s.mu.RLock()

tests/integration/godog/steps/custom_model_steps.go

@@ -68,7 +68,7 @@ func (m *Model) deleteModel(ctx context.Context, model string) error {
 }
 
 func (m *Model) waitForModelNameReady(ctx context.Context, name string) error {
-	return m.watcherStorage.WaitForKey(
+	return m.watcherStorage.WaitForModelCondition(
 		ctx,
 		name,
 		assertions.ModelReady)

tests/integration/godog/steps/model_steps.go

@@ -268,7 +268,7 @@ func (m *Model) ApplyModel(k *k8sclient.K8sClient) error {
 }
 
 func (m *Model) waitForModelReady(ctx context.Context) error {
-	return m.watcherStorage.WaitForObject(
+	return m.watcherStorage.WaitForObjectCondition(
 		ctx,
 		m.model,               // the k8s object being watched
 		assertions.ModelReady, // predicate from steps/assertions

tests/integration/godog/steps/pipeline_steps.go

@@ -87,17 +87,15 @@ func (p *Pipeline) applyScenarioLabel() {
 }
 
 func (p *Pipeline) waitForPipelineNameReady(ctx context.Context, name string) error {
-
-	return p.watcherStorage.WaitForKey(
+	return p.watcherStorage.WaitForPipelineCondition(
 		ctx,
 		name,
 		assertions.PipelineReady,
 	)
 }
 
 func (p *Pipeline) waitForPipelineReady(ctx context.Context) error {
-
-	return p.watcherStorage.WaitForObject(
+	return p.watcherStorage.WaitForObjectCondition(
 		ctx,
 		p.pipeline,
 		assertions.PipelineReady,