steps.go

package steps

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/cucumber/godog"
	jsonpatch "github.com/evanphx/json-patch"
	"github.com/google/go-cmp/cmp"
	"github.com/google/go-containerregistry/pkg/crane"
	"github.com/prometheus/common/expfmt"
	"github.com/prometheus/common/model"
	"github.com/spf13/pflag"
	"github.com/stretchr/testify/require"
	"helm.sh/helm/v3/pkg/release"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/util/sets"
	k8sresource "k8s.io/cli-runtime/pkg/resource"
	"k8s.io/component-base/featuregate"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/yaml"

	ocv1 "github.com/operator-framework/operator-controller/api/v1"
	"github.com/operator-framework/operator-controller/internal/operator-controller/features"
)

const (
	olmDeploymentName = "operator-controller-controller-manager"
	timeout           = 5 * time.Minute
	tick              = 1 * time.Second
)

var (
	olmNamespace        = "olmv1-system"
	kubeconfigPath      string
	k8sCli              string
	deployImageRegistry = sync.OnceValue(func() error {
		if os.Getenv("KIND_CLUSTER_NAME") == "" {
			return nil
		}
		cmd := exec.Command("bash", "-c", "make image-registry")
		dir, _ := os.LookupEnv("ROOT_DIR")
		if dir == "" {
			return fmt.Errorf("ROOT_DIR environment variable not set")
		}
		cmd.Dir = dir
		cmd.Env = append(os.Environ(), fmt.Sprintf("KUBECONFIG=%s", kubeconfigPath))
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr
		return cmd.Run()
	})
)

func RegisterSteps(sc *godog.ScenarioContext) {
	sc.Step(`^OLM is available$`, OLMisAvailable)
	sc.Step(`^(?i)bundle "([^"]+)" is installed in version "([^"]+)"$`, BundleInstalled)

	sc.Step(`^(?i)ClusterExtension is applied(?:\s+.*)?$`, ResourceIsApplied)
	sc.Step(`^(?i)ClusterExtension is updated to version "([^"]+)"$`, ClusterExtensionVersionUpdate)
	sc.Step(`^(?i)ClusterExtension is updated(?:\s+.*)?$`, ResourceIsApplied)
	sc.Step(`^(?i)ClusterExtension is available$`, ClusterExtensionIsAvailable)
	sc.Step(`^(?i)ClusterExtension is rolled out$`, ClusterExtensionIsRolledOut)
	sc.Step(`^(?i)ClusterExtension resources are created and labeled$`, ClusterExtensionResourcesCreatedAndAreLabeled)
	sc.Step(`^(?i)ClusterExtension is removed$`, ClusterExtensionIsRemoved)
	sc.Step(`^(?i)ClusterExtension (?:latest generation )?has (?:been )?reconciled(?: the latest generation)?$`, ClusterExtensionReconciledLatestGeneration)
	sc.Step(`^(?i)the ClusterExtension's constituent resources are removed$`, ClusterExtensionResourcesRemoved)
	sc.Step(`^(?i)ClusterExtension reports "([^"]+)" as active revision(s?)$`, ClusterExtensionReportsActiveRevisions)
	sc.Step(`^(?i)ClusterExtension reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+) and Message:$`, ClusterExtensionReportsCondition)
	sc.Step(`^(?i)ClusterExtension reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+) and Message includes:$`, ClusterExtensionReportsConditionWithMessageFragment)
	sc.Step(`^(?i)ClusterExtension reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+)$`, ClusterExtensionReportsConditionWithoutMsg)
	sc.Step(`^(?i)ClusterExtension reports ([[:alnum:]]+) as ([[:alnum:]]+)$`, ClusterExtensionReportsConditionWithoutReason)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+)$`, ClusterObjectSetReportsConditionWithoutMsg)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+) and Message:$`, ClusterObjectSetReportsConditionWithMsg)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+) and Message includes:$`, ClusterObjectSetReportsConditionWithMessageFragment)
	sc.Step(`^(?i)ClusterExtension reports ([[:alnum:]]+) transition between (\d+) and (\d+) minutes since its creation$`, ClusterExtensionReportsConditionTransitionTime)
	sc.Step(`^(?i)ClusterObjectSet is applied(?:\s+.*)?$`, ResourceIsApplied)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" reconciliation is triggered$`, TriggerClusterObjectSetReconciliation)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" has observed phase "([^"]+)" with a non-empty digest$`, ClusterObjectSetHasObservedPhase)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" is archived$`, ClusterObjectSetIsArchived)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" contains annotation "([^"]+)" with value$`, ClusterObjectSetHasAnnotationWithValue)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" has label "([^"]+)" with value "([^"]+)"$`, ClusterObjectSetHasLabelWithValue)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" phase objects are not found or not owned by the revision$`, ClusterObjectSetObjectsNotFoundOrNotOwned)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" phase objects are managed in Kubernetes secrets$`, ClusterObjectSetPhaseObjectsManagedInSecrets)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" referred secrets exist in "([^"]+)" namespace$`, ClusterObjectSetReferredSecretsExist)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" referred secrets are immutable$`, ClusterObjectSetReferredSecretsAreImmutable)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" referred secrets contain labels$`, ClusterObjectSetReferredSecretsContainLabels)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" referred secrets are owned by the object set$`, ClusterObjectSetReferredSecretsOwnedByObjectSet)
	sc.Step(`^(?i)ClusterObjectSet "([^"]+)" referred secrets have type "([^"]+)"$`, ClusterObjectSetReferredSecretsHaveType)

	sc.Step(`^(?i)resource "([^"]+)" is installed$`, ResourceAvailable)
	sc.Step(`^(?i)resource "([^"]+)" is available$`, ResourceAvailable)
	sc.Step(`^(?i)resource "([^"]+)" is removed$`, ResourceRemoved)
	sc.Step(`^(?i)resource "([^"]+)" is (?:eventually not found|not installed)$`, ResourceEventuallyNotFound)
	sc.Step(`^(?i)resource "([^"]+)" exists$`, ResourceAvailable)
	sc.Step(`^(?i)resource is applied$`, ResourceIsApplied)
	sc.Step(`^(?i)resource "deployment/test-operator" reports as (not ready|ready)$`, MarkTestOperatorNotReady)

	sc.Step(`^(?i)resource apply fails with error msg containing "([^"]+)"$`, ResourceApplyFails)
	sc.Step(`^(?i)resource "([^"]+)" is eventually restored$`, ResourceRestored)
	sc.Step(`^(?i)resource "([^"]+)" matches$`, ResourceMatches)
	sc.Step(`^(?i)rollout restart is performed on "([^"]+)"$`, RolloutRestartIsPerformed)
	sc.Step(`^(?i)annotations are added to "([^"]+)"$`, AnnotationsAreAdded)
	sc.Step(`^(?i)labels are added to "([^"]+)"$`, LabelsAreAdded)
	sc.Step(`^(?i)resource "([^"]+)" has annotations$`, ResourceHasAnnotations)
	sc.Step(`^(?i)resource "([^"]+)" has labels$`, ResourceHasLabels)
	sc.Step(`^(?i)deployment "([^"]+)" pod template has annotation "([^"]+)"$`, DeploymentPodTemplateHasAnnotation)
	sc.Step(`^(?i)deployment "([^"]+)" rollout is complete$`, DeploymentRolloutIsComplete)
	sc.Step(`^(?i)deployment "([^"]+)" has (\d+) replica sets?$`, DeploymentHasReplicaSets)
	sc.Step(`^(?i)ClusterExtension reconciliation is triggered$`, TriggerClusterExtensionReconciliation)

	sc.Step(`^(?i)ServiceAccount "([^"]*)" with permissions to install extensions is available in "([^"]*)" namespace$`, ServiceAccountWithNeededPermissionsIsAvailableInGivenNamespace)
	sc.Step(`^(?i)ServiceAccount "([^"]*)" with needed permissions is available in test namespace$`, ServiceAccountWithNeededPermissionsIsAvailableInTestNamespace)
	sc.Step(`^(?i)ServiceAccount "([^"]*)" without create permissions is available in test namespace$`, ServiceAccountWithoutCreatePermissionsIsAvailableInTestNamespace)
	sc.Step(`^(?i)ServiceAccount "([^"]*)" is available in test namespace$`, ServiceAccountIsAvailableInNamespace)
	sc.Step(`^(?i)ServiceAccount "([^"]*)" in test namespace is cluster admin$`, ServiceAccountWithClusterAdminPermissionsIsAvailableInNamespace)
	sc.Step(`^(?i)ServiceAccount "([^"]+)" in test namespace has permissions to fetch "([^"]+)" metrics$`, ServiceAccountWithFetchMetricsPermissions)
	sc.Step(`^(?i)ServiceAccount "([^"]+)" sends request to "([^"]+)" endpoint of "([^"]+)" service$`, SendMetricsRequest)

	sc.Step(`^"([^"]+)" catalog is updated to version "([^"]+)"$`, CatalogIsUpdatedToVersion)
	sc.Step(`^(?i)ClusterCatalog "([^"]+)" is updated to version "([^"]+)"$`, CatalogIsUpdatedToVersion)
	sc.Step(`^"([^"]+)" catalog serves bundles$`, CatalogServesBundles)
	sc.Step(`^(?i)ClusterCatalog "([^"]+)" serves bundles$`, CatalogServesBundles)
	sc.Step(`^"([^"]+)" catalog image version "([^"]+)" is also tagged as "([^"]+)"$`, TagCatalogImage)
	sc.Step(`^(?i)ClusterCatalog "([^"]+)" image version "([^"]+)" is also tagged as "([^"]+)"$`, TagCatalogImage)
	sc.Step(`^(?i)ClusterCatalog "([^"]+)" is deleted$`, CatalogIsDeleted)

	sc.Step(`^(?i)operator "([^"]+)" target namespace is "([^"]+)"$`, OperatorTargetNamespace)
	sc.Step(`^(?i)Prometheus metrics are returned in the response$`, PrometheusMetricsAreReturned)

	sc.Step(`^(?i)min value for (ClusterExtension|ClusterObjectSet) ((?:\.[a-zA-Z]+)+) is set to (\d+)$`, SetCRDFieldMinValue)

	sc.Step(`^(?i)the current ClusterExtension is tracked for cleanup$`, TrackCurrentClusterExtensionForCleanup)

	// TLS profile enforcement steps — deployment configuration
	sc.Step(`^(?i)the "([^"]+)" deployment is configured with custom TLS minimum version "([^"]+)"$`, ConfigureDeploymentWithCustomTLSVersion)
	sc.Step(`^(?i)the "([^"]+)" deployment is configured with custom TLS version "([^"]+)", ciphers "([^"]+)", and curves "([^"]+)"$`, ConfigureDeploymentWithCustomTLSFull)

	// TLS profile enforcement steps — connection assertions
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint accepts a TLS 1\.3 connection$`, MetricsEndpointAcceptsTLS13)
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint rejects a TLS 1\.2 connection$`, MetricsEndpointRejectsTLS12)
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint negotiates cipher "([^"]+)" over TLS 1\.2$`, MetricsEndpointNegotiatesTLS12Cipher)
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint rejects a TLS 1\.2 connection offering only cipher "([^"]+)"$`, MetricsEndpointRejectsTLS12ConnectionWithCipher)
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint accepts a TLS 1\.2 connection with cipher "([^"]+)" and curve "([^"]+)"$`, MetricsEndpointAcceptsTLS12ConnectionWithCurve)
	sc.Step(`^(?i)the "([^"]+)" metrics endpoint rejects a TLS 1\.2 connection with cipher "([^"]+)" and only curve "([^"]+)"$`, MetricsEndpointRejectsTLS12ConnectionWithCurve)

	// Upgrade-specific steps
	sc.Step(`^(?i)the latest stable OLM release is installed$`, LatestStableOLMReleaseIsInstalled)
	sc.Step(`^(?i)OLM is upgraded$`, OLMIsUpgraded)
	sc.Step(`^(?i)(catalogd|operator-controller) is ready to reconcile resources$`, ComponentIsReadyToReconcile)
	sc.Step(`^(?i)all (ClusterCatalog|ClusterExtension) resources are reconciled$`, allResourcesAreReconciled)
	sc.Step(`^(?i)(ClusterCatalog|ClusterExtension) is reconciled$`, ResourceTypeIsReconciled)
	sc.Step(`^(?i)ClusterCatalog reports ([[:alnum:]]+) as ([[:alnum:]]+) with Reason ([[:alnum:]]+)$`, ClusterCatalogReportsCondition)
}

