test: migrate fake and kubernetes v2 provider suites

e2e/suites/provider/cases/fake/operational_v2.go

@@ -0,0 +1,273 @@
+/*
+Copyright © The ESO 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
+
+    https://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 fake
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+	esv1alpha1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1alpha1"
+	genv1alpha1 "github.com/external-secrets/external-secrets/apis/generators/v1alpha1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[fake] v2 operational", Serial, Label("fake", "v2", "operational"), func() {
+	f := framework.New("eso-fake-v2-operational")
+	prov := NewProviderV2(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+	})
+
+	DescribeTable("external secret operational behavior",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.NamespacedProviderUnavailable(f, newFakeOperationalExternalSecretHarness(f, prov), "fake-operational-unavailable", "recovered")),
+		Entry(common.NamespacedProviderRestart(f, newFakeOperationalExternalSecretHarness(f, prov), "fake-operational-restart", "restarted")),
+		Entry(common.ClusterProviderUnavailable(f, newFakeOperationalExternalSecretHarness(f, prov), "fake-operational-cluster-unavailable", "cluster-recovered", esv1.AuthenticationScopeManifestNamespace)),
+		Entry(common.ClusterProviderRestart(f, newFakeOperationalExternalSecretHarness(f, prov), "fake-operational-cluster-restart", "cluster-restarted", esv1.AuthenticationScopeManifestNamespace)),
+	)
+
+	DescribeTable("push secret operational behavior",
+		framework.TableFuncWithPushSecret(f, prov, nil),
+		Entry(common.NamespacedPushSecretUnavailable(f, newFakeOperationalPushHarness(f, prov))),
+		Entry(common.ClusterProviderPushUnavailable(f, newFakeOperationalPushHarness(f, prov), esv1.AuthenticationScopeManifestNamespace)),
+	)
+
+	It("reuses one backend connection across many namespaced fake Provider consumers", func() {
+		const consumerCount = 10
+
+		for i := 0; i < consumerCount; i++ {
+			remoteKey := fmt.Sprintf("fake-operational-consumer-%d", i)
+			targetName := fmt.Sprintf("fake-operational-consumer-target-%d", i)
+			expectedValue := fmt.Sprintf("value-%d", i)
+
+			prov.CreateSecret(remoteKey, framework.SecretEntry{
+				Value: fmt.Sprintf(`{"value":"%s"}`, expectedValue),
+			})
+
+			Expect(f.CreateObjectWithRetry(&esv1.ExternalSecret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("fake-operational-consumer-es-%d", i),
+					Namespace: f.Namespace.Name,
+				},
+				Spec: esv1.ExternalSecretSpec{
+					RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+					SecretStoreRef: esv1.SecretStoreRef{
+						Name: f.Namespace.Name,
+						Kind: esv1.ProviderStoreKindStr,
+					},
+					Target: esv1.ExternalSecretTarget{
+						Name: targetName,
+					},
+					Data: []esv1.ExternalSecretData{{
+						SecretKey: "value",
+						RemoteRef: esv1.ExternalSecretDataRemoteRef{
+							Key:      remoteKey,
+							Property: "value",
+						},
+					}},
+				},
+			})).To(Succeed())
+
+			_, err := f.WaitForSecretValue(f.Namespace.Name, targetName, &corev1.Secret{
+				Type: corev1.SecretTypeOpaque,
+				Data: map[string][]byte{
+					"value": []byte(expectedValue),
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		Eventually(func(g Gomega) {
+			metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+			g.Expect(err).NotTo(HaveOccurred())
+
+			total := frameworkv2.SumMetricValues(metrics, "grpc_pool_connections_total", map[string]string{
+				"address": frameworkv2.ProviderAddress("fake"),
+			})
+			g.Expect(total).To(BeNumerically(">=", 1))
+			g.Expect(total).To(BeNumerically("<=", 2), "expected bounded connection reuse for one backend")
+			g.Expect(total).To(BeNumerically("<", consumerCount), "expected fewer pooled connections than consumers")
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+	})
+
+	It("reuses backend connections across multiple fake Provider resources that share one backend", func() {
+		const providerCount = 4
+
+		for i := 0; i < providerCount; i++ {
+			providerName := fmt.Sprintf("fake-fanout-provider-%d", i)
+			remoteKey := fmt.Sprintf("fake-fanout-remote-%d", i)
+			expectedValue := fmt.Sprintf("fanout-%d", i)
+			targetName := fmt.Sprintf("fake-fanout-target-%d", i)
+
+			frameworkv2.CreateProviderConnection(
+				f,
+				f.Namespace.Name,
+				providerName,
+				frameworkv2.ProviderAddress("fake"),
+				fakeProviderAPIVersion,
+				fakeProviderKind,
+				f.Namespace.Name,
+				"",
+			)
+			frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, providerName, defaultV2WaitTimeout)
+
+			prov.CreateSecret(remoteKey, framework.SecretEntry{
+				Value: fmt.Sprintf(`{"value":"%s"}`, expectedValue),
+			})
+
+			Expect(f.CreateObjectWithRetry(&esv1.ExternalSecret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("fake-fanout-es-%d", i),
+					Namespace: f.Namespace.Name,
+				},
+				Spec: esv1.ExternalSecretSpec{
+					RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+					SecretStoreRef: esv1.SecretStoreRef{
+						Name: providerName,
+						Kind: esv1.ProviderStoreKindStr,
+					},
+					Target: esv1.ExternalSecretTarget{
+						Name: targetName,
+					},
+					Data: []esv1.ExternalSecretData{{
+						SecretKey: "value",
+						RemoteRef: esv1.ExternalSecretDataRemoteRef{
+							Key:      remoteKey,
+							Property: "value",
+						},
+					}},
+				},
+			})).To(Succeed())
+
+			_, err := f.WaitForSecretValue(f.Namespace.Name, targetName, &corev1.Secret{
+				Type: corev1.SecretTypeOpaque,
+				Data: map[string][]byte{
+					"value": []byte(expectedValue),
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		Eventually(func(g Gomega) {
+			metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+			g.Expect(err).NotTo(HaveOccurred())
+
+			total := frameworkv2.SumMetricValues(metrics, "grpc_pool_connections_total", map[string]string{
+				"address": frameworkv2.ProviderAddress("fake"),
+			})
+			g.Expect(total).To(BeNumerically(">=", 1))
+			g.Expect(total).To(BeNumerically("<=", 2), "expected bounded connection reuse across shared backend fanout")
+			g.Expect(total).To(BeNumerically("<", providerCount), "expected fewer pooled connections than Provider resources")
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+	})
+
+	It("recovers generator-backed push traffic after fake provider outage", func() {
+		const (
+			generatorName   = "fake-operational-generator"
+			pushSecretName  = "fake-operational-generator-push"
+			remoteSecretKey = "fake-operational-generator-remote"
+		)
+
+		generator := &genv1alpha1.Fake{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: genv1alpha1.Group + "/" + genv1alpha1.Version,
+				Kind:       genv1alpha1.FakeKind,
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      generatorName,
+				Namespace: f.Namespace.Name,
+			},
+			Spec: genv1alpha1.FakeSpec{
+				Data: map[string]string{
+					"value": "before-outage",
+				},
+			},
+		}
+		Expect(f.CreateObjectWithRetry(generator)).To(Succeed())
+
+		pushSecret := &esv1alpha1.PushSecret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      pushSecretName,
+				Namespace: f.Namespace.Name,
+			},
+			Spec: esv1alpha1.PushSecretSpec{
+				RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+				SecretStoreRefs: []esv1alpha1.PushSecretStoreRef{{
+					Name:       f.Namespace.Name,
+					Kind:       esv1.ProviderStoreKindStr,
+					APIVersion: esv1.SchemeGroupVersion.String(),
+				}},
+				Selector: esv1alpha1.PushSecretSelector{
+					GeneratorRef: &esv1.GeneratorRef{
+						APIVersion: genv1alpha1.Group + "/" + genv1alpha1.Version,
+						Kind:       genv1alpha1.FakeKind,
+						Name:       generatorName,
+					},
+				},
+				Data: []esv1alpha1.PushSecretData{{
+					Match: esv1alpha1.PushSecretMatch{
+						SecretKey: "value",
+						RemoteRef: esv1alpha1.PushSecretRemoteRef{
+							RemoteKey: remoteSecretKey,
+							Property:  "value",
+						},
+					},
+				}},
+			},
+		}
+		Expect(f.CreateObjectWithRetry(pushSecret)).To(Succeed())
+
+		commonWaitForPushSecretReady(f, f.Namespace.Name, pushSecretName, corev1.ConditionTrue)
+		waitForPushedValueViaExternalSecret(f, esv1.SecretStoreRef{
+			Name: f.Namespace.Name,
+			Kind: esv1.ProviderStoreKindStr,
+		}, remoteSecretKey, "before-outage")
+
+		DeferCleanup(func() {
+			frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 1, defaultV2WaitTimeout)
+			frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+		})
+		frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 0, defaultV2WaitTimeout)
+		commonWaitForPushSecretReady(f, f.Namespace.Name, pushSecretName, corev1.ConditionFalse)
+
+		var updated genv1alpha1.Fake
+		Expect(f.CRClient.Get(context.Background(), types.NamespacedName{Name: generatorName, Namespace: f.Namespace.Name}, &updated)).To(Succeed())
+		updated.Spec.Data["value"] = "after-outage"
+		Expect(f.CRClient.Update(context.Background(), &updated)).To(Succeed())
+
+		frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 1, defaultV2WaitTimeout)
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+		commonWaitForPushSecretReady(f, f.Namespace.Name, pushSecretName, corev1.ConditionTrue)
+		waitForPushedValueViaExternalSecret(f, esv1.SecretStoreRef{
+			Name: f.Namespace.Name,
+			Kind: esv1.ProviderStoreKindStr,
+		}, remoteSecretKey, "after-outage")
+	})
+})

e2e/suites/provider/cases/fake/provider.go

@@ -17,19 +17,17 @@ limitations under the License.
 package fake
 
 import (
-	"encoding/json"
-
-	// nolint
-	. "github.com/onsi/ginkgo/v2"
-
-	// nolint
-	. "github.com/onsi/gomega"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/external-secrets/external-secrets-e2e/framework"
 	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	// nolint
+	. "github.com/onsi/ginkgo/v2"
+	// nolint
+	. "github.com/onsi/gomega"
 )
 
 type Provider struct {
@@ -53,9 +51,7 @@ func (s *Provider) CreateSecret(key string, val framework.SecretEntry) {
 	Expect(err).ToNot(HaveOccurred())
 	base := store.DeepCopy()
 
-	mapData := make(map[string]string)
-	_ = json.Unmarshal([]byte(val.Value), &mapData)
-	store.Spec.Provider.Fake.Data = append(store.Spec.Provider.Fake.Data, esv1.FakeProviderData{
+	store.Spec.Provider.Fake.Data = upsertFakeProviderData(store.Spec.Provider.Fake.Data, esv1.FakeProviderData{
 		Key:   key,
 		Value: val.Value,
 	})
@@ -75,13 +71,7 @@ func (s *Provider) DeleteSecret(key string) {
 	}, &store)
 	Expect(err).ToNot(HaveOccurred())
 	base := store.DeepCopy()
-	data := make([]esv1.FakeProviderData, 0)
-	for _, v := range store.Spec.Provider.Fake.Data {
-		if v.Key != key {
-			data = append(data, v)
-		}
-	}
-	store.Spec.Provider.Fake.Data = data
+	store.Spec.Provider.Fake.Data = removeFakeProviderData(store.Spec.Provider.Fake.Data, key, "")
 	err = s.framework.CRClient.Patch(GinkgoT().Context(), &store, client.MergeFrom(base))
 	Expect(err).ToNot(HaveOccurred())
 }
@@ -105,3 +95,24 @@ func (s *Provider) CreateStore() {
 	err := s.framework.CRClient.Create(GinkgoT().Context(), fakeStore)
 	Expect(err).ToNot(HaveOccurred())
 }
+
+func upsertFakeProviderData(data []esv1.FakeProviderData, entry esv1.FakeProviderData) []esv1.FakeProviderData {
+	for i := range data {
+		if data[i].Key == entry.Key && data[i].Version == entry.Version {
+			data[i] = entry
+			return data
+		}
+	}
+	return append(data, entry)
+}
+
+func removeFakeProviderData(data []esv1.FakeProviderData, key, version string) []esv1.FakeProviderData {
+	filtered := make([]esv1.FakeProviderData, 0, len(data))
+	for _, entry := range data {
+		if entry.Key == key && entry.Version == version {
+			continue
+		}
+		filtered = append(filtered, entry)
+	}
+	return filtered
+}

e2e/suites/provider/cases/fake/provider_v2.go

@@ -0,0 +1,519 @@
+/*
+Copyright © The ESO 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
+
+    https://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 fake
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+	esv1alpha1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1alpha1"
+	fakev2alpha1 "github.com/external-secrets/external-secrets/apis/provider/fake/v2alpha1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+const (
+	fakeProviderAPIVersion   = "provider.external-secrets.io/v2alpha1"
+	fakeProviderKind         = "Fake"
+	defaultV2WaitTimeout     = 3 * time.Minute
+	defaultV2PollInterval    = 2 * time.Second
+	defaultV2RefreshInterval = 10 * time.Second
+)
+
+var _ = Describe("[fake] v2 namespaced provider", Label("fake", "v2", "namespaced-provider"), func() {
+	f := framework.New("eso-fake-v2-provider")
+	prov := NewProviderV2(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+	})
+
+	DescribeTable("namespaced provider",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.FakeProviderSync(f)),
+		Entry(common.FakeProviderRefresh(f)),
+		Entry(common.FakeProviderFind(f)),
+		Entry(common.StatusNotUpdatedAfterSuccessfulSync(f)),
+	)
+})
+
+var _ = Describe("[fake] v2 cluster provider", Label("fake", "v2", "cluster-provider"), func() {
+	f := framework.New("eso-fake-v2-clusterprovider")
+	prov := NewProviderV2(f)
+	harness := newFakeClusterProviderExternalSecretHarness(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+	})
+
+	DescribeTable("cluster provider external secrets",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.ClusterProviderManifestNamespace(f, harness)),
+		Entry(common.ClusterProviderProviderNamespace(f, harness)),
+		Entry(common.ClusterProviderDeniedByConditions(f, harness)),
+	)
+})
+
+var _ = Describe("[fake] v2 push secret", Label("fake", "v2", "push-secret"), func() {
+	f := framework.New("eso-fake-v2-push")
+	prov := NewProviderV2(f)
+	harness := newFakeClusterProviderPushHarness(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+	})
+
+	DescribeTable("push secret",
+		framework.TableFuncWithPushSecret(f, prov, nil),
+		// The fake backend stores raw pushed values in provider memory, so provider-
+		// specific metadata validation such as namespaced remoteNamespace rejection
+		// is not meaningful here.
+		Entry(fakePushSecretImplicitProviderKind(f)),
+		Entry(common.ClusterProviderPushManifestNamespace(f, harness)),
+		Entry(common.ClusterProviderPushProviderNamespace(f, harness)),
+		Entry(common.ClusterProviderPushDeniedByConditions(f, harness)),
+	)
+})
+
+type ProviderV2 struct {
+	framework *framework.Framework
+}
+
+func NewProviderV2(f *framework.Framework) *ProviderV2 {
+	prov := &ProviderV2{
+		framework: f,
+	}
+	BeforeEach(prov.BeforeEach)
+	return prov
+}
+
+func (s *ProviderV2) BeforeEach() {
+	if !framework.IsV2ProviderMode() {
+		return
+	}
+
+	frameworkv2.ScaleDeploymentBySelectorAndWait(s.framework, fakeBackendTarget(), 1, defaultV2WaitTimeout)
+	s.createStore()
+	frameworkv2.WaitForProviderConnectionReady(s.framework, s.framework.Namespace.Name, s.framework.Namespace.Name, defaultV2WaitTimeout)
+}
+
+func (s *ProviderV2) CreateSecret(key string, val framework.SecretEntry) {
+	s.updateStore(func(fake *fakev2alpha1.Fake) {
+		fake.Spec.Data = upsertFakeProviderData(fake.Spec.Data, esv1.FakeProviderData{
+			Key:   key,
+			Value: val.Value,
+		})
+	})
+}
+
+func (s *ProviderV2) DeleteSecret(key string) {
+	s.updateStore(func(fake *fakev2alpha1.Fake) {
+		fake.Spec.Data = removeFakeProviderData(fake.Spec.Data, key, "")
+	})
+}
+
+func (s *ProviderV2) createStore() {
+	createFakeProviderConfig(s.framework, s.framework.Namespace.Name, s.framework.Namespace.Name)
+	frameworkv2.CreateProviderConnection(
+		s.framework,
+		s.framework.Namespace.Name,
+		s.framework.Namespace.Name,
+		frameworkv2.ProviderAddress("fake"),
+		fakeProviderAPIVersion,
+		fakeProviderKind,
+		s.framework.Namespace.Name,
+		"",
+	)
+}
+
+func (s *ProviderV2) updateStore(mutate func(*fakev2alpha1.Fake)) {
+	updateFakeProviderConfig(s.framework, s.framework.Namespace.Name, s.framework.Namespace.Name, mutate)
+}
+
+func fakeBackendTarget() frameworkv2.BackendTarget {
+	return frameworkv2.BackendTarget{
+		Namespace:        frameworkv2.ProviderNamespace,
+		PodLabelSelector: "app.kubernetes.io/name=external-secrets-provider-fake",
+	}
+}
+
+func (s *ProviderV2) prepareNamespacedOperationalRuntime() *common.OperationalRuntime {
+	return &common.OperationalRuntime{
+		Provider: s,
+		ProviderRef: esv1.SecretStoreRef{
+			Name: s.framework.Namespace.Name,
+			Kind: esv1.ProviderStoreKindStr,
+		},
+		DefaultRemoteNamespace: s.framework.Namespace.Name,
+		WaitForRemoteSecret: func(_, name, _ string, expectedValue string) {
+			waitForPushedValueViaExternalSecret(s.framework, esv1.SecretStoreRef{
+				Name: s.framework.Namespace.Name,
+				Kind: esv1.ProviderStoreKindStr,
+			}, name, expectedValue)
+		},
+		MakeUnavailable: func() {
+			frameworkv2.ScaleDeploymentBySelector(s.framework, fakeBackendTarget(), 0)
+		},
+		RestoreAvailability: func() {
+			frameworkv2.ScaleDeploymentBySelector(s.framework, fakeBackendTarget(), 1)
+		},
+		RestartBackend: func() {
+			frameworkv2.DeleteOneProviderPodBySelector(s.framework, fakeBackendTarget())
+		},
+	}
+}
+
+type fakeClusterProviderScenario struct {
+	f                    *framework.Framework
+	namePrefix           string
+	authScope            esv1.AuthenticationScope
+	fakeConfigNamespace  string
+	providerRefNamespace string
+	configName           string
+}
+
+func newFakeClusterProviderScenario(f *framework.Framework, prefix string, authScope esv1.AuthenticationScope) *fakeClusterProviderScenario {
+	providerNamespace := f.Namespace.Name
+	if authScope == esv1.AuthenticationScopeProviderNamespace {
+		providerNamespace = common.CreateProviderCaseNamespace(f, prefix+"-provider", defaultV2PollInterval)
+	}
+
+	s := &fakeClusterProviderScenario{
+		f:                    f,
+		namePrefix:           fmt.Sprintf("%s-%s", f.Namespace.Name, prefix),
+		authScope:            authScope,
+		fakeConfigNamespace:  fakeConfigNamespaceForAuthScope(authScope, f.Namespace.Name, providerNamespace),
+		providerRefNamespace: providerReferenceNamespace(authScope, providerNamespace),
+		configName:           fmt.Sprintf("%s-config", prefix),
+	}
+	createFakeProviderConfig(s.f, s.fakeConfigNamespace, s.configName)
+	return s
+}
+
+func (s *fakeClusterProviderScenario) createClusterProvider(conditions []esv1.ClusterSecretStoreCondition) string {
+	clusterProviderName := fmt.Sprintf("%s-cluster-provider", s.namePrefix)
+	frameworkv2.CreateClusterProviderConnection(
+		s.f,
+		clusterProviderName,
+		frameworkv2.ProviderAddress("fake"),
+		fakeProviderAPIVersion,
+		fakeProviderKind,
+		s.configName,
+		s.providerRefNamespace,
+		s.authScope,
+		conditions,
+	)
+	return clusterProviderName
+}
+
+func (s *fakeClusterProviderScenario) CreateSecret(key string, val framework.SecretEntry) {
+	updateFakeProviderConfig(s.f, s.fakeConfigNamespace, s.configName, func(fake *fakev2alpha1.Fake) {
+		fake.Spec.Data = upsertFakeProviderData(fake.Spec.Data, esv1.FakeProviderData{
+			Key:   key,
+			Value: val.Value,
+		})
+	})
+}
+
+func (s *fakeClusterProviderScenario) DeleteSecret(key string) {
+	updateFakeProviderConfig(s.f, s.fakeConfigNamespace, s.configName, func(fake *fakev2alpha1.Fake) {
+		fake.Spec.Data = removeFakeProviderData(fake.Spec.Data, key, "")
+	})
+}
+
+func newFakeClusterProviderExternalSecretHarness(f *framework.Framework) common.ClusterProviderExternalSecretHarness {
+	return common.ClusterProviderExternalSecretHarness{
+		Prepare: func(tc *framework.TestCase, cfg common.ClusterProviderConfig) *common.ClusterProviderExternalSecretRuntime {
+			s := newFakeClusterProviderScenario(f, cfg.Name, cfg.AuthScope)
+			clusterProviderName := s.createClusterProvider(cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.ClusterProviderExternalSecretRuntime{
+				ClusterProviderName: clusterProviderName,
+				Provider:            s,
+			}
+		},
+	}
+}
+
+func newFakeClusterProviderPushHarness(f *framework.Framework) common.ClusterProviderPushHarness {
+	return common.ClusterProviderPushHarness{
+		Prepare: func(tc *framework.TestCase, cfg common.ClusterProviderConfig) *common.ClusterProviderPushRuntime {
+			s := newFakeClusterProviderScenario(f, cfg.Name, cfg.AuthScope)
+			clusterProviderName := s.createClusterProvider(cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.ClusterProviderPushRuntime{
+				ClusterProviderName:    clusterProviderName,
+				DefaultRemoteNamespace: s.fakeConfigNamespace,
+				WaitForRemoteSecretValue: func(_, name, _ string, expectedValue string) {
+					waitForPushedValueViaExternalSecret(f, esv1.SecretStoreRef{
+						Name: clusterProviderName,
+						Kind: esv1.ClusterProviderStoreKindStr,
+					}, name, expectedValue)
+				},
+			}
+		},
+	}
+}
+
+func newFakeOperationalExternalSecretHarness(f *framework.Framework, prov *ProviderV2) common.OperationalExternalSecretHarness {
+	return common.OperationalExternalSecretHarness{
+		PrepareNamespaced: func(_ *framework.TestCase) *common.OperationalRuntime {
+			return prov.prepareNamespacedOperationalRuntime()
+		},
+		PrepareCluster: func(_ *framework.TestCase, cfg common.ClusterProviderConfig) *common.OperationalRuntime {
+			s := newFakeClusterProviderScenario(f, cfg.Name, cfg.AuthScope)
+			clusterProviderName := s.createClusterProvider(cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.OperationalRuntime{
+				Provider: s,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: clusterProviderName,
+					Kind: esv1.ClusterProviderStoreKindStr,
+				},
+				DefaultRemoteNamespace: s.fakeConfigNamespace,
+				WaitForRemoteSecret: func(_, name, _ string, expectedValue string) {
+					waitForPushedValueViaExternalSecret(f, esv1.SecretStoreRef{
+						Name: clusterProviderName,
+						Kind: esv1.ClusterProviderStoreKindStr,
+					}, name, expectedValue)
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 0, defaultV2WaitTimeout)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 1, defaultV2WaitTimeout)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, fakeBackendTarget())
+				},
+			}
+		},
+	}
+}
+
+func newFakeOperationalPushHarness(f *framework.Framework, prov *ProviderV2) common.OperationalPushSecretHarness {
+	return common.OperationalPushSecretHarness{
+		PrepareNamespaced: func(_ *framework.TestCase) *common.OperationalRuntime {
+			return prov.prepareNamespacedOperationalRuntime()
+		},
+		PrepareCluster: func(_ *framework.TestCase, cfg common.ClusterProviderConfig) *common.OperationalRuntime {
+			s := newFakeClusterProviderScenario(f, cfg.Name, cfg.AuthScope)
+			clusterProviderName := s.createClusterProvider(cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.OperationalRuntime{
+				Provider: s,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: clusterProviderName,
+					Kind: esv1.ClusterProviderStoreKindStr,
+				},
+				DefaultRemoteNamespace: s.fakeConfigNamespace,
+				WaitForRemoteSecret: func(_, name, _ string, expectedValue string) {
+					waitForPushedValueViaExternalSecret(f, esv1.SecretStoreRef{
+						Name: clusterProviderName,
+						Kind: esv1.ClusterProviderStoreKindStr,
+					}, name, expectedValue)
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 0, defaultV2WaitTimeout)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelectorAndWait(f, fakeBackendTarget(), 1, defaultV2WaitTimeout)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, fakeBackendTarget())
+				},
+			}
+		},
+	}
+}
+
+func fakePushSecretImplicitProviderKind(f *framework.Framework) (string, func(*framework.TestCase)) {
+	return "[fake] should support namespaced Provider refs when push kind is omitted", func(tc *framework.TestCase) {
+		tc.PushSecretSource = &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "fake-push-implicit-kind-source",
+				Namespace: f.Namespace.Name,
+			},
+			Data: map[string][]byte{
+				"value": []byte("implicit-kind-value"),
+			},
+		}
+		tc.PushSecret.ObjectMeta.Name = "fake-push-implicit-kind"
+		tc.PushSecret.Spec.SecretStoreRefs[0].Kind = ""
+		tc.PushSecret.Spec.Selector = esv1alpha1.PushSecretSelector{
+			Secret: &esv1alpha1.PushSecretSecret{
+				Name: tc.PushSecretSource.Name,
+			},
+		}
+		tc.PushSecret.Spec.Data = []esv1alpha1.PushSecretData{{
+			Match: esv1alpha1.PushSecretMatch{
+				SecretKey: "value",
+				RemoteRef: esv1alpha1.PushSecretRemoteRef{
+					RemoteKey: "fake-push-implicit-kind-remote",
+					Property:  "value",
+				},
+			},
+		}}
+		tc.VerifyPushSecretOutcome = func(ps *esv1alpha1.PushSecret, _ esv1.SecretsClient) {
+			commonWaitForPushSecretReady(tc.Framework, ps.Namespace, ps.Name, corev1.ConditionTrue)
+			waitForPushedValueViaExternalSecret(tc.Framework, esv1.SecretStoreRef{
+				Name: f.Namespace.Name,
+				Kind: esv1.ProviderStoreKindStr,
+			}, "fake-push-implicit-kind-remote", "implicit-kind-value")
+		}
+	}
+}
+
+func commonWaitForPushSecretReady(f *framework.Framework, namespace, name string, status corev1.ConditionStatus) {
+	Eventually(func(g Gomega) {
+		var ps esv1alpha1.PushSecret
+		g.Expect(f.CRClient.Get(context.Background(), types.NamespacedName{
+			Name:      name,
+			Namespace: namespace,
+		}, &ps)).To(Succeed())
+
+		for _, condition := range ps.Status.Conditions {
+			if condition.Type == esv1alpha1.PushSecretReady && condition.Status == status {
+				return
+			}
+		}
+		g.Expect(false).To(BeTrue())
+	}, time.Minute, 5*time.Second).Should(Succeed())
+}
+
+func waitForPushedValueViaExternalSecret(f *framework.Framework, storeRef esv1.SecretStoreRef, remoteKey, expectedValue string) {
+	externalSecretName := fmt.Sprintf("fake-readback-%s", remoteKey)
+	targetName := fmt.Sprintf("%s-target", externalSecretName)
+
+	externalSecret := &esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      externalSecretName,
+			Namespace: f.Namespace.Name,
+		},
+		Spec: esv1.ExternalSecretSpec{
+			RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+			SecretStoreRef:  storeRef,
+			Target: esv1.ExternalSecretTarget{
+				Name: targetName,
+			},
+			Data: []esv1.ExternalSecretData{{
+				SecretKey: "value",
+				RemoteRef: esv1.ExternalSecretDataRemoteRef{
+					Key: remoteKey,
+				},
+			}},
+		},
+	}
+	Expect(createOrUpdateReadbackExternalSecret(context.Background(), f, externalSecret)).To(Succeed())
+
+	DeferCleanup(func() {
+		err := f.CRClient.Delete(context.Background(), externalSecret)
+		if err != nil && !apierrors.IsNotFound(err) {
+			Expect(err).ToNot(HaveOccurred())
+		}
+	})
+
+	_, err := f.WaitForSecretValue(f.Namespace.Name, targetName, &corev1.Secret{
+		Type: corev1.SecretTypeOpaque,
+		Data: map[string][]byte{
+			"value": []byte(expectedValue),
+		},
+	})
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func createOrUpdateReadbackExternalSecret(ctx context.Context, f *framework.Framework, externalSecret *esv1.ExternalSecret) error {
+	if err := f.CreateObjectWithRetryContext(ctx, externalSecret); err != nil {
+		return err
+	}
+
+	var existing esv1.ExternalSecret
+	if err := f.CRClient.Get(ctx, client.ObjectKeyFromObject(externalSecret), &existing); err != nil {
+		return err
+	}
+	if reflect.DeepEqual(existing.Spec, externalSecret.Spec) {
+		return nil
+	}
+
+	externalSecret.SetResourceVersion(existing.GetResourceVersion())
+	externalSecret.SetUID(existing.GetUID())
+	return f.CRClient.Update(ctx, externalSecret)
+}
+
+func createFakeProviderConfig(f *framework.Framework, namespace, name string) {
+	Expect(f.CRClient.Create(context.Background(), &fakev2alpha1.Fake{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       fakeProviderKind,
+			APIVersion: fakeProviderAPIVersion,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: esv1.FakeProvider{
+			Data: []esv1.FakeProviderData{},
+		},
+	})).To(Succeed())
+}
+
+func updateFakeProviderConfig(f *framework.Framework, namespace, name string, mutate func(*fakev2alpha1.Fake)) {
+	var fake fakev2alpha1.Fake
+	Expect(f.CRClient.Get(context.Background(), types.NamespacedName{
+		Name:      name,
+		Namespace: namespace,
+	}, &fake)).To(Succeed())
+	base := fake.DeepCopy()
+	mutate(&fake)
+	Expect(f.CRClient.Patch(context.Background(), &fake, client.MergeFrom(base))).To(Succeed())
+}
+
+func providerReferenceNamespace(authScope esv1.AuthenticationScope, providerNamespace string) string {
+	if authScope == esv1.AuthenticationScopeProviderNamespace {
+		return providerNamespace
+	}
+	return ""
+}
+
+func fakeConfigNamespaceForAuthScope(authScope esv1.AuthenticationScope, manifestNamespace, providerNamespace string) string {
+	if authScope == esv1.AuthenticationScopeProviderNamespace {
+		return providerNamespace
+	}
+	return manifestNamespace
+}

e2e/suites/provider/cases/fake/provider_v2_test.go

@@ -0,0 +1,189 @@
+/*
+Copyright © The ESO 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
+
+    https://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 fake
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1api "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+)
+
+func TestFakeBackendTargetUsesProviderNamespaceAndSelector(t *testing.T) {
+	target := fakeBackendTarget()
+	if target.Namespace != frameworkv2.ProviderNamespace {
+		t.Fatalf("expected provider namespace %q, got %q", frameworkv2.ProviderNamespace, target.Namespace)
+	}
+	if target.PodLabelSelector != "app.kubernetes.io/name=external-secrets-provider-fake" {
+		t.Fatalf("unexpected selector %q", target.PodLabelSelector)
+	}
+}
+
+func TestUpsertFakeProviderDataReplacesMatchingEntry(t *testing.T) {
+	input := []esv1.FakeProviderData{
+		{Key: "other", Value: "untouched"},
+		{Key: "remote", Value: "old", Version: "v1"},
+	}
+
+	got := upsertFakeProviderData(input, esv1.FakeProviderData{
+		Key:     "remote",
+		Value:   "new",
+		Version: "v1",
+	})
+
+	want := []esv1.FakeProviderData{
+		{Key: "other", Value: "untouched"},
+		{Key: "remote", Value: "new", Version: "v1"},
+	}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Fatalf("upsertFakeProviderData mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestRemoveFakeProviderDataRemovesOnlyExactMatch(t *testing.T) {
+	input := []esv1.FakeProviderData{
+		{Key: "remote", Value: "keep-version", Version: "v2"},
+		{Key: "remote", Value: "drop-version", Version: "v1"},
+		{Key: "other", Value: "keep"},
+	}
+
+	got := removeFakeProviderData(input, "remote", "v1")
+
+	want := []esv1.FakeProviderData{
+		{Key: "remote", Value: "keep-version", Version: "v2"},
+		{Key: "other", Value: "keep"},
+	}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Fatalf("removeFakeProviderData mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestProviderReferenceNamespace(t *testing.T) {
+	if got := providerReferenceNamespace(esv1.AuthenticationScopeManifestNamespace, "provider-ns"); got != "" {
+		t.Fatalf("expected empty providerRef namespace for manifest scope, got %q", got)
+	}
+	if got := providerReferenceNamespace(esv1.AuthenticationScopeProviderNamespace, "provider-ns"); got != "provider-ns" {
+		t.Fatalf("expected providerRef namespace for provider scope, got %q", got)
+	}
+}
+
+func TestFakeConfigNamespaceForAuthScope(t *testing.T) {
+	if got := fakeConfigNamespaceForAuthScope(esv1.AuthenticationScopeManifestNamespace, "manifest-ns", "provider-ns"); got != "manifest-ns" {
+		t.Fatalf("expected manifest namespace for manifest scope, got %q", got)
+	}
+	if got := fakeConfigNamespaceForAuthScope(esv1.AuthenticationScopeProviderNamespace, "manifest-ns", "provider-ns"); got != "provider-ns" {
+		t.Fatalf("expected provider namespace for provider scope, got %q", got)
+	}
+}
+
+func TestCreateOrUpdateReadbackExternalSecretUpdatesExistingObject(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := esv1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add external secrets scheme: %v", err)
+	}
+
+	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "readback",
+			Namespace: "default",
+		},
+		Spec: esv1.ExternalSecretSpec{
+			Target: esv1.ExternalSecretTarget{Name: "old-target"},
+		},
+	}).Build()
+
+	updated := &esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "readback",
+			Namespace: "default",
+		},
+		Spec: esv1.ExternalSecretSpec{
+			Target: esv1.ExternalSecretTarget{Name: "new-target"},
+		},
+	}
+	f := &framework.Framework{CRClient: cl}
+	if err := createOrUpdateReadbackExternalSecret(context.Background(), f, updated); err != nil {
+		t.Fatalf("createOrUpdateReadbackExternalSecret returned error: %v", err)
+	}
+
+	var got esv1.ExternalSecret
+	if err := cl.Get(context.Background(), client.ObjectKeyFromObject(updated), &got); err != nil {
+		t.Fatalf("get external secret: %v", err)
+	}
+	if got.Spec.Target.Name != "new-target" {
+		t.Fatalf("expected updated target name, got %q", got.Spec.Target.Name)
+	}
+}
+
+type flakyReadbackCreateClient struct {
+	client.Client
+	createErrs  []error
+	createCalls int
+}
+
+func (c *flakyReadbackCreateClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
+	callIndex := c.createCalls
+	c.createCalls++
+	if callIndex < len(c.createErrs) && c.createErrs[callIndex] != nil {
+		return c.createErrs[callIndex]
+	}
+	return c.Client.Create(ctx, obj, opts...)
+}
+
+func TestCreateOrUpdateReadbackExternalSecretRetriesMissingAPIResourceErrors(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := esv1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add external secrets scheme: %v", err)
+	}
+
+	baseClient := fake.NewClientBuilder().WithScheme(scheme).Build()
+	cl := &flakyReadbackCreateClient{
+		Client: baseClient,
+		createErrs: []error{
+			&metav1api.NoResourceMatchError{},
+		},
+	}
+	f := &framework.Framework{CRClient: cl}
+
+	externalSecret := &esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "readback",
+			Namespace: "default",
+		},
+	}
+
+	if err := createOrUpdateReadbackExternalSecret(context.Background(), f, externalSecret); err != nil {
+		t.Fatalf("createOrUpdateReadbackExternalSecret returned error: %v", err)
+	}
+	if cl.createCalls != 2 {
+		t.Fatalf("expected 2 create calls, got %d", cl.createCalls)
+	}
+
+	var got esv1.ExternalSecret
+	if err := baseClient.Get(context.Background(), client.ObjectKeyFromObject(externalSecret), &got); err != nil {
+		t.Fatalf("get external secret: %v", err)
+	}
+}

e2e/suites/provider/cases/fake/regressions.go

@@ -19,6 +19,7 @@ package fake
 import (
 	"github.com/external-secrets/external-secrets-e2e/framework"
 	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+
 	. "github.com/onsi/ginkgo/v2"
 )
 
@@ -26,6 +27,13 @@ var _ = Describe("[fake] ", Label("fake"), func() {
 	f := framework.New("eso-fake")
 	prov := NewProvider(f)
 
+	DescribeTable("namespaced provider",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.FakeProviderSync(f)),
+		Entry(common.FakeProviderRefresh(f)),
+		Entry(common.FakeProviderFind(f)),
+	)
+
 	// we use the fake provider to test regressions
 	DescribeTable("controller regressions",
 		framework.TableFuncWithExternalSecret(f, prov),

e2e/suites/provider/cases/kubernetes/clusterprovider_v2.go

@@ -0,0 +1,267 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[kubernetes] v2 cluster provider", Label("kubernetes", "v2", "cluster-provider"), func() {
+	f := framework.New("eso-kubernetes-v2-clusterprovider")
+	prov := NewProvider(f)
+	harness := newKubernetesClusterProviderExternalSecretHarness(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+	})
+
+	DescribeTable("cluster provider external secrets",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.ClusterProviderManifestNamespace(f, harness)),
+		Entry(common.ClusterProviderProviderNamespace(f, harness)),
+		Entry(common.ClusterProviderManifestNamespaceRecovery(f, harness)),
+		Entry(common.ClusterProviderProviderNamespaceRecovery(f, harness)),
+		Entry(common.ClusterProviderDeniedByConditions(f, harness)),
+	)
+})
+
+func newKubernetesClusterProviderExternalSecretHarness(f *framework.Framework) common.ClusterProviderExternalSecretHarness {
+	return common.ClusterProviderExternalSecretHarness{
+		Prepare: func(tc *framework.TestCase, cfg common.ClusterProviderConfig) *common.ClusterProviderExternalSecretRuntime {
+			s := newClusterProviderV2Scenario(f, cfg.Name, cfg.AuthScope)
+			s.allowRemoteAccessForScope(cfg.AuthScope, cfg.Name)
+
+			clusterProviderName := s.createClusterProvider(cfg.Name, cfg.AuthScope, cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.ClusterProviderExternalSecretRuntime{
+				ClusterProviderName: clusterProviderName,
+				Provider:            s,
+				BreakAuth: func() {
+					updateKubernetesProviderServiceAccount(f, s.backendNamespace, s.providerConfigName(cfg.Name), "missing-service-account")
+				},
+				RepairAuth: func() {
+					updateKubernetesProviderServiceAccount(f, s.backendNamespace, s.providerConfigName(cfg.Name), s.serviceAccount)
+				},
+			}
+		},
+	}
+}
+
+type clusterProviderV2Scenario struct {
+	f                    *framework.Framework
+	namePrefix           string
+	workloadNamespace    string
+	providerNamespace    string
+	backendNamespace     string
+	providerRefNamespace string
+	remoteNamespace      string
+	serviceAccount       string
+	caBundle             []byte
+}
+
+type clusterProviderV2Layout struct {
+	backendNamespace     string
+	providerNamespace    string
+	providerRefNamespace string
+}
+
+func newClusterProviderV2Layout(workloadNamespace, prefix string, authScope esv1.AuthenticationScope, createProviderNamespace func(string) string) clusterProviderV2Layout {
+	providerNamespace := workloadNamespace
+	if authScope == esv1.AuthenticationScopeProviderNamespace && createProviderNamespace != nil {
+		providerNamespace = createProviderNamespace(prefix + "-provider")
+	}
+
+	backendNamespace := workloadNamespace
+	providerRefNamespace := ""
+	if authScope == esv1.AuthenticationScopeProviderNamespace {
+		backendNamespace = providerNamespace
+		providerRefNamespace = providerNamespace
+	}
+
+	return clusterProviderV2Layout{
+		backendNamespace:     backendNamespace,
+		providerNamespace:    providerNamespace,
+		providerRefNamespace: providerRefNamespace,
+	}
+}
+
+func newClusterProviderV2Scenario(f *framework.Framework, prefix string, authScope esv1.AuthenticationScope) *clusterProviderV2Scenario {
+	layout := newClusterProviderV2Layout(f.Namespace.Name, prefix, authScope, func(providerPrefix string) string {
+		return createE2ENamespace(f, providerPrefix)
+	})
+	s := &clusterProviderV2Scenario{
+		f:                    f,
+		namePrefix:           fmt.Sprintf("%s-%s", f.Namespace.Name, prefix),
+		workloadNamespace:    f.Namespace.Name,
+		providerNamespace:    layout.providerNamespace,
+		backendNamespace:     layout.backendNamespace,
+		providerRefNamespace: layout.providerRefNamespace,
+		serviceAccount:       "eso-auth",
+		caBundle:             frameworkv2.GetClusterCABundle(f, f.Namespace.Name),
+	}
+
+	s.remoteNamespace = createE2ENamespace(f, prefix+"-remote")
+
+	s.createServiceAccount(s.workloadNamespace)
+	if s.providerNamespace != s.workloadNamespace {
+		s.createServiceAccount(s.providerNamespace)
+	}
+
+	return s
+}
+
+func (s *clusterProviderV2Scenario) createServiceAccount(namespace string) {
+	Expect(s.f.CRClient.Create(context.Background(), &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      s.serviceAccount,
+			Namespace: namespace,
+		},
+	})).To(Succeed())
+}
+
+func (s *clusterProviderV2Scenario) allowRemoteAccessFrom(serviceAccountNamespace, suffix string) {
+	frameworkv2.CreateKubernetesAccessRole(
+		s.f,
+		fmt.Sprintf("%s-access-%s", s.namePrefix, suffix),
+		s.serviceAccount,
+		serviceAccountNamespace,
+		s.remoteNamespace,
+	)
+}
+
+func (s *clusterProviderV2Scenario) allowRemoteAccessForScope(authScope esv1.AuthenticationScope, suffix string) {
+	serviceAccountNamespace := s.workloadNamespace
+	if authScope == esv1.AuthenticationScopeProviderNamespace {
+		serviceAccountNamespace = s.providerNamespace
+	}
+	s.allowRemoteAccessFrom(serviceAccountNamespace, suffix)
+}
+
+func (s *clusterProviderV2Scenario) createClusterProvider(suffix string, authScope esv1.AuthenticationScope, conditions []esv1.ClusterSecretStoreCondition) string {
+	providerConfigName := s.providerConfigName(suffix)
+	frameworkv2.CreateKubernetesProvider(
+		s.f,
+		s.backendNamespace,
+		providerConfigName,
+		s.remoteNamespace,
+		s.serviceAccount,
+		nil,
+		s.caBundle,
+	)
+
+	clusterProviderName := fmt.Sprintf("%s-cluster-provider-%s", s.namePrefix, suffix)
+	frameworkv2.CreateClusterProviderConnection(
+		s.f,
+		clusterProviderName,
+		frameworkv2.ProviderAddress("kubernetes"),
+		kubernetesProviderAPIVersion,
+		"Kubernetes",
+		providerConfigName,
+		s.providerRefNamespace,
+		authScope,
+		conditions,
+	)
+	return clusterProviderName
+}
+
+func (s *clusterProviderV2Scenario) providerConfigName(suffix string) string {
+	return fmt.Sprintf("%s-config-%s", s.namePrefix, suffix)
+}
+
+func (s *clusterProviderV2Scenario) CreateSecret(key string, val framework.SecretEntry) {
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      key,
+			Namespace: s.remoteNamespace,
+			Labels:    val.Tags,
+		},
+		Data: make(map[string][]byte),
+	}
+	stringMap := make(map[string]string)
+	err := json.Unmarshal([]byte(val.Value), &stringMap)
+	Expect(err).ToNot(HaveOccurred())
+
+	for k, v := range stringMap {
+		secret.Data[k] = []byte(v)
+	}
+	Expect(s.f.CRClient.Create(GinkgoT().Context(), secret)).To(Succeed())
+}
+
+func (s *clusterProviderV2Scenario) DeleteSecret(key string) {
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      key,
+			Namespace: s.remoteNamespace,
+		},
+	}
+	Expect(s.f.CRClient.Delete(GinkgoT().Context(), secret)).To(Succeed())
+}
+
+func deleteExternalSecretAndWait(ctx context.Context, kubeClient client.Client, key types.NamespacedName) error {
+	externalSecret := &esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      key.Name,
+			Namespace: key.Namespace,
+		},
+	}
+
+	err := kubeClient.Delete(ctx, externalSecret)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+
+	return wait.PollUntilContextTimeout(ctx, defaultV2PollInterval, 2*time.Minute, true, func(ctx context.Context) (bool, error) {
+		var existing esv1.ExternalSecret
+		err := kubeClient.Get(ctx, key, &existing)
+		if apierrors.IsNotFound(err) {
+			return true, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		return false, nil
+	})
+}
+
+func externalSecretConditionHasStatus(condition *esv1.ExternalSecretStatusCondition, want corev1.ConditionStatus) bool {
+	return condition != nil && condition.Status == want
+}
+
+func createE2ENamespace(f *framework.Framework, prefix string) string {
+	return common.CreateProviderCaseNamespace(f, prefix, defaultV2PollInterval)
+}

e2e/suites/provider/cases/kubernetes/clusterprovider_v2_test.go

@@ -0,0 +1,140 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	. "github.com/onsi/gomega"
+)
+
+func TestExternalSecretConditionHasStatus(t *testing.T) {
+	t.Helper()
+	RegisterTestingT(t)
+
+	condition := &esv1.ExternalSecretStatusCondition{Status: corev1.ConditionTrue}
+
+	if !externalSecretConditionHasStatus(condition, corev1.ConditionTrue) {
+		t.Fatalf("expected helper to match ExternalSecret corev1 condition status")
+	}
+
+	if externalSecretConditionHasStatus(condition, corev1.ConditionFalse) {
+		t.Fatalf("expected helper not to match a different ExternalSecret corev1 condition status")
+	}
+}
+
+func TestDeleteExternalSecretAndWaitDeletesExternalSecret(t *testing.T) {
+	t.Helper()
+	RegisterTestingT(t)
+
+	cl := newClusterProviderScenarioTestClient(t, &esv1.ExternalSecret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "example",
+			Namespace: "test",
+		},
+	})
+
+	err := deleteExternalSecretAndWait(context.Background(), cl, types.NamespacedName{
+		Name:      "example",
+		Namespace: "test",
+	})
+	Expect(err).NotTo(HaveOccurred())
+
+	var externalSecret esv1.ExternalSecret
+	err = cl.Get(context.Background(), client.ObjectKey{Name: "example", Namespace: "test"}, &externalSecret)
+	Expect(apierrors.IsNotFound(err)).To(BeTrue())
+}
+
+func TestDeleteExternalSecretAndWaitIgnoresMissingExternalSecret(t *testing.T) {
+	t.Helper()
+	RegisterTestingT(t)
+
+	cl := newClusterProviderScenarioTestClient(t)
+
+	err := deleteExternalSecretAndWait(context.Background(), cl, types.NamespacedName{
+		Name:      "missing",
+		Namespace: "test",
+	})
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func TestClusterProviderV2NamespacesForManifestScope(t *testing.T) {
+	t.Helper()
+
+	layout := newClusterProviderV2Layout("workload-ns", "case", esv1.AuthenticationScopeManifestNamespace, func(string) string {
+		t.Fatal("did not expect provider namespace factory to run for manifest scope")
+		return ""
+	})
+
+	if layout.backendNamespace != "workload-ns" {
+		t.Fatalf("expected backend namespace to use workload namespace, got %q", layout.backendNamespace)
+	}
+	if layout.providerRefNamespace != "" {
+		t.Fatalf("expected provider ref namespace to be omitted for manifest scope, got %q", layout.providerRefNamespace)
+	}
+	if layout.providerNamespace != "workload-ns" {
+		t.Fatalf("expected provider namespace to use workload namespace, got %q", layout.providerNamespace)
+	}
+}
+
+func TestClusterProviderV2NamespacesForProviderScope(t *testing.T) {
+	t.Helper()
+
+	calledWith := ""
+	layout := newClusterProviderV2Layout("workload-ns", "case", esv1.AuthenticationScopeProviderNamespace, func(prefix string) string {
+		calledWith = prefix
+		return "provider-ns"
+	})
+
+	if calledWith != "case-provider" {
+		t.Fatalf("expected provider namespace factory to use case-provider prefix, got %q", calledWith)
+	}
+	if layout.backendNamespace != "provider-ns" {
+		t.Fatalf("expected backend namespace to use provider namespace, got %q", layout.backendNamespace)
+	}
+	if layout.providerRefNamespace != "provider-ns" {
+		t.Fatalf("expected provider ref namespace to use provider namespace, got %q", layout.providerRefNamespace)
+	}
+	if layout.providerNamespace != "provider-ns" {
+		t.Fatalf("expected provider namespace to use provider namespace, got %q", layout.providerNamespace)
+	}
+}
+
+func newClusterProviderScenarioTestClient(t *testing.T, objs ...client.Object) client.Client {
+	t.Helper()
+
+	scheme := runtime.NewScheme()
+	Expect(corev1.AddToScheme(scheme)).To(Succeed())
+	Expect(esv1.AddToScheme(scheme)).To(Succeed())
+
+	builder := fake.NewClientBuilder().WithScheme(scheme)
+	if len(objs) > 0 {
+		builder = builder.WithObjects(objs...)
+	}
+	return builder.Build()
+}

e2e/suites/provider/cases/kubernetes/kubernetes.go

@@ -19,18 +19,27 @@ package kubernetes
 import (
 	"fmt"
 
-	// nolint
-	. "github.com/onsi/ginkgo/v2"
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/external-secrets/external-secrets-e2e/framework"
 	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
 	esapi "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	// nolint
+	. "github.com/onsi/ginkgo/v2"
 )
 
 const referentAuth = "with referent auth"
 
-var _ = Describe("[kubernetes] ", Label("kubernetes"), func() {
+func describeLabels() []any {
+	labels := []any{Label("kubernetes")}
+	if framework.IsV2ProviderMode() {
+		labels = append(labels, Label("v2"))
+	}
+	return labels
+}
+
+var _ = Describe("[kubernetes] ", append(describeLabels(), func() {
 	f := framework.New("eso-kubernetes")
 	prov := NewProvider(f)
 
@@ -50,10 +59,14 @@ var _ = Describe("[kubernetes] ", Label("kubernetes"), func() {
 		framework.Compose(referentAuth, f, common.JSONDataWithProperty, withReferentStore),
 		framework.Compose(referentAuth, f, common.JSONDataWithoutTargetName, withReferentStore),
 	)
-})
+})...)
 
 func withReferentStore(tc *framework.TestCase) {
 	tc.ExternalSecret.Spec.SecretStoreRef.Name = referentStoreName(tc.Framework)
+	if framework.IsV2ProviderMode() {
+		tc.ExternalSecret.Spec.SecretStoreRef.Kind = esapi.ClusterProviderStoreKindStr
+		return
+	}
 	tc.ExternalSecret.Spec.SecretStoreRef.Kind = esapi.ClusterSecretStoreKind
 }
 

e2e/suites/provider/cases/kubernetes/metrics_v2.go

@@ -0,0 +1,263 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[kubernetes] v2 metrics", Label("kubernetes", "v2", "metrics"), func() {
+	f := framework.New("eso-kubernetes-v2-metrics")
+	NewProvider(f)
+
+	controllerClientMethods := []string{"GetSecret", "GetSecretMap"}
+	providerClientMethods := []string{
+		"/provider.v1.SecretStoreProvider/GetSecret",
+		"/provider.v1.SecretStoreProvider/GetSecretMap",
+	}
+
+	hasSuccessfulClientRequest := func(metrics frameworkv2.MetricsMap, methods []string) bool {
+		for _, method := range methods {
+			value, found := frameworkv2.GetMetricValue(metrics, "grpc_client_requests_total", map[string]string{
+				"method": method,
+				"status": "success",
+			})
+			if found && value >= 1.0 {
+				return true
+			}
+		}
+		return false
+	}
+
+	hasSuccessfulServerRequest := func(metrics frameworkv2.MetricsMap, methods []string) bool {
+		for _, method := range methods {
+			value, found := frameworkv2.GetMetricValue(metrics, "grpc_server_requests_total", map[string]string{
+				"method": method,
+				"status": "success",
+			})
+			if found && value >= 1.0 {
+				return true
+			}
+		}
+		return false
+	}
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+		frameworkv2.WaitForClusterProviderReady(f, referentStoreName(f), defaultV2WaitTimeout)
+	})
+
+	It("exposes Provider and ClusterProvider controller metrics", func() {
+		metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+		Expect(err).ToNot(HaveOccurred())
+
+		frameworkv2.ExpectMetricExists(metrics, "provider_status_condition")
+		frameworkv2.ExpectMetricValue(metrics, "provider_status_condition", map[string]string{
+			"name":      f.Namespace.Name,
+			"namespace": f.Namespace.Name,
+			"condition": "Ready",
+			"status":    "True",
+		}, 1.0)
+		frameworkv2.ExpectMetricGreaterThan(metrics, "provider_reconcile_duration", map[string]string{
+			"name":      f.Namespace.Name,
+			"namespace": f.Namespace.Name,
+		}, 0.0)
+
+		frameworkv2.ExpectMetricExists(metrics, "clusterprovider_status_condition")
+		frameworkv2.ExpectMetricValue(metrics, "clusterprovider_status_condition", map[string]string{
+			"name":      referentStoreName(f),
+			"condition": "Ready",
+			"status":    "True",
+		}, 1.0)
+		frameworkv2.ExpectMetricGreaterThan(metrics, "clusterprovider_reconcile_duration", map[string]string{
+			"name": referentStoreName(f),
+		}, 0.0)
+	})
+
+	It("tracks client, server, and cache metrics during secret sync", func() {
+		externalSecretName := "test-es-metrics"
+		targetSecretName := "test-secret-metrics"
+		tcSecretOne := fmt.Sprintf("%s-one", f.Namespace.Name)
+		tcSecretTwo := fmt.Sprintf("%s-two", f.Namespace.Name)
+
+		secretOne := &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      tcSecretOne,
+				Namespace: f.Namespace.Name,
+			},
+			Data: map[string][]byte{
+				"foo": []byte("bar"),
+			},
+		}
+		secretTwo := &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      tcSecretTwo,
+				Namespace: f.Namespace.Name,
+			},
+			Data: map[string][]byte{
+				"baz": []byte("qux"),
+			},
+		}
+		Expect(f.CRClient.Create(context.Background(), secretOne)).To(Succeed())
+		Expect(f.CRClient.Create(context.Background(), secretTwo)).To(Succeed())
+
+		es := &esv1.ExternalSecret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      externalSecretName,
+				Namespace: f.Namespace.Name,
+			},
+			Spec: esv1.ExternalSecretSpec{
+				RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+				SecretStoreRef: esv1.SecretStoreRef{
+					Name: f.Namespace.Name,
+					Kind: esv1.ProviderStoreKindStr,
+				},
+				Target: esv1.ExternalSecretTarget{
+					Name: targetSecretName,
+				},
+				Data: []esv1.ExternalSecretData{
+					{
+						SecretKey: "one",
+						RemoteRef: esv1.ExternalSecretDataRemoteRef{
+							Key:      tcSecretOne,
+							Property: "foo",
+						},
+					},
+					{
+						SecretKey: "two",
+						RemoteRef: esv1.ExternalSecretDataRemoteRef{
+							Key:      tcSecretTwo,
+							Property: "baz",
+						},
+					},
+				},
+			},
+		}
+		Expect(f.CRClient.Create(context.Background(), es)).To(Succeed())
+
+		Eventually(func(g Gomega) {
+			var secret corev1.Secret
+			g.Expect(f.CRClient.Get(context.Background(), types.NamespacedName{Name: targetSecretName, Namespace: f.Namespace.Name}, &secret)).To(Succeed())
+			g.Expect(secret.Data["one"]).To(Equal([]byte("bar")))
+			g.Expect(secret.Data["two"]).To(Equal([]byte("qux")))
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+
+		Eventually(func() bool {
+			metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+			if err != nil {
+				return false
+			}
+			return hasSuccessfulClientRequest(metrics, controllerClientMethods)
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(BeTrue())
+
+		Eventually(func() bool {
+			metrics, err := frameworkv2.ScrapeProviderMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace, "kubernetes")
+			if err != nil {
+				return false
+			}
+			return hasSuccessfulServerRequest(metrics, providerClientMethods)
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(BeTrue())
+
+		Eventually(func() bool {
+			metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+			if err != nil {
+				return false
+			}
+			value, found := frameworkv2.GetMetricValue(metrics, "clientmanager_cache_hits_total", map[string]string{
+				"provider_type": "provider",
+			})
+			return found && value >= 1.0
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(BeTrue())
+	})
+
+	It("reuses one backend connection across many namespaced kubernetes Provider consumers", func() {
+		const consumerCount = 6
+
+		for i := 0; i < consumerCount; i++ {
+			remoteKey := fmt.Sprintf("%s-operational-metric-%d", f.Namespace.Name, i)
+			targetName := fmt.Sprintf("kubernetes-operational-metric-target-%d", i)
+			expectedValue := fmt.Sprintf("metric-value-%d", i)
+
+			Expect(f.CRClient.Create(context.Background(), &corev1.Secret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      remoteKey,
+					Namespace: f.Namespace.Name,
+				},
+				Data: map[string][]byte{
+					"value": []byte(expectedValue),
+				},
+			})).To(Succeed())
+
+			Expect(f.CRClient.Create(context.Background(), &esv1.ExternalSecret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("kubernetes-operational-metric-es-%d", i),
+					Namespace: f.Namespace.Name,
+				},
+				Spec: esv1.ExternalSecretSpec{
+					RefreshInterval: &metav1.Duration{Duration: defaultV2RefreshInterval},
+					SecretStoreRef: esv1.SecretStoreRef{
+						Name: f.Namespace.Name,
+						Kind: esv1.ProviderStoreKindStr,
+					},
+					Target: esv1.ExternalSecretTarget{
+						Name: targetName,
+					},
+					Data: []esv1.ExternalSecretData{{
+						SecretKey: "value",
+						RemoteRef: esv1.ExternalSecretDataRemoteRef{
+							Key:      remoteKey,
+							Property: "value",
+						},
+					}},
+				},
+			})).To(Succeed())
+
+			Eventually(func(g Gomega) {
+				var secret corev1.Secret
+				g.Expect(f.CRClient.Get(context.Background(), types.NamespacedName{Name: targetName, Namespace: f.Namespace.Name}, &secret)).To(Succeed())
+				g.Expect(secret.Data["value"]).To(Equal([]byte(expectedValue)))
+			}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+		}
+
+		Eventually(func(g Gomega) {
+			metrics, err := frameworkv2.ScrapeControllerMetrics(context.Background(), f.KubeConfig, f.KubeClientSet, frameworkv2.ProviderNamespace)
+			g.Expect(err).NotTo(HaveOccurred())
+
+			total := frameworkv2.SumMetricValues(metrics, "grpc_pool_connections_total", map[string]string{
+				"address": frameworkv2.ProviderAddress("kubernetes"),
+			})
+			g.Expect(total).To(BeNumerically(">=", 1))
+			g.Expect(total).To(BeNumerically("<=", 4), "expected bounded connection reuse for kubernetes backend")
+			g.Expect(total).To(BeNumerically("<", consumerCount), "expected fewer pooled connections than consumers")
+		}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+	})
+})

e2e/suites/provider/cases/kubernetes/operational_v2.go

@@ -0,0 +1,167 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+
+	. "github.com/onsi/ginkgo/v2"
+)
+
+var _ = Describe("[kubernetes] v2 operational", Serial, Label("kubernetes", "v2", "operational"), func() {
+	f := framework.New("eso-kubernetes-v2-operational")
+	prov := NewProvider(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+	})
+
+	DescribeTable("external secret operational behavior",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.NamespacedProviderUnavailable(f, newKubernetesOperationalExternalSecretHarness(f, prov), "kubernetes-operational-unavailable", "recovered")),
+		Entry(common.NamespacedProviderRestart(f, newKubernetesOperationalExternalSecretHarness(f, prov), "kubernetes-operational-restart", "restarted")),
+		Entry(common.ClusterProviderUnavailable(f, newKubernetesOperationalExternalSecretHarness(f, prov), "kubernetes-operational-cluster-unavailable", "cluster-recovered", esv1.AuthenticationScopeManifestNamespace)),
+		Entry(common.ClusterProviderRestart(f, newKubernetesOperationalExternalSecretHarness(f, prov), "kubernetes-operational-cluster-restart", "cluster-restarted", esv1.AuthenticationScopeManifestNamespace)),
+	)
+
+	DescribeTable("push secret operational behavior",
+		framework.TableFuncWithPushSecret(f, prov, nil),
+		Entry(common.NamespacedPushSecretUnavailable(f, newKubernetesOperationalPushHarness(f, prov))),
+		Entry(common.ClusterProviderPushUnavailable(f, newKubernetesOperationalPushHarness(f, prov), esv1.AuthenticationScopeManifestNamespace)),
+	)
+})
+
+func kubernetesBackendTarget() frameworkv2.BackendTarget {
+	return frameworkv2.BackendTarget{
+		Namespace:        frameworkv2.ProviderNamespace,
+		PodLabelSelector: "app.kubernetes.io/name=external-secrets-provider-kubernetes",
+	}
+}
+
+func newKubernetesOperationalExternalSecretHarness(f *framework.Framework, prov *Provider) common.OperationalExternalSecretHarness {
+	return common.OperationalExternalSecretHarness{
+		PrepareNamespaced: func(_ *framework.TestCase) *common.OperationalRuntime {
+			return &common.OperationalRuntime{
+				Provider: prov,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: f.Namespace.Name,
+					Kind: esv1.ProviderStoreKindStr,
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 0)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 1)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, kubernetesBackendTarget())
+				},
+			}
+		},
+		PrepareCluster: func(_ *framework.TestCase, cfg common.ClusterProviderConfig) *common.OperationalRuntime {
+			s := newClusterProviderV2Scenario(f, cfg.Name, cfg.AuthScope)
+			s.allowRemoteAccessForScope(cfg.AuthScope, cfg.Name)
+
+			clusterProviderName := s.createClusterProvider(cfg.Name, cfg.AuthScope, cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.OperationalRuntime{
+				Provider: s,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: clusterProviderName,
+					Kind: esv1.ClusterProviderStoreKindStr,
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 0)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 1)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, kubernetesBackendTarget())
+				},
+			}
+		},
+	}
+}
+
+func newKubernetesOperationalPushHarness(f *framework.Framework, prov *Provider) common.OperationalPushSecretHarness {
+	return common.OperationalPushSecretHarness{
+		PrepareNamespaced: func(_ *framework.TestCase) *common.OperationalRuntime {
+			return &common.OperationalRuntime{
+				Provider: prov,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: f.Namespace.Name,
+					Kind: esv1.ProviderStoreKindStr,
+				},
+				DefaultRemoteNamespace: f.Namespace.Name,
+				WaitForRemoteSecret: func(namespace, name, key, expectedValue string) {
+					waitForSecretValueInNamespace(f, namespace, name, key, expectedValue)
+				},
+				ExpectNoRemoteSecret: func(namespace, name string) {
+					expectNoSecretInNamespace(f, namespace, name)
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 0)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 1)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, kubernetesBackendTarget())
+				},
+			}
+		},
+		PrepareCluster: func(_ *framework.TestCase, cfg common.ClusterProviderConfig) *common.OperationalRuntime {
+			s := newClusterProviderV2Scenario(f, cfg.Name, cfg.AuthScope)
+			s.allowRemoteAccessForScope(cfg.AuthScope, cfg.Name)
+
+			clusterProviderName := s.createClusterProvider(cfg.Name, cfg.AuthScope, cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			return &common.OperationalRuntime{
+				Provider: s,
+				ProviderRef: esv1.SecretStoreRef{
+					Name: clusterProviderName,
+					Kind: esv1.ClusterProviderStoreKindStr,
+				},
+				DefaultRemoteNamespace: s.remoteNamespace,
+				WaitForRemoteSecret: func(namespace, name, key, expectedValue string) {
+					waitForSecretValueInNamespace(f, namespace, name, key, expectedValue)
+				},
+				ExpectNoRemoteSecret: func(namespace, name string) {
+					expectNoSecretInNamespace(f, namespace, name)
+				},
+				MakeUnavailable: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 0)
+				},
+				RestoreAvailability: func() {
+					frameworkv2.ScaleDeploymentBySelector(f, kubernetesBackendTarget(), 1)
+				},
+				RestartBackend: func() {
+					frameworkv2.DeleteOneProviderPodBySelector(f, kubernetesBackendTarget())
+				},
+			}
+		},
+	}
+}

e2e/suites/provider/cases/kubernetes/operational_v2_test.go

@@ -0,0 +1,33 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"testing"
+
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+)
+
+func TestKubernetesBackendTargetUsesProviderNamespaceAndSelector(t *testing.T) {
+	target := kubernetesBackendTarget()
+	if target.Namespace != frameworkv2.ProviderNamespace {
+		t.Fatalf("expected provider namespace %q, got %q", frameworkv2.ProviderNamespace, target.Namespace)
+	}
+	if target.PodLabelSelector != "app.kubernetes.io/name=external-secrets-provider-kubernetes" {
+		t.Fatalf("unexpected selector %q", target.PodLabelSelector)
+	}
+}

e2e/suites/provider/cases/kubernetes/provider.go

@@ -19,22 +19,26 @@ package kubernetes
 import (
 	"encoding/json"
 	"fmt"
+	"time"
 
-	// nolint
-	. "github.com/onsi/ginkgo/v2"
-
-	// nolint
-	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/core/v1"
 	rbac "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
 	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
 	esmeta "github.com/external-secrets/external-secrets/apis/meta/v1"
+
+	// nolint
+	. "github.com/onsi/ginkgo/v2"
+	// nolint
+	. "github.com/onsi/gomega"
 )
 
+const kubernetesProviderAPIVersion = "provider.external-secrets.io/v2alpha1"
+
 type Provider struct {
 	framework *framework.Framework
 }
@@ -68,6 +72,12 @@ func (s *Provider) CreateSecret(key string, val framework.SecretEntry) {
 }
 
 func (s *Provider) BeforeEach() {
+	if framework.IsV2ProviderMode() {
+		s.CreateStoreV2()
+		s.CreateReferentStoreV2()
+		return
+	}
+
 	s.CreateStore()
 	s.CreateReferentStore()
 }
@@ -153,12 +163,12 @@ func makeDefaultStore(suffix, namespace string) (*rbac.Role, *rbac.RoleBinding,
 }
 
 func (s *Provider) CreateStore() {
-	rb, role, store := makeDefaultStore("", s.framework.Namespace.Name)
+	role, roleBinding, store := makeDefaultStore("", s.framework.Namespace.Name)
 
 	err := s.framework.CRClient.Create(GinkgoT().Context(), role)
 	Expect(err).ToNot(HaveOccurred())
 
-	err = s.framework.CRClient.Create(GinkgoT().Context(), rb)
+	err = s.framework.CRClient.Create(GinkgoT().Context(), roleBinding)
 	Expect(err).ToNot(HaveOccurred())
 
 	err = s.framework.CRClient.Create(GinkgoT().Context(), store)
@@ -166,7 +176,7 @@ func (s *Provider) CreateStore() {
 }
 
 func (s *Provider) CreateReferentStore() {
-	rb, role, store := makeDefaultStore("referent", s.framework.Namespace.Name)
+	role, roleBinding, store := makeDefaultStore("referent", s.framework.Namespace.Name)
 	// ServiceAccount Namespace is not set, this will be inferred
 	// from the ExternalSecret
 	css := &esv1.ClusterSecretStore{
@@ -180,13 +190,98 @@ func (s *Provider) CreateReferentStore() {
 	err := s.framework.CRClient.Create(GinkgoT().Context(), role)
 	Expect(err).ToNot(HaveOccurred())
 
-	err = s.framework.CRClient.Create(GinkgoT().Context(), rb)
+	err = s.framework.CRClient.Create(GinkgoT().Context(), roleBinding)
 	Expect(err).ToNot(HaveOccurred())
 
 	err = s.framework.CRClient.Create(GinkgoT().Context(), css)
 	Expect(err).ToNot(HaveOccurred())
 }
 
+func (s *Provider) CreateStoreV2() {
+	namespace := s.framework.Namespace.Name
+
+	frameworkv2.CreateKubernetesAccessRole(
+		s.framework,
+		storeAccessName(namespace, ""),
+		frameworkv2.DefaultSAName,
+		namespace,
+		namespace,
+	)
+
+	serviceAccountNamespace := namespace
+	frameworkv2.CreateKubernetesProvider(
+		s.framework,
+		namespace,
+		providerConfigName(namespace, ""),
+		namespace,
+		frameworkv2.DefaultSAName,
+		&serviceAccountNamespace,
+		frameworkv2.GetClusterCABundle(s.framework, namespace),
+	)
+
+	frameworkv2.CreateProviderConnection(
+		s.framework,
+		namespace,
+		namespace,
+		frameworkv2.ProviderAddress("kubernetes"),
+		kubernetesProviderAPIVersion,
+		"Kubernetes",
+		providerConfigName(namespace, ""),
+		namespace,
+	)
+	frameworkv2.WaitForProviderConnectionReady(s.framework, namespace, namespace, 30*time.Second)
+}
+
+func (s *Provider) CreateReferentStoreV2() {
+	namespace := s.framework.Namespace.Name
+	referentName := referentStoreName(s.framework)
+
+	frameworkv2.CreateKubernetesAccessRole(
+		s.framework,
+		storeAccessName(namespace, "referent"),
+		frameworkv2.DefaultSAName,
+		namespace,
+		namespace,
+	)
+
+	frameworkv2.CreateKubernetesProvider(
+		s.framework,
+		namespace,
+		providerConfigName(namespace, "referent"),
+		namespace,
+		frameworkv2.DefaultSAName,
+		nil,
+		frameworkv2.GetClusterCABundle(s.framework, namespace),
+	)
+
+	frameworkv2.CreateClusterProviderConnection(
+		s.framework,
+		referentName,
+		frameworkv2.ProviderAddress("kubernetes"),
+		kubernetesProviderAPIVersion,
+		"Kubernetes",
+		providerConfigName(namespace, "referent"),
+		namespace,
+		esv1.AuthenticationScopeManifestNamespace,
+		nil,
+	)
+	frameworkv2.WaitForClusterProviderReady(s.framework, referentName, 30*time.Second)
+}
+
 func referentStoreName(f *framework.Framework) string {
 	return fmt.Sprintf("%s-referent", f.Namespace.Name)
 }
+
+func providerConfigName(namespace, suffix string) string {
+	if suffix == "" {
+		return fmt.Sprintf("%s-kubernetes", namespace)
+	}
+	return fmt.Sprintf("%s-kubernetes-%s", namespace, suffix)
+}
+
+func storeAccessName(namespace, suffix string) string {
+	if suffix == "" {
+		return fmt.Sprintf("%s-provider-access", namespace)
+	}
+	return fmt.Sprintf("%s-provider-access-%s", namespace, suffix)
+}

e2e/suites/provider/cases/kubernetes/provider_v2.go

@@ -0,0 +1,218 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+	esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
+	esmeta "github.com/external-secrets/external-secrets/apis/meta/v1"
+	k8sv2alpha1 "github.com/external-secrets/external-secrets/apis/provider/kubernetes/v2alpha1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[kubernetes] v2 namespaced provider", Label("kubernetes", "v2", "namespaced-provider"), func() {
+	f := framework.New("eso-kubernetes-v2-provider")
+	prov := NewProvider(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+	})
+
+	DescribeTable("namespaced provider read paths",
+		framework.TableFuncWithExternalSecret(f, prov),
+		Entry(common.NamespacedProviderSync(f, common.NamespacedProviderSyncConfig{
+			Description:        "[kubernetes] should sync an ExternalSecret through a namespaced Provider",
+			ExternalSecretName: "provider-v2-es",
+			TargetSecretName:   "provider-v2-target",
+			RemoteKey:          "provider-v2-remote",
+			RemoteSecretValue:  `{"value":"provider-v2-value"}`,
+			RemoteProperty:     "value",
+			SecretKey:          "value",
+			ExpectedValue:      "provider-v2-value",
+		})),
+		Entry(common.NamespacedProviderRefresh(f, common.NamespacedProviderRefreshConfig{
+			Description:         "[kubernetes] should refresh synced secrets after the remote Kubernetes secret changes",
+			ExternalSecretName:  "provider-v2-refresh-es",
+			TargetSecretName:    "provider-v2-refresh-target",
+			RemoteKey:           "provider-v2-refresh-remote",
+			InitialSecretValue:  `{"value":"provider-v2-initial"}`,
+			UpdatedSecretValue:  `{"value":"provider-v2-updated"}`,
+			RemoteProperty:      "value",
+			SecretKey:           "value",
+			InitialExpectedData: "provider-v2-initial",
+			UpdatedExpectedData: "provider-v2-updated",
+			RefreshInterval:     defaultV2RefreshInterval,
+			WaitTimeout:         30 * time.Second,
+			UpdateRemoteSecret: func(_ *framework.TestCase, _ framework.SecretStoreProvider) {
+				updateRemoteSecretValue(f, f.Namespace.Name, "provider-v2-refresh-remote", "provider-v2-updated")
+			},
+		})),
+		Entry(common.NamespacedProviderFind(f, common.NamespacedProviderFindConfig{
+			Description:        "[kubernetes] should sync ExternalSecret dataFrom.find through a namespaced Provider",
+			ExternalSecretName: "provider-v2-find-es",
+			TargetSecretName:   "provider-v2-find-target",
+			MatchRegExp:        "provider-v2-find-(one|two)",
+			MatchingSecrets: map[string]string{
+				"provider-v2-find-one": `{"value":"provider-v2-one"}`,
+				"provider-v2-find-two": `{"value":"provider-v2-two"}`,
+			},
+			IgnoredSecrets: map[string]string{
+				"provider-v2-ignore": `{"value":"provider-v2-ignore"}`,
+			},
+		})),
+	)
+
+	It("recovers after repairing namespaced Provider auth", func() {
+		prov.CreateSecret("provider-v2-recovery-remote", framework.SecretEntry{
+			Value: `{"value":"provider-v2-recovered"}`,
+		})
+
+		updateKubernetesProviderServiceAccount(f, f.Namespace.Name, providerConfigName(f.Namespace.Name, ""), "missing-service-account")
+
+		frameworkv2.WaitForProviderConnectionNotReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+
+		externalSecret := &esv1.ExternalSecret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "provider-v2-recovery-es",
+				Namespace: f.Namespace.Name,
+			},
+			Spec: esv1.ExternalSecretSpec{
+				SecretStoreRef: esv1.SecretStoreRef{
+					Name: f.Namespace.Name,
+					Kind: esv1.ProviderStoreKindStr,
+				},
+				RefreshInterval: &metav1.Duration{Duration: time.Hour},
+				Target: esv1.ExternalSecretTarget{
+					Name: "provider-v2-recovery-target",
+				},
+				Data: []esv1.ExternalSecretData{{
+					SecretKey: "value",
+					RemoteRef: esv1.ExternalSecretDataRemoteRef{
+						Key:      "provider-v2-recovery-remote",
+						Property: "value",
+					},
+				}},
+			},
+		}
+		Expect(f.CRClient.Create(context.Background(), externalSecret)).To(Succeed())
+
+		waitForExternalSecretReadyStatus(f, f.Namespace.Name, externalSecret.Name, corev1.ConditionFalse)
+		expectSecretToBeAbsent(f, f.Namespace.Name, "provider-v2-recovery-target")
+
+		updateKubernetesProviderServiceAccount(f, f.Namespace.Name, providerConfigName(f.Namespace.Name, ""), frameworkv2.DefaultSAName)
+
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+
+		_, err := waitForSecretValueWithin(f, 30*time.Second, f.Namespace.Name, "provider-v2-recovery-target", &corev1.Secret{
+			Type: corev1.SecretTypeOpaque,
+			Data: map[string][]byte{
+				"value": []byte("provider-v2-recovered"),
+			},
+		})
+		Expect(err).NotTo(HaveOccurred())
+	})
+})
+
+func updateRemoteSecretValue(f *framework.Framework, namespace, name, value string) {
+	var secret corev1.Secret
+	Expect(f.CRClient.Get(context.Background(), types.NamespacedName{
+		Name:      name,
+		Namespace: namespace,
+	}, &secret)).To(Succeed())
+
+	secret.Data["value"] = []byte(value)
+	Expect(f.CRClient.Update(context.Background(), &secret)).To(Succeed())
+}
+
+func updateKubernetesProviderServiceAccount(f *framework.Framework, namespace, name, serviceAccountName string) {
+	var providerConfig k8sv2alpha1.Kubernetes
+	Expect(f.CRClient.Get(context.Background(), types.NamespacedName{
+		Name:      name,
+		Namespace: namespace,
+	}, &providerConfig)).To(Succeed())
+
+	if providerConfig.Spec.Auth == nil || providerConfig.Spec.Auth.ServiceAccount == nil {
+		providerConfig.Spec.Auth = &esv1.KubernetesAuth{
+			ServiceAccount: &esmeta.ServiceAccountSelector{},
+		}
+	}
+	providerConfig.Spec.Auth.ServiceAccount.Name = serviceAccountName
+	Expect(f.CRClient.Update(context.Background(), &providerConfig)).To(Succeed())
+}
+
+func waitForExternalSecretReadyStatus(f *framework.Framework, namespace, name string, expectedStatus corev1.ConditionStatus) {
+	Eventually(func(g Gomega) {
+		var externalSecret esv1.ExternalSecret
+		g.Expect(f.CRClient.Get(context.Background(), types.NamespacedName{
+			Name:      name,
+			Namespace: namespace,
+		}, &externalSecret)).To(Succeed())
+
+		condition := esv1.GetExternalSecretCondition(externalSecret.Status, esv1.ExternalSecretReady)
+		g.Expect(condition).NotTo(BeNil())
+		g.Expect(condition.Status).To(Equal(expectedStatus))
+	}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+}
+
+func waitForSecretValueWithin(f *framework.Framework, timeout time.Duration, namespace, name string, expected *corev1.Secret) (*corev1.Secret, error) {
+	secret := &corev1.Secret{}
+	err := wait.PollUntilContextTimeout(context.Background(), time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+		err := f.CRClient.Get(context.Background(), types.NamespacedName{
+			Namespace: namespace,
+			Name:      name,
+		}, secret)
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		match, matchErr := Equal(expected.Data).Match(secret.Data)
+		if matchErr != nil {
+			return false, matchErr
+		}
+		return secret.Type == expected.Type && match, nil
+	})
+	return secret, err
+}
+
+func expectSecretToBeAbsent(f *framework.Framework, namespace, name string) {
+	Consistently(func() bool {
+		var secret corev1.Secret
+		err := f.CRClient.Get(context.Background(), types.NamespacedName{
+			Name:      name,
+			Namespace: namespace,
+		}, &secret)
+		return apierrors.IsNotFound(err)
+	}, defaultV2RefreshInterval, defaultV2PollInterval).Should(BeTrue())
+}

e2e/suites/provider/cases/kubernetes/push_v2.go

@@ -0,0 +1,117 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/external-secrets/external-secrets-e2e/framework"
+	frameworkv2 "github.com/external-secrets/external-secrets-e2e/framework/v2"
+	"github.com/external-secrets/external-secrets-e2e/suites/provider/cases/common"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[kubernetes] v2 push secret", Label("kubernetes", "v2", "push-secret"), func() {
+	f := framework.New("eso-kubernetes-v2-push")
+	prov := NewProvider(f)
+	harness := newKubernetesClusterProviderPushHarness(f)
+
+	BeforeEach(func() {
+		if !framework.IsV2ProviderMode() {
+			Skip("v2 mode only")
+		}
+		frameworkv2.WaitForProviderConnectionReady(f, f.Namespace.Name, f.Namespace.Name, defaultV2WaitTimeout)
+	})
+
+	DescribeTable("push secret",
+		framework.TableFuncWithPushSecret(f, prov, nil),
+		Entry(common.PushSecretPreservesSourceMetadata(f)),
+		Entry(common.PushSecretImplicitProviderKind(f)),
+		Entry(common.PushSecretRejectsNamespacedRemoteNamespaceOverride(f)),
+		Entry(common.ClusterProviderPushManifestNamespace(f, harness)),
+		Entry(common.ClusterProviderPushProviderNamespace(f, harness)),
+		Entry(common.ClusterProviderPushManifestNamespaceRecovery(f, harness)),
+		Entry(common.ClusterProviderPushProviderNamespaceRecovery(f, harness)),
+		Entry(common.ClusterProviderPushAllowsRemoteNamespaceOverride(f, harness)),
+		Entry(common.ClusterProviderPushDeniedByConditions(f, harness)),
+	)
+})
+
+func newKubernetesClusterProviderPushHarness(f *framework.Framework) common.ClusterProviderPushHarness {
+	return common.ClusterProviderPushHarness{
+		Prepare: func(tc *framework.TestCase, cfg common.ClusterProviderConfig) *common.ClusterProviderPushRuntime {
+			s := newClusterProviderV2Scenario(f, cfg.Name, cfg.AuthScope)
+			s.allowRemoteAccessForScope(cfg.AuthScope, cfg.Name)
+
+			clusterProviderName := s.createClusterProvider(cfg.Name, cfg.AuthScope, cfg.Conditions)
+			frameworkv2.WaitForClusterProviderReady(f, clusterProviderName, defaultV2WaitTimeout)
+
+			// Kubernetes push harness supports all optional ClusterProvider push capabilities.
+			return &common.ClusterProviderPushRuntime{
+				ClusterProviderName:    clusterProviderName,
+				DefaultRemoteNamespace: s.remoteNamespace,
+				BreakAuth: func() {
+					updateKubernetesProviderServiceAccount(f, s.backendNamespace, s.providerConfigName(cfg.Name), "missing-service-account")
+				},
+				RepairAuth: func() {
+					updateKubernetesProviderServiceAccount(f, s.backendNamespace, s.providerConfigName(cfg.Name), s.serviceAccount)
+				},
+				WaitForRemoteSecretValue: func(namespace, name, key, expectedValue string) {
+					waitForSecretValueInNamespace(f, namespace, name, key, expectedValue)
+				},
+				ExpectNoRemoteSecret: func(namespace, name string) {
+					expectNoSecretInNamespace(f, namespace, name)
+				},
+				CreateWritableRemoteScope: func(prefix string) string {
+					namespace := createE2ENamespace(f, prefix)
+					frameworkv2.CreateKubernetesAccessRole(
+						f,
+						fmt.Sprintf("%s-access-%s", s.namePrefix, prefix),
+						s.serviceAccount,
+						s.workloadNamespace,
+						namespace,
+					)
+					return namespace
+				},
+			}
+		},
+	}
+}
+
+func waitForSecretValueInNamespace(f *framework.Framework, namespace, name, key, expectedValue string) {
+	Eventually(func(g Gomega) {
+		var secret corev1.Secret
+		g.Expect(f.CRClient.Get(context.Background(), types.NamespacedName{Name: name, Namespace: namespace}, &secret)).To(Succeed())
+		g.Expect(secret.Data).To(HaveKeyWithValue(key, []byte(expectedValue)))
+	}, defaultV2WaitTimeout, defaultV2PollInterval).Should(Succeed())
+}
+
+func expectNoSecretInNamespace(f *framework.Framework, namespace, name string) {
+	Consistently(func() bool {
+		var secret corev1.Secret
+		err := f.CRClient.Get(context.Background(), types.NamespacedName{Name: name, Namespace: namespace}, &secret)
+		return apierrors.IsNotFound(err)
+	}, 10*time.Second, defaultV2PollInterval).Should(BeTrue())
+}

e2e/suites/provider/cases/kubernetes/v2_constants.go

@@ -0,0 +1,25 @@
+/*
+Copyright © The ESO 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
+
+    https://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 kubernetes
+
+import "time"
+
+const (
+	defaultV2WaitTimeout     = 60 * time.Second
+	defaultV2PollInterval    = 2 * time.Second
+	defaultV2RefreshInterval = 10 * time.Second
+)