func init() {
	flagSet := pflag.CommandLine
	flagSet.StringVar(&k8sCli, "k8s.cli", "kubectl", "Path to k8s cli")
	if v, found := os.LookupEnv("KUBECONFIG"); found {
		kubeconfigPath = v
	} else {
		home, err := os.UserHomeDir()
		if err != nil {
			panic(fmt.Sprintf("cannot determine user home directory: %v", err))
		}
		flagSet.StringVar(&kubeconfigPath, "kubeconfig", filepath.Join(home, ".kube", "config"), "Paths to a kubeconfig. Only required if out-of-cluster.")
	}
}

func k8sClient(args ...string) (string, error) {
	cmd := exec.Command(k8sCli, args...)
	logger.V(1).Info("Running", "command", strings.Join(cmd.Args, " "))
	cmd.Env = append(os.Environ(), fmt.Sprintf("KUBECONFIG=%s", kubeconfigPath))
	b, err := cmd.Output()
	if err != nil {
		logger.V(1).Info("Failed to run", "command", strings.Join(cmd.Args, " "), "stderr", stderrOutput(err), "error", err)
	}
	output := string(b)
	logger.V(1).Info("Output", "command", strings.Join(cmd.Args, " "), "output", output)
	return output, err
}

func k8scliWithInput(yaml string, args ...string) (string, error) {
	cmd := exec.Command(k8sCli, args...)
	cmd.Stdin = bytes.NewBufferString(yaml)
	cmd.Env = append(os.Environ(), fmt.Sprintf("KUBECONFIG=%s", kubeconfigPath))
	b, err := cmd.Output()
	return string(b), err
}

// OLMisAvailable waits for the OLM operator-controller deployment to become available. Polls with timeout.
func OLMisAvailable(ctx context.Context) error {
	require.Eventually(godog.T(ctx), func() bool {
		v, err := k8sClient("get", "deployment", "-n", olmNamespace, olmDeploymentName, "-o", "jsonpath='{.status.conditions[?(@.type==\"Available\")].status}'")
		if err != nil {
			return false
		}
		return v == "'True'"
	}, timeout, tick)
	return nil
}

// BundleInstalled waits for the ClusterExtension to report the specified bundle name and version as installed. Polls with timeout.
func BundleInstalled(ctx context.Context, name, version string) error {
	sc := scenarioCtx(ctx)
	waitFor(ctx, func() bool {
		v, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o", "jsonpath={.status.install.bundle}")
		if err != nil {
			return false
		}
		var bundle map[string]interface{}
		if err := json.Unmarshal([]byte(v), &bundle); err != nil {
			return false
		}
		return bundle["name"] == name && bundle["version"] == version
	})
	return nil
}

func toUnstructured(yamlContent string) (*unstructured.Unstructured, error) {
	var u map[string]any
	if err := yaml.Unmarshal([]byte(yamlContent), &u); err != nil {
		return nil, err
	}
	return &unstructured.Unstructured{Object: u}, nil
}

func substituteScenarioVars(content string, sc *scenarioContext) string {
	vars := map[string]string{
		"TEST_NAMESPACE": sc.namespace,
		"NAME":           sc.clusterExtensionName,
		"COS_NAME":       sc.clusterObjectSetName,
		"CATALOG_IMG":    "docker-registry.operator-controller-e2e.svc.cluster.local:5000/e2e/test-catalog:v1",
	}
	if v, found := os.LookupEnv("CATALOG_IMG"); found {
		vars["CATALOG_IMG"] = v
	}
	return templateContent(content, vars)
}

// ResourceApplyFails waits for kubectl apply of the provided YAML to fail with the expected error message. Polls with timeout.
func ResourceApplyFails(ctx context.Context, errMsg string, yamlTemplate *godog.DocString) error {
	sc := scenarioCtx(ctx)
	yamlContent := substituteScenarioVars(yamlTemplate.Content, sc)
	_, err := toUnstructured(yamlContent)
	if err != nil {
		return fmt.Errorf("failed to parse resource yaml: %v", err)
	}
	waitFor(ctx, func() bool {
		_, err := k8scliWithInput(yamlContent, "apply", "-f", "-")
		if err == nil {
			return false
		}
		if stdErr := stderrOutput(err); !strings.Contains(stdErr, errMsg) {
			return false
		}
		return true
	})
	return nil
}

// TrackCurrentClusterExtensionForCleanup saves the current ClusterExtension name in the cleanup list
// so it gets deleted at the end of the scenario. Call this before applying a second ClusterExtension
// in the same scenario, because ResourceIsApplied overwrites the tracked name.
func TrackCurrentClusterExtensionForCleanup(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	if sc.clusterExtensionName != "" {
		sc.addedResources = append(sc.addedResources, resource{name: sc.clusterExtensionName, kind: "clusterextension"})
	}
	return nil
}

// ClusterExtensionVersionUpdate patches the ClusterExtension's catalog version to the specified value.
func ClusterExtensionVersionUpdate(ctx context.Context, version string) error {
	sc := scenarioCtx(ctx)
	patch := map[string]any{
		"spec": map[string]any{
			"source": map[string]any{
				"catalog": map[string]any{
					"version": version,
				},
			},
		},
	}
	pb, err := json.Marshal(patch)
	if err != nil {
		return err
	}
	_, err = k8sClient("patch", "clusterextension", sc.clusterExtensionName, "--type", "merge", "-p", string(pb))
	return err
}

// ResourceIsApplied applies the provided YAML resource to the cluster and in case of ClusterExtension or ClusterObjectSet it captures
// its name in the test context so that it can be referred to in later steps with ${NAME} or ${COS_NAME}, respectively
func ResourceIsApplied(ctx context.Context, yamlTemplate *godog.DocString) error {
	sc := scenarioCtx(ctx)
	yamlContent := substituteScenarioVars(yamlTemplate.Content, sc)
	res, err := toUnstructured(yamlContent)
	if err != nil {
		return fmt.Errorf("failed to parse resource yaml: %v", err)
	}
	out, err := k8scliWithInput(yamlContent, "apply", "-f", "-")
	if err != nil {
		return fmt.Errorf("failed to apply resource %v; err: %w; stderr: %s", out, err, stderrOutput(err))
	}
	if res.GetKind() == "ClusterExtension" {
		sc.clusterExtensionName = res.GetName()
	} else if res.GetKind() == "ClusterObjectSet" {
		sc.clusterObjectSetName = res.GetName()
	} else {
		namespace := res.GetNamespace()
		if namespace == "" {
			namespace = sc.namespace
		}
		sc.addedResources = append(sc.addedResources, resource{
			name:      res.GetName(),
			kind:      strings.ToLower(res.GetKind()),
			namespace: namespace,
		})
	}
	return nil
}

// ClusterExtensionIsAvailable waits for the ClusterExtension's Installed condition to be True. Polls with timeout.
func ClusterExtensionIsAvailable(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	require.Eventually(godog.T(ctx), func() bool {
		v, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o", "jsonpath={.status.conditions[?(@.type==\"Installed\")].status}")
		if err != nil {
			return false
		}
		return v == "True"
	}, timeout, tick)
	return nil
}

// ClusterExtensionReconciledLatestGeneration waits for the ClusterExtension's observedGeneration to match its metadata generation. Polls with timeout.
func ClusterExtensionReconciledLatestGeneration(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	waitFor(ctx, func() bool {
		// Get both generation and observedGeneration in a single kubectl call
		output, err := k8sClient("get", "clusterextension", sc.clusterExtensionName,
			"-o", "jsonpath={.metadata.generation},{.status.conditions[?(@.type=='Progressing')].observedGeneration}")
		if err != nil || output == "" {
			return false
		}
		parts := strings.Split(output, ",")
		if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
			return false
		}
		// Both exist and are equal means reconciliation happened
		return parts[0] == parts[1]
	})
	return nil
}

// ClusterExtensionIsRolledOut waits for the ClusterExtension's Progressing condition to be True with reason Succeeded,
// then gathers its constituent resources into the scenario context. Polls with timeout.
func ClusterExtensionIsRolledOut(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	require.Eventually(godog.T(ctx), func() bool {
		v, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o", "jsonpath={.status.conditions[?(@.type==\"Progressing\")]}")
		if err != nil {
			return false
		}

		var condition map[string]interface{}
		if err := json.Unmarshal([]byte(v), &condition); err != nil {
			return false
		}

		return condition["status"] == "True" && condition["reason"] == "Succeeded" && condition["type"] == "Progressing"
	}, timeout, tick)

	// Save ClusterExtension resources to test context for posterior checks
	if err := sc.GatherClusterExtensionObjects(); err != nil {
		return err
	}
	return nil
}

// ClusterExtensionResourcesCreatedAndAreLabeled verifies each constituent resource has the expected OLM owner-kind
// and owner-name labels. Polls with timeout per resource.
func ClusterExtensionResourcesCreatedAndAreLabeled(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	if len(sc.GetClusterExtensionObjects()) == 0 {
		return fmt.Errorf("extension objects not found in context")
	}

	for _, obj := range sc.GetClusterExtensionObjects() {
		waitFor(ctx, func() bool {
			kind := obj.GetObjectKind().GroupVersionKind().Kind
			clusterObj, err := getResource(kind, obj.GetName(), obj.GetNamespace())
			if err != nil {
				logger.V(1).Error(err, "error getting resource", "name", obj.GetName(), "namespace", obj.GetNamespace(), "kind", kind)
				return false
			}

			labels := clusterObj.GetLabels()
			if labels == nil {
				logger.V(1).Info("no labels found for resource", "name", obj.GetName(), "namespace", obj.GetNamespace(), "kind", kind)
				return false
			}

			for key, expectedValue := range map[string]string{
				"olm.operatorframework.io/owner-kind": "ClusterExtension",
				"olm.operatorframework.io/owner-name": sc.clusterExtensionName,
			} {
				if labels[key] != expectedValue {
					logger.V(1).Info("invalid resource label value", "name", obj.GetName(), "namespace", obj.GetNamespace(), "kind", kind, "label", key, "expected", expectedValue, "actual", labels["olm.operatorframework.io/owner-kind"])
					return false
				}
			}
			return true
		})
	}
	return nil
}

// ClusterExtensionIsRemoved deletes the current ClusterExtension, saving its state for potential restore checks.
func ClusterExtensionIsRemoved(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	return ResourceRemoved(ctx, fmt.Sprintf("clusterextension/%s", sc.clusterExtensionName))
}

// ClusterExtensionResourcesRemoved waits for each previously gathered constituent resource to be deleted. Polls with timeout per resource.
func ClusterExtensionResourcesRemoved(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	if len(sc.GetClusterExtensionObjects()) == 0 {
		return fmt.Errorf("extension objects not found in context")
	}
	for _, obj := range sc.GetClusterExtensionObjects() {
		if err := ResourceEventuallyNotFound(ctx, fmt.Sprintf("%s/%s", obj.GetObjectKind().GroupVersionKind().Kind, obj.GetName())); err != nil {
			return err
		}
	}
	return nil
}

func waitFor(ctx context.Context, conditionFn func() bool) {
	require.Eventually(godog.T(ctx), conditionFn, timeout, tick)
}

type msgMatchFn func(string) bool

func alwaysMatch(_ string) bool { return true }

func isFeatureGateEnabled(feature featuregate.Feature) bool {
	enabled, found := featureGates[feature]
	return enabled && found
}

func messageComparison(ctx context.Context, msg *godog.DocString) msgMatchFn {
	msgCmp := alwaysMatch
	if msg != nil {
		expectedMsg := substituteScenarioVars(strings.Join(strings.Fields(msg.Content), " "), scenarioCtx(ctx))
		msgCmp = func(actual string) bool {
			return actual == expectedMsg
		}
	}
	return msgCmp
}

func waitForCondition(ctx context.Context, resourceType, resourceName, conditionType, conditionStatus string, conditionReason *string, msgCmp msgMatchFn) error {
	require.Eventually(godog.T(ctx), func() bool {
		v, err := k8sClient("get", resourceType, resourceName, "-o", fmt.Sprintf("jsonpath={.status.conditions[?(@.type==\"%s\")]}", conditionType))
		if err != nil {
			return false
		}

		var condition metav1.Condition
		if err := json.Unmarshal([]byte(v), &condition); err != nil {
			return false
		}
		if condition.Status != metav1.ConditionStatus(conditionStatus) {
			return false
		}
		if conditionReason != nil && condition.Reason != *conditionReason {
			return false
		}
		if msgCmp != nil && !msgCmp(condition.Message) {
			return false
		}

		return true
	}, timeout, tick)
	return nil
}

func waitForExtensionCondition(ctx context.Context, conditionType, conditionStatus string, conditionReason *string, msgCmp msgMatchFn) error {
	sc := scenarioCtx(ctx)
	return waitForCondition(ctx, "clusterextension", sc.clusterExtensionName, conditionType, conditionStatus, conditionReason, msgCmp)
}

// ClusterExtensionReportsCondition waits for the ClusterExtension to have a condition matching the specified type,
// status, reason, and exact message. Polls with timeout.
func ClusterExtensionReportsCondition(ctx context.Context, conditionType, conditionStatus, conditionReason string, msg *godog.DocString) error {
	return waitForExtensionCondition(ctx, conditionType, conditionStatus, &conditionReason, messageComparison(ctx, msg))
}

// ClusterExtensionReportsConditionWithMessageFragment waits for the ClusterExtension to have a condition matching
// type, status, and reason, with a message containing the specified fragment. Polls with timeout.
func ClusterExtensionReportsConditionWithMessageFragment(ctx context.Context, conditionType, conditionStatus, conditionReason string, msgFragment *godog.DocString) error {
	msgCmp := alwaysMatch
	if msgFragment != nil {
		expectedMsgFragment := substituteScenarioVars(strings.Join(strings.Fields(msgFragment.Content), " "), scenarioCtx(ctx))
		msgCmp = func(actualMsg string) bool {
			normalizedActual := strings.Join(strings.Fields(actualMsg), " ")
			return strings.Contains(normalizedActual, expectedMsgFragment)
		}
	}
	return waitForExtensionCondition(ctx, conditionType, conditionStatus, &conditionReason, msgCmp)
}

// ClusterExtensionReportsConditionWithoutMsg waits for the ClusterExtension to have a condition matching type,
// status, and reason, without checking the message. Polls with timeout.
func ClusterExtensionReportsConditionWithoutMsg(ctx context.Context, conditionType, conditionStatus, conditionReason string) error {
	return ClusterExtensionReportsCondition(ctx, conditionType, conditionStatus, conditionReason, nil)
}

// ClusterExtensionReportsConditionWithoutReason waits for the ClusterExtension to have a condition matching type
// and status, without checking reason or message. Polls with timeout.
func ClusterExtensionReportsConditionWithoutReason(ctx context.Context, conditionType, conditionStatus string) error {
	return waitForExtensionCondition(ctx, conditionType, conditionStatus, nil, nil)
}

// ClusterExtensionReportsConditionTransitionTime asserts that a condition's lastTransitionTime falls within
// the specified minute range since the ClusterExtension's creation.
func ClusterExtensionReportsConditionTransitionTime(ctx context.Context, conditionType string, minMinutes, maxMinutes int) error {
	sc := scenarioCtx(ctx)
	t := godog.T(ctx)

	// Get the ClusterExtension's creation timestamp and condition's lastTransitionTime
	v, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o",
		fmt.Sprintf("jsonpath={.metadata.creationTimestamp},{.status.conditions[?(@.type==\"%s\")].lastTransitionTime}", conditionType))
	require.NoError(t, err)

	parts := strings.Split(v, ",")
	require.Len(t, parts, 2, "expected creationTimestamp and lastTransitionTime but got: %s", v)

	creationTimestamp, err := time.Parse(time.RFC3339, parts[0])
	require.NoError(t, err, "failed to parse creationTimestamp")

	lastTransitionTime, err := time.Parse(time.RFC3339, parts[1])
	require.NoError(t, err, "failed to parse lastTransitionTime")

	transitionDuration := lastTransitionTime.Sub(creationTimestamp)
	minDuration := time.Duration(minMinutes) * time.Minute
	maxDuration := time.Duration(maxMinutes) * time.Minute

	require.GreaterOrEqual(t, transitionDuration, minDuration,
		"condition %s transitioned too early: %v since creation (expected >= %v)", conditionType, transitionDuration, minDuration)
	require.LessOrEqual(t, transitionDuration, maxDuration,
		"condition %s transitioned too late: %v since creation (expected <= %v)", conditionType, transitionDuration, maxDuration)

	return nil
}

// ClusterExtensionReportsActiveRevisions waits for the ClusterExtension's active revisions to match the expected
// set of revision names. Polls with timeout.
func ClusterExtensionReportsActiveRevisions(ctx context.Context, rawRevisionNames string) error {
	sc := scenarioCtx(ctx)
	expectedRevisionNames := sets.New[string]()
	for _, rev := range strings.Split(rawRevisionNames, ",") {
		expectedRevisionNames.Insert(substituteScenarioVars(strings.TrimSpace(rev), sc))
	}

	waitFor(ctx, func() bool {
		v, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o", "jsonpath={.status.activeRevisions}")
		if err != nil {
			return false
		}
		var activeRevisions []ocv1.RevisionStatus
		if err := json.Unmarshal([]byte(v), &activeRevisions); err != nil {
			return false
		}
		activeRevisionsNames := sets.New[string]()
		for _, rev := range activeRevisions {
			activeRevisionsNames.Insert(rev.Name)
		}
		return activeRevisionsNames.Equal(expectedRevisionNames)
	})
	return nil
}

// ClusterObjectSetReportsConditionWithoutMsg waits for the named ClusterObjectSet to have a condition
// matching type, status, and reason. Polls with timeout.
func ClusterObjectSetReportsConditionWithoutMsg(ctx context.Context, revisionName, conditionType, conditionStatus, conditionReason string) error {
	return waitForCondition(ctx, "clusterobjectset", substituteScenarioVars(revisionName, scenarioCtx(ctx)), conditionType, conditionStatus, &conditionReason, nil)
}

// ClusterObjectSetReportsConditionWithMsg waits for the named ClusterObjectSet to have a condition
// matching type, status, reason, and message. Polls with timeout.
func ClusterObjectSetReportsConditionWithMsg(ctx context.Context, revisionName, conditionType, conditionStatus, conditionReason string, msg *godog.DocString) error {
	return waitForCondition(ctx, "clusterobjectset", substituteScenarioVars(revisionName, scenarioCtx(ctx)), conditionType, conditionStatus, &conditionReason, messageComparison(ctx, msg))
}

// ClusterObjectSetReportsConditionWithMessageFragment waits for the named ClusterObjectSet to have a condition
// matching type, status, reason, with a message containing the specified fragment. Polls with timeout.
func ClusterObjectSetReportsConditionWithMessageFragment(ctx context.Context, revisionName, conditionType, conditionStatus, conditionReason string, msgFragment *godog.DocString) error {
	msgCmp := alwaysMatch
	if msgFragment != nil {
		expectedMsgFragment := substituteScenarioVars(strings.Join(strings.Fields(msgFragment.Content), " "), scenarioCtx(ctx))
		msgCmp = func(actualMsg string) bool {
			normalizedActual := strings.Join(strings.Fields(actualMsg), " ")
			return strings.Contains(normalizedActual, expectedMsgFragment)
		}
	}
	return waitForCondition(ctx, "clusterobjectset", substituteScenarioVars(revisionName, scenarioCtx(ctx)), conditionType, conditionStatus, &conditionReason, msgCmp)
}

// TriggerClusterObjectSetReconciliation annotates the named ClusterObjectSet
// to trigger a new reconciliation cycle.
func TriggerClusterObjectSetReconciliation(ctx context.Context, cosName string) error {
	sc := scenarioCtx(ctx)
	cosName = substituteScenarioVars(cosName, sc)
	_, err := k8sClient("annotate", "clusterobjectset", cosName, "--overwrite",
		fmt.Sprintf("e2e-trigger=%d", time.Now().UnixNano()))
	return err
}

// ClusterObjectSetHasObservedPhase waits for the named ClusterObjectSet to have
// an observedPhases entry matching the given phase name with a non-empty digest. Polls with timeout.
func ClusterObjectSetHasObservedPhase(ctx context.Context, cosName, phaseName string) error {
	sc := scenarioCtx(ctx)
	cosName = substituteScenarioVars(cosName, sc)
	phaseName = substituteScenarioVars(phaseName, sc)

	waitFor(ctx, func() bool {
		out, err := k8sClient("get", "clusterobjectset", cosName, "-o",
			fmt.Sprintf(`jsonpath={.status.observedPhases[?(@.name=="%s")].digest}`, phaseName))
		if err != nil {
			return false
		}
		return strings.TrimSpace(out) != ""
	})
	return nil
}

// ClusterObjectSetIsArchived waits for the named ClusterObjectSet to have Progressing=False
// with reason Archived. Polls with timeout.
func ClusterObjectSetIsArchived(ctx context.Context, revisionName string) error {
	return waitForCondition(ctx, "clusterobjectset", substituteScenarioVars(revisionName, scenarioCtx(ctx)), "Progressing", "False", ptr.To("Archived"), nil)
}

// ClusterObjectSetHasAnnotationWithValue waits for the named ClusterObjectSet to have the specified
// annotation with the expected value. Polls with timeout.
func ClusterObjectSetHasAnnotationWithValue(ctx context.Context, revisionName, annotationKey string, annotationValue *godog.DocString) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)
	expectedValue := ""
	if annotationValue != nil {
		expectedValue = annotationValue.Content
	}
	waitFor(ctx, func() bool {
		obj, err := getResource("clusterobjectset", revisionName, "")
		if err != nil {
			logger.V(1).Error(err, "failed to get clusterobjectset", "name", revisionName)
			return false
		}
		if obj.GetAnnotations() == nil {
			return false
		}
		return obj.GetAnnotations()[annotationKey] == expectedValue
	})
	return nil
}

// ClusterObjectSetHasLabelWithValue waits for the named ClusterObjectSet to have the specified label
// with the expected value. Polls with timeout.
func ClusterObjectSetHasLabelWithValue(ctx context.Context, revisionName, labelKey, labelValue string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)
	labelValue = substituteScenarioVars(labelValue, sc)
	waitFor(ctx, func() bool {
		obj, err := getResource("clusterobjectset", revisionName, "")
		if err != nil {
			logger.V(1).Error(err, "failed to get clusterobjectset", "name", revisionName)
			return false
		}
		if obj.GetLabels() == nil {
			return false
		}
		return obj.GetLabels()[labelKey] == labelValue
	})
	return nil
}

// ClusterObjectSetObjectsNotFoundOrNotOwned waits for all objects described in the named
// ClusterObjectSet's phases to either not exist on the cluster or not contain the revision
// in their ownerReferences. Polls with timeout.
func ClusterObjectSetObjectsNotFoundOrNotOwned(ctx context.Context, revisionName string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	// Get the ClusterObjectSet to extract its phase objects
	var rev ocv1.ClusterObjectSet
	waitFor(ctx, func() bool {
		out, err := k8sClient("get", "clusterobjectset", revisionName, "-o", "json")
		if err != nil {
			return false
		}
		return json.Unmarshal([]byte(out), &rev) == nil
	})

	// For each object in each phase, verify it either doesn't exist or
	// doesn't have the ClusterObjectSet in its ownerReferences
	for i, phase := range rev.Spec.Phases {
		for j, phaseObj := range phase.Objects {
			var obj *unstructured.Unstructured
			switch {
			case phaseObj.Ref.Name != "":
				resolved, err := resolveObjectRef(phaseObj.Ref)
				if err != nil {
					return fmt.Errorf("resolving ref in phase %q object %d: %w", phase.Name, j, err)
				}
				obj = resolved
			case len(phaseObj.Object.Object) > 0:
				obj = &phaseObj.Object
			default:
				return fmt.Errorf("clusterobjectset %q phase %d object %d has neither ref nor inline object", revisionName, i, j)
			}
			kind := obj.GetKind()
			name := obj.GetName()
			namespace := obj.GetNamespace()

			if kind == "" {
				return fmt.Errorf("clusterobjectset %q has a phase object with empty kind", revisionName)
			}
			if name == "" {
				return fmt.Errorf("clusterobjectset %q has a phase object with empty name (kind %q, namespace %q)", revisionName, kind, namespace)
			}

			waitFor(ctx, func() bool {
				args := []string{"get", kind, name, "--ignore-not-found", "-o", "json"}
				if namespace != "" {
					args = append(args, "-n", namespace)
				}
				out, err := k8sClient(args...)
				if err != nil {
					return false
				}
				// If output is empty, the resource does not exist — condition satisfied
				if strings.TrimSpace(out) == "" {
					return true
				}
				clusterObj, err := toUnstructured(out)
				if err != nil {
					return false
				}

				// Check that no ownerReference points to this ClusterObjectSet
				for _, ref := range clusterObj.GetOwnerReferences() {
					if ref.Kind == ocv1.ClusterObjectSetKind && ref.Name == revisionName && ref.UID == rev.UID {
						logger.V(1).Info("object still owned by revision",
							"kind", kind, "name", name, "namespace", namespace,
							"revision", revisionName)
						return false
					}
				}
				return true
			})
		}
	}
	return nil
}

// ClusterObjectSetPhaseObjectsManagedInSecrets verifies that every object in every phase of the named
// ClusterObjectSet uses a ref (not an inline object). Polls with timeout.
func ClusterObjectSetPhaseObjectsManagedInSecrets(ctx context.Context, revisionName string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	waitFor(ctx, func() bool {
		obj, err := getResource("clusterobjectset", revisionName, "")
		if err != nil {
			return false
		}
		phases, ok, _ := unstructured.NestedSlice(obj.Object, "spec", "phases")
		if !ok || len(phases) == 0 {
			return false
		}
		for _, p := range phases {
			phase, ok := p.(map[string]interface{})
			if !ok {
				return false
			}
			objects, ok, _ := unstructured.NestedSlice(phase, "objects")
			if !ok || len(objects) == 0 {
				return false
			}
			for _, o := range objects {
				obj, ok := o.(map[string]interface{})
				if !ok {
					return false
				}
				ref, refOK, _ := unstructured.NestedMap(obj, "ref")
				if !refOK || len(ref) == 0 {
					logger.V(1).Info("object does not use ref", "revision", revisionName)
					return false
				}
				name, _, _ := unstructured.NestedString(ref, "name")
				if name == "" {
					logger.V(1).Info("ref has empty name", "revision", revisionName)
					return false
				}
			}
		}
		return true
	})
	return nil
}

// ClusterObjectSetReferredSecretsExist verifies that all Secrets referenced by the named
// ClusterObjectSet's phase objects exist in the given namespace. Polls with timeout.
func ClusterObjectSetReferredSecretsExist(ctx context.Context, revisionName, namespace string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)
	namespace = substituteScenarioVars(strings.TrimSpace(namespace), sc)

	secretNames, err := collectReferredSecretNames(ctx, revisionName)
	if err != nil {
		return err
	}

	for _, name := range secretNames {
		waitFor(ctx, func() bool {
			_, err := getResource("secret", name, namespace)
			return err == nil
		})
	}
	return nil
}

// ClusterObjectSetReferredSecretsAreImmutable verifies that all referred Secrets for the named
// ClusterObjectSet are immutable. Polls with timeout.
func ClusterObjectSetReferredSecretsAreImmutable(ctx context.Context, revisionName string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	secrets, err := listReferredSecrets(ctx, revisionName)
	if err != nil {
		return err
	}
	if len(secrets) == 0 {
		return fmt.Errorf("no referred secrets found for revision %q", revisionName)
	}
	for _, s := range secrets {
		if s.Immutable == nil || !*s.Immutable {
			return fmt.Errorf("referred secret %s/%s is not immutable", s.Namespace, s.Name)
		}
	}
	return nil
}

// ClusterObjectSetReferredSecretsContainLabels verifies that all referred Secrets for the named
// ClusterObjectSet have the expected labels specified in the data table. Polls with timeout.
func ClusterObjectSetReferredSecretsContainLabels(ctx context.Context, revisionName string, table *godog.Table) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	expected, err := parseKeyValueTable(table)
	if err != nil {
		return fmt.Errorf("invalid labels table: %w", err)
	}
	for k, v := range expected {
		expected[k] = substituteScenarioVars(v, sc)
	}

	waitFor(ctx, func() bool {
		secrets, err := listReferredSecrets(ctx, revisionName)
		if err != nil || len(secrets) == 0 {
			return false
		}
		for _, s := range secrets {
			if _, _, ok := matchLabels(s.Labels, expected); !ok {
				return false
			}
		}
		return true
	})
	return nil
}

// ClusterObjectSetReferredSecretsOwnedByObjectSet verifies that all referred Secrets for the named
// ClusterObjectSet have an ownerReference pointing to the ClusterObjectSet with controller=true.
func ClusterObjectSetReferredSecretsOwnedByObjectSet(ctx context.Context, revisionName string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	cosObj, err := getResource("clusterobjectset", revisionName, "")
	if err != nil {
		return fmt.Errorf("getting ClusterObjectSet %q: %w", revisionName, err)
	}
	cosUID := cosObj.GetUID()

	secrets, err := listReferredSecrets(ctx, revisionName)
	if err != nil {
		return err
	}
	if len(secrets) == 0 {
		return fmt.Errorf("no referred secrets found for revision %q", revisionName)
	}

	for _, s := range secrets {
		found := false
		for _, ref := range s.OwnerReferences {
			if ref.Kind == ocv1.ClusterObjectSetKind && ref.Name == revisionName && ref.UID == cosUID {
				if ref.Controller == nil || !*ref.Controller {
					return fmt.Errorf("secret %s/%s has ownerReference to ClusterObjectSet but controller is not true", s.Namespace, s.Name)
				}
				found = true
				break
			}
		}
		if !found {
			return fmt.Errorf("secret %s/%s does not have ownerReference to ClusterObjectSet %q (uid %s)", s.Namespace, s.Name, revisionName, cosUID)
		}
	}
	return nil
}

// ClusterObjectSetReferredSecretsHaveType verifies that all referred Secrets for the named
// ClusterObjectSet have the specified Secret type.
func ClusterObjectSetReferredSecretsHaveType(ctx context.Context, revisionName, expectedType string) error {
	sc := scenarioCtx(ctx)
	revisionName = substituteScenarioVars(strings.TrimSpace(revisionName), sc)

	secrets, err := listReferredSecrets(ctx, revisionName)
	if err != nil {
		return err
	}
	if len(secrets) == 0 {
		return fmt.Errorf("no referred secrets found for revision %q", revisionName)
	}

	for _, s := range secrets {
		if string(s.Type) != expectedType {
			return fmt.Errorf("secret %s/%s has type %q, expected %q", s.Namespace, s.Name, s.Type, expectedType)
		}
	}
	return nil
}

// collectReferredSecretNames returns the unique set of Secret names referenced by the ClusterObjectSet's phase objects.
func collectReferredSecretNames(ctx context.Context, revisionName string) ([]string, error) {
	var names []string
	seen := sets.New[string]()

	var obj *unstructured.Unstructured
	waitFor(ctx, func() bool {
		var err error
		obj, err = getResource("clusterobjectset", revisionName, "")
		return err == nil
	})

	phases, _, _ := unstructured.NestedSlice(obj.Object, "spec", "phases")
	for _, p := range phases {
		phase, ok := p.(map[string]interface{})
		if !ok {
			continue
		}
		objects, _, _ := unstructured.NestedSlice(phase, "objects")
		for _, o := range objects {
			phaseObj, ok := o.(map[string]interface{})
			if !ok {
				continue
			}
			name, _, _ := unstructured.NestedString(phaseObj, "ref", "name")
			if name != "" && !seen.Has(name) {
				seen.Insert(name)
				names = append(names, name)
			}
		}
	}
	if len(names) == 0 {
		return nil, fmt.Errorf("no referred secret names found in ClusterObjectSet %q", revisionName)
	}
	return names, nil
}

// listReferredSecrets lists all Secrets in the OLM namespace that have the revision-name label
// matching the given revision name.
func listReferredSecrets(_ context.Context, revisionName string) ([]corev1.Secret, error) {
	out, err := k8sClient("get", "secrets", "-n", olmNamespace,
		"-l", "olm.operatorframework.io/revision-name="+revisionName, "-o", "json")
	if err != nil {
		return nil, fmt.Errorf("listing referred secrets for revision %q: %w", revisionName, err)
	}
	var secretList corev1.SecretList
	if err := json.Unmarshal([]byte(out), &secretList); err != nil {
		return nil, fmt.Errorf("unmarshalling Secret list: %w", err)
	}
	return secretList.Items, nil
}

// ResourceAvailable waits for the specified resource (kind/name format) to exist in the test namespace. Polls with timeout.
func ResourceAvailable(ctx context.Context, resource string) error {
	sc := scenarioCtx(ctx)
	resource = substituteScenarioVars(resource, sc)
	kind, name, found := strings.Cut(resource, "/")
	if !found {
		return fmt.Errorf("resource %s is not in the format <kind>/<name>", resource)
	}
	waitFor(ctx, func() bool {
		_, err := k8sClient("get", kind, name, "-n", sc.namespace)
		return err == nil
	})
	return nil
}

// ResourceRemoved saves the resource state for potential restore checks, then deletes it from the test namespace.
func ResourceRemoved(ctx context.Context, resource string) error {
	sc := scenarioCtx(ctx)
	resource = substituteScenarioVars(resource, sc)
	kind, name, found := strings.Cut(resource, "/")
	if !found {
		return fmt.Errorf("resource %s is not in the format <kind>/<name>", resource)
	}
	yaml, err := k8sClient("get", kind, name, "-n", sc.namespace, "-o", "yaml")
	if err != nil {
		return err
	}
	obj, err := toUnstructured(yaml)
	if err != nil {
		return err
	}
	sc.removedResources = append(sc.removedResources, *obj)
	_, err = k8sClient("delete", kind, name, "-n", sc.namespace)
	return err
}

// ResourceEventuallyNotFound waits for the specified resource to be fully deleted from the test namespace. Polls with timeout.
func ResourceEventuallyNotFound(ctx context.Context, resource string) error {
	sc := scenarioCtx(ctx)
	resource = substituteScenarioVars(resource, sc)
	kind, name, found := strings.Cut(resource, "/")
	if !found {
		return fmt.Errorf("resource %s is not in the format <kind>/<name>", resource)
	}

	waitFor(ctx, func() bool {
		obj, err := k8sClient("get", kind, name, "-n", sc.namespace, "--ignore-not-found", "-o", "yaml")
		return err == nil && strings.TrimSpace(obj) == ""
	})
	return nil
}

// ResourceMatches waits for the specified resource to match the expected content using JSON merge patch comparison. Polls with timeout.
func ResourceMatches(ctx context.Context, resource string, requiredContentTemplate *godog.DocString) error {
	sc := scenarioCtx(ctx)
	resource = substituteScenarioVars(resource, sc)
	kind, name, found := strings.Cut(resource, "/")
	if !found {
		return fmt.Errorf("resource %s is not in the format <kind>/<name>", resource)
	}
	requiredContent, err := toUnstructured(substituteScenarioVars(requiredContentTemplate.Content, sc))
	if err != nil {
		return fmt.Errorf("failed to parse required resource yaml: %v", err)
	}
	waitFor(ctx, func() bool {
		objJson, err := k8sClient("get", kind, name, "-n", sc.namespace, "-o", "json")
		if err != nil {
			return false
		}
		obj, err := toUnstructured(objJson)
		if err != nil {
			return false
		}
		patch, err := json.Marshal(requiredContent.Object)
		if err != nil {
			return false
		}
		updJson, err := jsonpatch.MergePatch([]byte(objJson), patch)
		if err != nil {
			return false
		}
		upd, err := toUnstructured(string(updJson))
		if err != nil {
			return false
		}

		return len(cmp.Diff(upd.Object, obj.Object)) == 0
	})
	return nil
}

// ResourceRestored waits for a previously removed resource to be re-created with matching spec or data. Polls with timeout.
func ResourceRestored(ctx context.Context, resource string) error {
	sc := scenarioCtx(ctx)
	resource = substituteScenarioVars(resource, sc)
	kind, name, found := strings.Cut(resource, "/")
	if !found {
		return fmt.Errorf("resource %s is not in the format <kind>/<name>", resource)
	}
	waitFor(ctx, func() bool {
		yaml, err := k8sClient("get", kind, name, "-n", sc.namespace, "-o", "yaml")
		if err != nil {
			return false
		}
		obj, err := toUnstructured(yaml)
		if err != nil {
			return false
		}
		ct := obj.GetCreationTimestamp()

		for i, removed := range sc.removedResources {
			rct := removed.GetCreationTimestamp()
			if removed.GetName() == obj.GetName() && removed.GetKind() == obj.GetKind() && rct.Before(&ct) {
				switch kind {
				case "configmap":
					if !reflect.DeepEqual(removed.Object["data"], obj.Object["data"]) {
						return false
					}
				default:
					if !reflect.DeepEqual(removed.Object["spec"], obj.Object["spec"]) {
						return false
					}
				}
				sc.removedResources = append(sc.removedResources[:i], sc.removedResources[i+1:]...)
				return true
			}
		}
		return false
	})
	return nil
}

func applyServiceAccount(ctx context.Context, serviceAccount string, keyValue ...string) error {
	sc := scenarioCtx(ctx)
	vars := extendMap(map[string]string{
		"TEST_NAMESPACE":       sc.namespace,
		"SERVICE_ACCOUNT_NAME": serviceAccount,
		"SERVICEACCOUNT_NAME":  serviceAccount,
	}, keyValue...)

	_, thisFile, _, _ := runtime.Caller(0)
	yaml, err := templateYaml(filepath.Join(filepath.Dir(thisFile), "testdata", "serviceaccount-template.yaml"), vars)
	if err != nil {
		return fmt.Errorf("failed to template ServiceAccount yaml: %v", err)
	}

	// Apply the ServiceAccount configuration
	_, err = k8scliWithInput(yaml, "apply", "-f", "-")
	if err != nil {
		return fmt.Errorf("failed to apply ServiceAccount configuration: %v: %s", err, stderrOutput(err))
	}

	return nil
}

func applyPermissionsToServiceAccount(ctx context.Context, serviceAccount, rbacTemplate string, keyValue ...string) error {
	sc := scenarioCtx(ctx)
	if err := applyServiceAccount(ctx, serviceAccount); err != nil {
		return err
	}
	vars := extendMap(map[string]string{
		"TEST_NAMESPACE":         sc.namespace,
		"SERVICE_ACCOUNT_NAME":   serviceAccount,
		"SERVICEACCOUNT_NAME":    serviceAccount,
		"CLUSTER_EXTENSION_NAME": sc.clusterExtensionName,
		"CLUSTEREXTENSION_NAME":  sc.clusterExtensionName,
	}, keyValue...)

	_, thisFile, _, _ := runtime.Caller(0)
	yaml, err := templateYaml(filepath.Join(filepath.Dir(thisFile), "testdata", rbacTemplate), vars)
	if err != nil {
		return fmt.Errorf("failed to template RBAC yaml: %v", err)
	}

	// Apply the RBAC configuration
	_, err = k8scliWithInput(yaml, "apply", "-f", "-")
	if err != nil {
		return fmt.Errorf("failed to apply RBAC configuration: %v: %s", err, stderrOutput(err))
	}

	return nil
}

// ServiceAccountIsAvailableInNamespace creates a ServiceAccount in the test namespace without RBAC permissions.
func ServiceAccountIsAvailableInNamespace(ctx context.Context, serviceAccount string) error {
	return applyServiceAccount(ctx, serviceAccount)
}

// ServiceAccountWithNeededPermissionsIsAvailableInTestNamespace creates a ServiceAccount and applies standard RBAC permissions.
// The RBAC template is selected based on the service account and BoxcutterRuntime feature gate: <service-account>-<helm|boxcutter>-rbac-template.yaml
func ServiceAccountWithNeededPermissionsIsAvailableInTestNamespace(ctx context.Context, serviceAccount string) error {
	kernel := "helm"
	if enabled, found := featureGates[features.BoxcutterRuntime]; found && enabled {
		kernel = "boxcutter"
	}
	rbacTemplate := fmt.Sprintf("%s-%s-rbac-template.yaml", serviceAccount, kernel)
	return applyPermissionsToServiceAccount(ctx, serviceAccount, rbacTemplate)
}

// ServiceAccountWithoutCreatePermissionsIsAvailableInTestNamespace creates a ServiceAccount with permissions that
// intentionally exclude the "create" verb to test preflight permission validation for Boxcutter applier.
// This is used to verify that the preflight check properly detects missing CREATE permissions.
// Note: This function requires both @BoxcutterRuntime and @PreflightPermissions tags.
func ServiceAccountWithoutCreatePermissionsIsAvailableInTestNamespace(ctx context.Context, serviceAccount string) error {
	// This test is only valid with Boxcutter runtime enabled
	if !isFeatureGateEnabled(features.BoxcutterRuntime) {
		return fmt.Errorf("this step requires BoxcutterRuntime feature gate to be enabled")
	}
	// It also requires preflight permissions checks to be enabled
	if !isFeatureGateEnabled(features.PreflightPermissions) {
		return fmt.Errorf("this step requires PreflightPermissions feature gate to be enabled")
	}
	rbacTemplate := fmt.Sprintf("%s-boxcutter-no-create-rbac-template.yaml", serviceAccount)
	return applyPermissionsToServiceAccount(ctx, serviceAccount, rbacTemplate)
}

// ServiceAccountWithNeededPermissionsIsAvailableInGivenNamespace creates a ServiceAccount and enables creation of any cluster extension on behalf of this account.
func ServiceAccountWithNeededPermissionsIsAvailableInGivenNamespace(ctx context.Context, serviceAccount string, ns string) error {
	sc := scenarioCtx(ctx)
	sc.addedResources = append(sc.addedResources, resource{name: ns, kind: "namespace"})
	if err := applyServiceAccount(ctx, serviceAccount, "TEST_NAMESPACE", ns); err != nil {
		return err
	}
	return applyPermissionsToServiceAccount(ctx, serviceAccount, "rbac-template-for-all-extensions.yaml", "TEST_NAMESPACE", ns)
}

// ServiceAccountWithClusterAdminPermissionsIsAvailableInNamespace creates a ServiceAccount and applies cluster-admin RBAC.
func ServiceAccountWithClusterAdminPermissionsIsAvailableInNamespace(ctx context.Context, serviceAccount string) error {
	return applyPermissionsToServiceAccount(ctx, serviceAccount, "cluster-admin-rbac-template.yaml")
}

// ServiceAccountWithFetchMetricsPermissions creates a ServiceAccount and applies metrics-reader RBAC for the specified controller.
func ServiceAccountWithFetchMetricsPermissions(ctx context.Context, serviceAccount string, controllerName string) error {
	return applyPermissionsToServiceAccount(ctx, serviceAccount, "metrics-reader-rbac-template.yaml", "CONTROLLER_NAME", controllerName)
}

func httpGet(url string, token string) (*http.Response, error) {
	tr := &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // we don't care about the certificate
	}
	client := &http.Client{Transport: tr}

	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Authorization", "Bearer "+token)

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func randomAvailablePort() (int, error) {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		return 0, err
	}
	defer l.Close()
	return l.Addr().(*net.TCPAddr).Port, nil
}

// SendMetricsRequest sets up port-forwarding to the controller's service pods and waits for the metrics endpoint
// to return a successful response. Stores the response body per pod in the scenario context. Polls with timeout.
func SendMetricsRequest(ctx context.Context, serviceAccount string, endpoint string, controllerName string) error {
	sc := scenarioCtx(ctx)
	serviceNs, err := k8sClient("get", "service", "-A", "-o", fmt.Sprintf(`jsonpath={.items[?(@.metadata.name=="%s-service")].metadata.namespace}`, controllerName))
	if err != nil {
		return err
	}
	v, err := k8sClient("get", "service", "-n", serviceNs, fmt.Sprintf("%s-service", controllerName), "-o", "json")
	if err != nil {
		return err
	}
	var service corev1.Service
	if err := json.Unmarshal([]byte(v), &service); err != nil {
		return err
	}
	podNameCmd := []string{"get", "pod", "-n", olmNamespace, "-o", "jsonpath={.items}"}
	for k, v := range service.Spec.Selector {
		podNameCmd = append(podNameCmd, fmt.Sprintf("--selector=%s=%s", k, v))
	}
	v, err = k8sClient(podNameCmd...)
	if err != nil {
		return err
	}

	var pods []corev1.Pod
	if err := json.Unmarshal([]byte(v), &pods); err != nil {
		return err
	}
	token, err := k8sClient("create", "token", serviceAccount, "-n", sc.namespace)
	if err != nil {
		return err
	}
	var metricsPort int32
	for _, p := range service.Spec.Ports {
		if p.Name == "metrics" {
			metricsPort = p.Port
			break
		}
	}
	sc.metricsResponse = make(map[string]string)
	for _, p := range pods {
		port, err := randomAvailablePort()
		if err != nil {
			return err
		}
		portForwardCmd := exec.Command(k8sCli, "port-forward", "-n", p.Namespace, fmt.Sprintf("pod/%s", p.Name), fmt.Sprintf("%d:%d", port, metricsPort)) //nolint:gosec // perfectly safe to start port-forwarder for provided controller name
		logger.V(1).Info("starting port-forward", "command", strings.Join(portForwardCmd.Args, " "))
		if err := portForwardCmd.Start(); err != nil {
			logger.Error(err, fmt.Sprintf("failed to start port-forward for pod %s", p.Name))
			return err
		}
		waitFor(ctx, func() bool {
			resp, err := httpGet(fmt.Sprintf("https://localhost:%d%s", port, endpoint), token)
			if err != nil {
				return false
			}
			defer resp.Body.Close()

			if resp.StatusCode == http.StatusOK {
				b, err := io.ReadAll(resp.Body)
				if err != nil {
					return false
				}
				sc.metricsResponse[p.Name] = string(b)
				return true
			}
			b, err := io.ReadAll(resp.Body)
			if err != nil {
				return false
			}
			logger.V(1).Info("failed to get metrics", "pod", p.Name, "response", string(b))
			return false
		})
		if err := portForwardCmd.Process.Kill(); err != nil {
			return err
		}
		if _, err := portForwardCmd.Process.Wait(); err != nil {
			return err
		}
	}

	return nil
}

// CatalogIsUpdatedToVersion patches the ClusterCatalog's image reference to the specified version tag.
func CatalogIsUpdatedToVersion(name, version string) error {
	ref, err := k8sClient("get", "clustercatalog", fmt.Sprintf("%s-catalog", name), "-o", "jsonpath={.spec.source.image.ref}")
	if err != nil {
		return err
	}
	i := strings.LastIndexByte(ref, ':')
	if i == -1 {
		return fmt.Errorf("failed to find tag in image reference %s", ref)
	}
	base := ref[:i]
	patch := map[string]any{
		"spec": map[string]any{
			"source": map[string]any{
				"image": map[string]any{
					"ref": fmt.Sprintf("%s:%s", base, version),
				},
			},
		},
	}
	pb, err := json.Marshal(patch)
	if err != nil {
		return err
	}
	_, err = k8sClient("patch", "clustercatalog", fmt.Sprintf("%s-catalog", name), "--type", "merge", "-p", string(pb))
	return err
}

// CatalogServesBundles applies the ClusterCatalog YAML template to create a catalog that serves bundles.
func CatalogServesBundles(ctx context.Context, catalogName string) error {
	if err := deployImageRegistry(); err != nil {
		return err
	}
	sc := scenarioCtx(ctx)
	sc.clusterCatalogName = fmt.Sprintf("%s-catalog", catalogName)

	_, thisFile, _, _ := runtime.Caller(0)
	yamlContent, err := os.ReadFile(filepath.Join(filepath.Dir(thisFile), "testdata", fmt.Sprintf("%s-catalog-template.yaml", catalogName)))
	if err != nil {
		return fmt.Errorf("failed to read catalog yaml: %v", err)
	}

	_, err = k8scliWithInput(substituteScenarioVars(string(yamlContent), sc), "apply", "-f", "-")
	if err != nil {
		return fmt.Errorf("failed to apply catalog: %v", err)
	}

	return nil
}

// TagCatalogImage adds a new tag to a catalog container image in the local registry using crane.
func TagCatalogImage(name, oldTag, newTag string) error {
	imageRef := fmt.Sprintf("%s/%s", os.Getenv("LOCAL_REGISTRY_HOST"), fmt.Sprintf("e2e/%s-catalog:%s", name, oldTag))
	return crane.Tag(imageRef, newTag, crane.Insecure)
}

// CatalogIsDeleted deletes the named ClusterCatalog resource and waits for it to be removed.
func CatalogIsDeleted(ctx context.Context, catalogName string) error {
	catalogFullName := fmt.Sprintf("%s-catalog", catalogName)
	_, err := k8sClient("delete", "clustercatalog", catalogFullName, "--ignore-not-found=true", "--wait=true")
	if err != nil {
		return fmt.Errorf("failed to delete catalog: %v", err)
	}
	return nil
}

// PrometheusMetricsAreReturned validates that each pod's stored metrics response is non-empty and parses as valid Prometheus text format.
func PrometheusMetricsAreReturned(ctx context.Context) error {
	sc := scenarioCtx(ctx)
	for podName, mr := range sc.metricsResponse {
		if mr == "" {
			return fmt.Errorf("metrics response is empty for pod %s", podName)
		}
		parser := expfmt.NewTextParser(model.UTF8Validation)
		metricsFamilies, err := parser.TextToMetricFamilies(strings.NewReader(mr))
		if err != nil {
			return fmt.Errorf("failed to parse metrics response for pod %s: %v", podName, err)
		}
		if len(metricsFamilies) == 0 {
			return fmt.Errorf("metrics response does not contain any metrics for pod %s", podName)
		}
	}
	return nil
}

// OperatorTargetNamespace asserts that the operator deployment has the expected olm.targetNamespaces annotation.
func OperatorTargetNamespace(ctx context.Context, operator, namespace string) error {
	sc := scenarioCtx(ctx)
	namespace = substituteScenarioVars(namespace, sc)
	raw, err := k8sClient("get", "deployment", "-n", sc.namespace, operator, "-o", "json")
	if err != nil {
		return err
	}
	d := &appsv1.Deployment{}
	if err := json.Unmarshal([]byte(raw), d); err != nil {
		return err
	}

	if tns := d.Spec.Template.Annotations["olm.targetNamespaces"]; tns != namespace {
		return fmt.Errorf("expected target namespace %s, got %s", namespace, tns)
	}
	return nil
}

// MarkTestOperatorNotReady controls the test-operator's readiness probe by removing or creating the readiness file in its pod.
func MarkTestOperatorNotReady(ctx context.Context, state string) error {
	sc := scenarioCtx(ctx)
	v, err := k8sClient("get", "deployment", "-n", sc.namespace, "test-operator", "-o", "jsonpath={.spec.selector.matchLabels}")
	if err != nil {
		return err
	}
	var labels map[string]string
	if err := json.Unmarshal([]byte(v), &labels); err != nil {
		return err
	}
	podNameCmd := []string{"get", "pod", "-n", sc.namespace, "-o", "jsonpath={.items[0].metadata.name}"}
	for k, v := range labels {
		podNameCmd = append(podNameCmd, fmt.Sprintf("--selector=%s=%s", k, v))
	}
	podName, err := k8sClient(podNameCmd...)
	if err != nil {
		return err
	}
	var op string
	switch state {
	case "not ready":
		op = "rm"
	case "ready":
		op = "touch"
	default:
		return fmt.Errorf("invalid state %s", state)
	}
	_, err = k8sClient("exec", podName, "-n", sc.namespace, "--", op, "/var/www/ready")
	return err
}

// SetCRDFieldMinValue patches a CRD to set the minimum value for a field.
// jsonPath is in the format ".spec.fieldName" and gets converted to the CRD schema path.
func SetCRDFieldMinValue(_ context.Context, resourceType, jsonPath string, minValue int) error {
	var crdName string
	switch resourceType {
	case "ClusterExtension":
		crdName = "clusterextensions.olm.operatorframework.io"
	case "ClusterObjectSet":
		crdName = "clusterobjectsets.olm.operatorframework.io"
	default:
		return fmt.Errorf("unsupported resource type: %s", resourceType)
	}

	// Convert JSON path like ".spec.progressDeadlineMinutes" to CRD schema path
	// e.g., ".spec.progressDeadlineMinutes" -> "properties/spec/properties/progressDeadlineMinutes"
	parts := strings.Split(strings.TrimPrefix(jsonPath, "."), ".")
	schemaParts := make([]string, 0, 2*len(parts))
	for _, part := range parts {
		schemaParts = append(schemaParts, "properties", part)
	}
	patchPath := fmt.Sprintf("/spec/versions/0/schema/openAPIV3Schema/%s/minimum", strings.Join(schemaParts, "/"))

	patch := fmt.Sprintf(`[{"op": "replace", "path": "%s", "value": %d}]`, patchPath, minValue)
	_, err := k8sClient("patch", "crd", crdName, "--type=json", "-p", patch)
	return err
}

// templateYaml applies values to the template located in templatePath and returns the result or any errors reading
// the template file
func templateYaml(templatePath string, values map[string]string) (string, error) {
	yamlContent, err := os.ReadFile(templatePath)
	if err != nil {
		return "", fmt.Errorf("failed to read template file '%s': %v", templatePath, err)
	}
	return templateContent(string(yamlContent), values), nil
}

// templateContent applies values to content and returns the result
func templateContent(content string, values map[string]string) string {
	m := func(k string) string {
		if v, found := values[k]; found {
			return v
		}
		return ""
	}

	// Replace template variables
	return os.Expand(content, m)
}

// extendMap extends m with the key/values in keyValue, which is expected to be of even size
func extendMap(m map[string]string, keyValue ...string) map[string]string {
	if len(keyValue) > 0 {
		for i := 0; i < len(keyValue); i += 2 {
			m[keyValue[i]] = keyValue[i+1]
		}
	}
	return m
}

func getResource(kind string, name string, namespace string) (*unstructured.Unstructured, error) {
	out, err := k8sClient("get", kind, name, "-n", namespace, "-o", "yaml")
	if err != nil {
		return nil, err
	}
	obj, err := toUnstructured(out)
	if err != nil {
		return nil, err
	}
	return obj, nil
}

// listExtensionResources returns a slice of client.Object containing all resources for a ClusterExtension
// this method is best called when the extension has been installed successfully. An error is returned if there was
// any issue in determining the extension's resources.
func listExtensionResources(extName string) ([]client.Object, error) {
	if enabled, found := featureGates[features.BoxcutterRuntime]; found && enabled {
		return listExtensionRevisionResources(extName)
	}
	return listHelmReleaseResources(extName)
}

// listHelmReleaseResources returns a slice of client.Object containing all resources for a ClusterExtension's
// Helm release. Note: The current implementation does not support release secrets chunked across multiple secrets
func listHelmReleaseResources(extName string) ([]client.Object, error) {
	secret, err := helmReleaseSecretForExtension(extName)
	if err != nil {
		return nil, fmt.Errorf("failed to get helm release secret for extension %s: %w", extName, err)
	}

	rel, err := helmReleaseFromSecret(secret)
	if err != nil {
		return nil, fmt.Errorf("failed to get helm release from secret for cluster extension '%s': %w", extName, err)
	}

	objs, err := collectHelmReleaseObjects(rel)
	if err != nil {
		return nil, fmt.Errorf("failed to collect helm release objects for cluster extension '%s': %w", extName, err)
	}
	return objs, nil
}

// helmReleaseSecretForExtension returns the Helm release secret for the extension with name extName
func helmReleaseSecretForExtension(extName string) (*corev1.Secret, error) {
	out, err := k8sClient("get", "secrets", "-n", olmNamespace,
		"-l", fmt.Sprintf("name=%s,status=deployed", extName),
		"--field-selector", "type=operatorframework.io/index.v1", "-o", "json")
	if err != nil {
		return nil, err
	}
	if strings.TrimSpace(out) == "" {
		return nil, err
	}

	var secretList corev1.SecretList
	if err = json.Unmarshal([]byte(out), &secretList); err != nil {
		return nil, err
	}
	if len(secretList.Items) != 1 {
		return nil, err
	}
	return &secretList.Items[0], nil
}

// helmReleaseFromSecret returns the Helm Release object encoded in the secret. Note: this function does not yet support
// releases chunked over multiple Secrets
func helmReleaseFromSecret(secret *corev1.Secret) (*release.Release, error) {
	// OLM uses a custom release backend that compresses the release data
	gzReader, err := gzip.NewReader(strings.NewReader(string(secret.Data["chunk"])))
	if err != nil {
		return nil, err
	}
	defer gzReader.Close()

	releaseJsonBytes, err := io.ReadAll(gzReader)
	if err != nil {
		return nil, err
	}

	var rel release.Release
	if err = json.Unmarshal(releaseJsonBytes, &rel); err != nil {
		return nil, err
	}
	return &rel, nil
}

// collectHelmReleaseObjects returns a slice of client.Object containing the manifests in rel
func collectHelmReleaseObjects(rel *release.Release) ([]client.Object, error) {
	result := k8sresource.NewLocalBuilder().Flatten().Unstructured().Stream(strings.NewReader(rel.Manifest), rel.Name).Do()
	if err := result.Err(); err != nil {
		return nil, err
	}
	infos, err := result.Infos()
	if err != nil {
		return nil, err
	}

	objs := make([]client.Object, 0, len(infos))
	for _, info := range infos {
		clientObject, ok := info.Object.(client.Object)
		if !ok {
			return nil, fmt.Errorf("object of type %T does not implement client.Object", info.Object)
		}
		objs = append(objs, clientObject)
	}
	return objs, nil
}

// listExtensionRevisionResources lists objects in the phases of the latest active revision
func listExtensionRevisionResources(extName string) ([]client.Object, error) {
	rev, err := latestActiveRevisionForExtension(extName)
	if err != nil {
		return nil, fmt.Errorf("failed to get latest active revision for extension %s: %w", extName, err)
	}

	var objs []client.Object
	for i := range rev.Spec.Phases {
		phase := &rev.Spec.Phases[i]
		for j := range phase.Objects {
			specObj := &phase.Objects[j]
			switch {
			case specObj.Ref.Name != "":
				resolved, err := resolveObjectRef(specObj.Ref)
				if err != nil {
					return nil, fmt.Errorf("resolving ref in phase %q object %d: %w", phase.Name, j, err)
				}
				objs = append(objs, resolved)
			case len(specObj.Object.Object) > 0:
				objs = append(objs, &specObj.Object)
			default:
				return nil, fmt.Errorf("object %d in phase %q has neither object nor ref", j, phase.Name)
			}
		}
	}

	return objs, nil
}

// resolveObjectRef fetches an object from a Secret ref using kubectl.
func resolveObjectRef(ref ocv1.ObjectSourceRef) (*unstructured.Unstructured, error) {
	out, err := k8sClient("get", "secret", ref.Name, "-n", ref.Namespace, "-o", "json")
	if err != nil {
		return nil, fmt.Errorf("getting Secret %s/%s: %w", ref.Namespace, ref.Name, err)
	}
	var secret corev1.Secret
	if err := json.Unmarshal([]byte(out), &secret); err != nil {
		return nil, fmt.Errorf("unmarshaling Secret %s/%s: %w", ref.Namespace, ref.Name, err)
	}
	data, ok := secret.Data[ref.Key]
	if !ok {
		return nil, fmt.Errorf("key %q not found in Secret %s/%s", ref.Key, ref.Namespace, ref.Name)
	}
	// Auto-detect gzip compression (magic bytes 0x1f 0x8b)
	if len(data) >= 2 && data[0] == 0x1f && data[1] == 0x8b {
		reader, err := gzip.NewReader(bytes.NewReader(data))
		if err != nil {
			return nil, fmt.Errorf("creating gzip reader for key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)
		}
		defer reader.Close()
		const maxDecompressedSize = 10 * 1024 * 1024 // 10 MiB
		limited := io.LimitReader(reader, maxDecompressedSize+1)
		decompressed, err := io.ReadAll(limited)
		if err != nil {
			return nil, fmt.Errorf("decompressing key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)
		}
		if len(decompressed) > maxDecompressedSize {
			return nil, fmt.Errorf("decompressed data for key %q in Secret %s/%s exceeds maximum size (%d bytes)", ref.Key, ref.Namespace, ref.Name, maxDecompressedSize)
		}
		data = decompressed
	}
	obj := &unstructured.Unstructured{}
	if err := json.Unmarshal(data, &obj.Object); err != nil {
		return nil, fmt.Errorf("unmarshaling object from key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)
	}
	return obj, nil
}

// latestActiveRevisionForExtension returns the latest active revision for the extension called extName
func latestActiveRevisionForExtension(extName string) (*ocv1.ClusterObjectSet, error) {
	out, err := k8sClient("get", "clusterobjectsets", "-l", fmt.Sprintf("olm.operatorframework.io/owner-name=%s", extName), "-o", "json")
	if err != nil {
		return nil, fmt.Errorf("error listing revisions for extension '%s': %w", extName, err)
	}
	if strings.TrimSpace(out) == "" {
		return nil, fmt.Errorf("no revisions found for extension '%s'", extName)
	}
	var revisionList ocv1.ClusterObjectSetList
	if err := json.Unmarshal([]byte(out), &revisionList); err != nil {
		return nil, fmt.Errorf("error unmarshalling revisions for extension '%s': %w", extName, err)
	}

	var latest *ocv1.ClusterObjectSet
	for i := range revisionList.Items {
		rev := &revisionList.Items[i]
		if rev.Spec.LifecycleState != ocv1.ClusterObjectSetLifecycleStateActive {
			continue
		}
		if latest == nil || rev.Spec.Revision > latest.Spec.Revision {
			latest = rev
		}
	}

	if latest == nil {
		return nil, fmt.Errorf("no active revisions found for extension '%s'", extName)
	}

	return latest, nil
}

// AnnotationsAreAdded adds annotations from a data table to a resource using kubectl annotate.
// The table must have "key" and "value" columns.
// Only supports namespaced resources (always uses sc.namespace).
func AnnotationsAreAdded(ctx context.Context, resourceName string, table *godog.Table) error {
	sc := scenarioCtx(ctx)
	resourceName = substituteScenarioVars(resourceName, sc)

	pairs, err := parseKeyValueTable(table)
	if err != nil {
		return fmt.Errorf("invalid annotations table: %w", err)
	}

	kind, name, ok := strings.Cut(resourceName, "/")
	if !ok {
		return fmt.Errorf("invalid resource name format: %q (expected kind/name)", resourceName)
	}

	args := []string{"annotate", kind, name}
	for k, v := range pairs {
		args = append(args, fmt.Sprintf("%s=%s", k, v))
	}
	args = append(args, "--overwrite", "-n", sc.namespace)
	if _, err := k8sClient(args...); err != nil {
		return fmt.Errorf("failed to annotate %s: %w; stderr: %s", resourceName, err, stderrOutput(err))
	}
	return nil
}

// LabelsAreAdded adds labels from a data table to a resource using kubectl label.
// The table must have "key" and "value" columns.
// Only supports namespaced resources (always uses sc.namespace).
func LabelsAreAdded(ctx context.Context, resourceName string, table *godog.Table) error {
	sc := scenarioCtx(ctx)
	resourceName = substituteScenarioVars(resourceName, sc)

	pairs, err := parseKeyValueTable(table)
	if err != nil {
		return fmt.Errorf("invalid labels table: %w", err)
	}

	kind, name, ok := strings.Cut(resourceName, "/")
	if !ok {
		return fmt.Errorf("invalid resource name format: %q (expected kind/name)", resourceName)
	}

	args := []string{"label", kind, name}
	for k, v := range pairs {
		args = append(args, fmt.Sprintf("%s=%s", k, v))
	}
	args = append(args, "--overwrite", "-n", sc.namespace)
	if _, err := k8sClient(args...); err != nil {
		return fmt.Errorf("failed to label %s: %w; stderr: %s", resourceName, err, stderrOutput(err))
	}
	return nil
}

// ResourceHasAnnotations waits for a resource to have all annotations specified in the data table.
// The table must have "key" and "value" columns.
// Only supports namespaced resources (always uses sc.namespace).
func ResourceHasAnnotations(ctx context.Context, resourceName string, table *godog.Table) error {
	sc := scenarioCtx(ctx)
	resourceName = substituteScenarioVars(resourceName, sc)

	expected, err := parseKeyValueTable(table)
	if err != nil {
		return fmt.Errorf("invalid annotations table: %w", err)
	}

	kind, name, ok := strings.Cut(resourceName, "/")
	if !ok {
		return fmt.Errorf("invalid resource name format: %q (expected kind/name)", resourceName)
	}

	waitFor(ctx, func() bool {
		out, err := k8sClient("get", kind, name, "-n", sc.namespace, "-o", "json")
		if err != nil {
			return false
		}
		var obj unstructured.Unstructured
		if err := json.Unmarshal([]byte(out), &obj); err != nil {
			return false
		}
		annotations := obj.GetAnnotations()
		for k, v := range expected {
			if actual, found := annotations[k]; !found || actual != v {
				logger.V(1).Info("Annotation not yet present or value mismatch", "resource", resourceName, "key", k, "expected", v, "actual", actual)
				return false
			}
		}
		return true
	})
	return nil
}

// matchLabels checks that actual labels contain all expected key-value pairs.
// Returns the first mismatched key, the actual value, and false if a mismatch is found.
func matchLabels(actual, expected map[string]string) (string, string, bool) {
	for k, v := range expected {
		if a, found := actual[k]; !found || a != v {
			return k, a, false
		}
	}
	return "", "", true
}

// ResourceHasLabels waits for a resource to have all labels specified in the data table.
// The table must have "key" and "value" columns.
// Only supports namespaced resources (always uses sc.namespace).
func ResourceHasLabels(ctx context.Context, resourceName string, table *godog.Table) error {
	sc := scenarioCtx(ctx)
	resourceName = substituteScenarioVars(resourceName, sc)

	expected, err := parseKeyValueTable(table)
	if err != nil {
		return fmt.Errorf("invalid labels table: %w", err)
	}

	kind, name, ok := strings.Cut(resourceName, "/")
	if !ok {
		return fmt.Errorf("invalid resource name format: %q (expected kind/name)", resourceName)
	}

	waitFor(ctx, func() bool {
		out, err := k8sClient("get", kind, name, "-n", sc.namespace, "-o", "json")
		if err != nil {
			return false
		}
		var obj unstructured.Unstructured
		if err := json.Unmarshal([]byte(out), &obj); err != nil {
			return false
		}
		if key, got, ok := matchLabels(obj.GetLabels(), expected); !ok {
			logger.V(1).Info("Label not yet present or value mismatch", "resource", resourceName, "key", key, "expected", expected[key], "actual", got)
			return false
		}
		return true
	})
	return nil
}

// nestedString traverses a nested map[string]interface{} by the given keys
// and returns the leaf value as a string.
func nestedString(obj map[string]interface{}, keys ...string) (string, bool) {
	if len(keys) == 0 {
		return "", false
	}
	current := obj
	for _, k := range keys[:len(keys)-1] {
		next, ok := current[k].(map[string]interface{})
		if !ok {
			return "", false
		}
		current = next
	}
	v, ok := current[keys[len(keys)-1]].(string)
	return v, ok
}

// parseKeyValueTable extracts key-value pairs from a godog data table.
// The table must have a header row with "key" and "value" columns.
func parseKeyValueTable(table *godog.Table) (map[string]string, error) {
	if len(table.Rows) < 2 {
		return nil, fmt.Errorf("table must have a header row and at least one data row")
	}

	header := table.Rows[0]
	keyIdx, valueIdx := -1, -1
	for i, cell := range header.Cells {
		switch cell.Value {
		case "key":
			keyIdx = i
		case "value":
			valueIdx = i
		}
	}
	if keyIdx == -1 || valueIdx == -1 {
		return nil, fmt.Errorf("table must have 'key' and 'value' columns")
	}

	maxIdx := keyIdx
	if valueIdx > maxIdx {
		maxIdx = valueIdx
	}

	result := make(map[string]string, len(table.Rows)-1)
	for rowIdx, row := range table.Rows[1:] {
		if len(row.Cells) <= maxIdx {
			return nil, fmt.Errorf("table row %d has %d cells, expected at least %d", rowIdx+2, len(row.Cells), maxIdx+1)
		}
		result[row.Cells[keyIdx].Value] = row.Cells[valueIdx].Value
	}
	return result, nil
}

// RolloutRestartIsPerformed runs "kubectl rollout restart deployment/<name>".
// See: https://github.com/operator-framework/operator-lifecycle-manager/issues/3392
func RolloutRestartIsPerformed(ctx context.Context, resourceName string) error {
	sc := scenarioCtx(ctx)
	resourceName = substituteScenarioVars(resourceName, sc)

	kind, deploymentName, ok := strings.Cut(resourceName, "/")
	if !ok {
		return fmt.Errorf("invalid resource name format: %q (expected kind/name)", resourceName)
	}

	if kind != "deployment" {
		return fmt.Errorf("only deployment resources are supported for rollout restart, got: %q", kind)
	}

	// Run kubectl rollout restart to add the restart annotation.
	// This is the real command users run, so we test actual user behavior.
	out, err := k8sClient("rollout", "restart", resourceName, "-n", sc.namespace)
	if err != nil {
		return fmt.Errorf("failed to rollout restart %s: %w; stderr: %s", resourceName, err, stderrOutput(err))
	}

	logger.V(1).Info("Rollout restart initiated", "deployment", deploymentName, "output", out)

	return nil
}

// DeploymentPodTemplateHasAnnotation waits for the deployment's pod template to have
// the given annotation key. Uses JSON parsing to avoid JSONPath issues with dots
// in annotation keys. Polls with timeout.
func DeploymentPodTemplateHasAnnotation(ctx context.Context, deploymentName, annotationKey string) error {
	sc := scenarioCtx(ctx)
	deploymentName = substituteScenarioVars(deploymentName, sc)

	waitFor(ctx, func() bool {
		out, err := k8sClient("get", "deployment", deploymentName, "-n", sc.namespace, "-o", "json")
		if err != nil {
			return false
		}
		var d appsv1.Deployment
		if err := json.Unmarshal([]byte(out), &d); err != nil {
			return false
		}
		if v, found := d.Spec.Template.Annotations[annotationKey]; found {
			logger.V(1).Info("Pod template annotation found", "deployment", deploymentName, "key", annotationKey, "value", v)
			return true
		}
		logger.V(1).Info("Pod template annotation not yet present", "deployment", deploymentName, "key", annotationKey, "annotations", d.Spec.Template.Annotations)
		return false
	})
	return nil
}

// TriggerClusterExtensionReconciliation patches the ClusterExtension spec to bump
// its metadata generation, forcing the controller to run a full reconciliation loop.
// Use with "ClusterExtension has been reconciled the latest generation" to confirm
// the controller processed the change before asserting on the cluster state.
//
// We flip install.preflight.crdUpgradeSafety.enforcement between "None" and "Strict"
// because it is a real spec field that the API server will persist (unlike unknown
// fields, which are pruned by structural schemas). Toggling ensures that each call
// results in a spec change, reliably bumping .metadata.generation.
func TriggerClusterExtensionReconciliation(ctx context.Context) error {
	sc := scenarioCtx(ctx)

	out, err := k8sClient("get", "clusterextension", sc.clusterExtensionName, "-o", "json")
	if err != nil {
		return fmt.Errorf("failed to get ClusterExtension %s: %w; stderr: %s", sc.clusterExtensionName, err, stderrOutput(err))
	}

	var obj map[string]interface{}
	if err := json.Unmarshal([]byte(out), &obj); err != nil {
		return fmt.Errorf("failed to unmarshal ClusterExtension %s JSON: %w", sc.clusterExtensionName, err)
	}

	currentEnforcement, _ := nestedString(obj, "spec", "install", "preflight", "crdUpgradeSafety", "enforcement")

	newEnforcement := "None"
	if currentEnforcement == "None" {
		newEnforcement = "Strict"
	}

	payload := fmt.Sprintf(`{"spec":{"install":{"preflight":{"crdUpgradeSafety":{"enforcement":%q}}}}}`, newEnforcement)
	_, err = k8sClient("patch", "clusterextension", sc.clusterExtensionName,
		"--type=merge",
		"-p", payload)
	if err != nil {
		return fmt.Errorf("failed to trigger reconciliation for ClusterExtension %s: %w; stderr: %s", sc.clusterExtensionName, err, stderrOutput(err))
	}
	return nil
}

// DeploymentRolloutIsComplete verifies that a deployment rollout has completed successfully.
// This ensures the new ReplicaSet is fully scaled up and the old one is scaled down.
func DeploymentRolloutIsComplete(ctx context.Context, deploymentName string) error {
	sc := scenarioCtx(ctx)
	deploymentName = substituteScenarioVars(deploymentName, sc)

	waitFor(ctx, func() bool {
		out, err := k8sClient("rollout", "status", "deployment/"+deploymentName, "-n", sc.namespace, "--watch=false")
		if err != nil {
			logger.V(1).Info("Failed to get rollout status", "deployment", deploymentName, "error", err)
			return false
		}
		// Successful rollout shows "successfully rolled out"
		if strings.Contains(out, "successfully rolled out") {
			logger.V(1).Info("Rollout completed successfully", "deployment", deploymentName)
			return true
		}
		logger.V(1).Info("Rollout not yet complete", "deployment", deploymentName, "status", out)
		return false
	})
	return nil
}

// DeploymentHasReplicaSets verifies that a deployment has the expected number of ReplicaSets
// and that at least one owned ReplicaSet is active with pods running.
func DeploymentHasReplicaSets(ctx context.Context, deploymentName string, expectedCount int) error {
	sc := scenarioCtx(ctx)
	deploymentName = substituteScenarioVars(deploymentName, sc)

	waitFor(ctx, func() bool {
		deploymentOut, err := k8sClient("get", "deployment", deploymentName, "-n", sc.namespace, "-o", "json")
		if err != nil {
			logger.V(1).Info("Failed to get deployment", "deployment", deploymentName, "error", err)
			return false
		}

		var deployment appsv1.Deployment
		if err := json.Unmarshal([]byte(deploymentOut), &deployment); err != nil {
			logger.V(1).Info("Failed to parse deployment", "error", err)
			return false
		}

		args := []string{"get", "rs", "-n", sc.namespace}
		if deployment.Spec.Selector != nil && len(deployment.Spec.Selector.MatchLabels) > 0 {
			var parts []string
			for k, v := range deployment.Spec.Selector.MatchLabels {
				parts = append(parts, fmt.Sprintf("%s=%s", k, v))
			}
			args = append(args, "-l", strings.Join(parts, ","))
		}
		args = append(args, "-o", "json")
		out, err := k8sClient(args...)
		if err != nil {
			logger.V(1).Info("Failed to get ReplicaSets", "deployment", deploymentName, "error", err)
			return false
		}

		var rsCandidates struct {
			Items []appsv1.ReplicaSet `json:"items"`
		}
		if err := json.Unmarshal([]byte(out), &rsCandidates); err != nil {
			logger.V(1).Info("Failed to parse ReplicaSets", "error", err)
			return false
		}

		var rsList []appsv1.ReplicaSet
		for _, rs := range rsCandidates.Items {
			for _, owner := range rs.OwnerReferences {
				if owner.Kind == "Deployment" && owner.UID == deployment.UID {
					rsList = append(rsList, rs)
					break
				}
			}
		}

		if len(rsList) != expectedCount {
			logger.V(1).Info("ReplicaSet count does not match expected value yet", "deployment", deploymentName, "current", len(rsList), "expected", expectedCount)
			return false
		}

		// Verify at least one ReplicaSet has active replicas
		hasActiveRS := false
		for _, rs := range rsList {
			if rs.Status.Replicas > 0 && rs.Status.ReadyReplicas > 0 {
				hasActiveRS = true
				logger.V(1).Info("Found active ReplicaSet", "name", rs.Name, "replicas", rs.Status.Replicas, "ready", rs.Status.ReadyReplicas)
			}
		}

		if !hasActiveRS {
			logger.V(1).Info("No active ReplicaSet found yet", "deployment", deploymentName)
			return false
		}

		logger.V(1).Info("ReplicaSet verification passed", "deployment", deploymentName, "count", len(rsList))
		return true
	})
	return nil
}