diff --git a/Makefile b/Makefile
index bc007df5dd0c..aa2a309cd65a 100644
--- a/Makefile
+++ b/Makefile
@@ -562,11 +562,15 @@ generate-doctoc:
 	TRACE=$(TRACE) ./hack/generate-doctoc.sh
 
 .PHONY: generate-e2e-templates
-generate-e2e-templates: $(KUSTOMIZE) $(addprefix generate-e2e-templates-, v1.0 v1.5 v1.6 main) ## Generate cluster templates for all versions
+generate-e2e-templates: $(KUSTOMIZE) $(addprefix generate-e2e-templates-, v0.4 v1.0 v1.5 v1.6 main) ## Generate cluster templates for all versions
 
 DOCKER_TEMPLATES := test/e2e/data/infrastructure-docker
 INMEMORY_TEMPLATES := test/e2e/data/infrastructure-inmemory
 
+.PHONY: generate-e2e-templates-v0.4
+generate-e2e-templates-v0.4: $(KUSTOMIZE)
+	$(KUSTOMIZE) build $(DOCKER_TEMPLATES)/v0.4/cluster-template --load-restrictor LoadRestrictionsNone > $(DOCKER_TEMPLATES)/v0.4/cluster-template.yaml
+
 .PHONY: generate-e2e-templates-v1.0
 generate-e2e-templates-v1.0: $(KUSTOMIZE)
 	$(KUSTOMIZE) build $(DOCKER_TEMPLATES)/v1.0/cluster-template --load-restrictor LoadRestrictionsNone > $(DOCKER_TEMPLATES)/v1.0/cluster-template.yaml
diff --git a/test/e2e/clusterctl_upgrade.go b/test/e2e/clusterctl_upgrade.go
index b563b9ebb6d4..67f0aeab9ca1 100644
--- a/test/e2e/clusterctl_upgrade.go
+++ b/test/e2e/clusterctl_upgrade.go
@@ -25,15 +25,19 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	apiruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/discovery"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -118,7 +122,23 @@ type ClusterctlUpgradeSpecInput struct {
 	WorkloadFlavor              string
 	// WorkloadKubernetesVersion is Kubernetes version used to create the workload cluster, e.g. `v1.25.0`
 	WorkloadKubernetesVersion string
-	// Custom providers can be specified to upgrade to a pre-release or a custom version instead of upgrading to the latest using contact
+
+	// Upgrades allows to define upgrade sequences.
+	// If not set, the test will upgrade once to the v1beta1 contract.
+	// For some examples see clusterctl_upgrade_test.go
+	Upgrades []ClusterctlUpgradeSpecInputUpgrade
+}
+
+// ClusterctlUpgradeSpecInputUpgrade defines an upgrade.
+type ClusterctlUpgradeSpecInputUpgrade struct {
+	// UpgradeWithBinary can be used to set the clusterctl binary to use for the provider upgrade. The spec will interpolate the
+	// strings `{OS}` and `{ARCH}` to `runtime.GOOS` and `runtime.GOARCH` respectively, e.g. https://github.com/kubernetes-sigs/cluster-api/releases/download/v0.3.23/clusterctl-{OS}-{ARCH}
+	// If not set, the test will use the ApplyUpgrade function of the clusterctl library.
+	WithBinary string
+	// Contract defines the contract to upgrade to.
+	// Either the contract *or* the *Provider fields should be defined
+	// For some examples see clusterctl_upgrade_test.go
+	Contract                  string
 	CoreProvider              string
 	BootstrapProviders        []string
 	ControlPlaneProviders     []string
@@ -213,6 +233,15 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 			initKubernetesVersion = input.E2EConfig.GetVariable(initWithKubernetesVersion)
 		}
 
+		if len(input.Upgrades) == 0 {
+			// Upgrade once to v1beta1 if no upgrades are specified.
+			input.Upgrades = []ClusterctlUpgradeSpecInputUpgrade{
+				{
+					Contract: clusterv1.GroupVersion.Version,
+				},
+			}
+		}
+
 		Expect(input.E2EConfig.Variables).To(HaveKey(KubernetesVersion))
 		Expect(os.MkdirAll(input.ArtifactFolder, 0750)).To(Succeed(), "Invalid argument. input.ArtifactFolder can't be created for %s spec", specName)
 
@@ -274,17 +303,11 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 		// Get a ClusterProxy so we can interact with the workload cluster
 		managementClusterProxy = input.BootstrapClusterProxy.GetWorkloadCluster(ctx, cluster.Namespace, cluster.Name, framework.WithMachineLogCollector(input.BootstrapClusterProxy.GetLogCollector()))
 
-		// Download the older clusterctl version to be used for setting up the management cluster to be upgraded
-		log.Logf("Downloading clusterctl binary from %s", initClusterctlBinaryURL)
-		clusterctlBinaryPath := downloadToTmpFile(ctx, initClusterctlBinaryURL)
+		// Download the clusterctl version that should be used to initially set up the management cluster (which is later upgraded).
+		Byf("Downloading clusterctl binary from %s", initClusterctlBinaryURL)
+		clusterctlBinaryPath, clusterctlConfigPath := setupClusterctl(ctx, initClusterctlBinaryURL, input.ClusterctlConfigPath)
 		defer os.Remove(clusterctlBinaryPath) // clean up
 
-		err := os.Chmod(clusterctlBinaryPath, 0744) //nolint:gosec
-		Expect(err).ToNot(HaveOccurred(), "failed to chmod temporary file")
-
-		// Adjusts the clusterctlConfigPath in case the clusterctl version <= v1.3 (thus using a config file with only the providers supported in those versions)
-		clusterctlConfigPath := clusterctl.AdjustConfigPathForBinary(clusterctlBinaryPath, input.ClusterctlConfigPath)
-
 		By("Initializing the workload cluster with older versions of providers")
 
 		if input.PreInit != nil {
@@ -336,11 +359,6 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 
 		By("THE MANAGEMENT CLUSTER WITH THE OLDER VERSION OF PROVIDERS IS UP&RUNNING!")
 
-		machineCRD := &apiextensionsv1.CustomResourceDefinition{}
-		if err := managementClusterProxy.GetClient().Get(ctx, client.ObjectKey{Name: "machines.cluster.x-k8s.io"}, machineCRD); err != nil {
-			Expect(err).ToNot(HaveOccurred(), "failed to retrieve a machine CRD")
-		}
-
 		Byf("Creating a namespace for hosting the %s test workload cluster", specName)
 		testNamespace, testCancelWatches = framework.CreateNamespaceAndWatchEvents(ctx, framework.CreateNamespaceAndWatchEventsInput{
 			Creator:   managementClusterProxy.GetClient(),
@@ -361,6 +379,7 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 		// so we are getting a template using the downloaded version of clusterctl, applying it, and wait for machines to be provisioned.
 
 		workloadClusterName = fmt.Sprintf("%s-%s", specName, util.RandomString(6))
+		workloadClusterNamespace := testNamespace.Name
 		kubernetesVersion := input.WorkloadKubernetesVersion
 		if kubernetesVersion == "" {
 			kubernetesVersion = input.E2EConfig.GetVariable(KubernetesVersion)
@@ -380,7 +399,7 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 			// select template
 			Flavor: input.WorkloadFlavor,
 			// define template variables
-			Namespace:                testNamespace.Name,
+			Namespace:                workloadClusterNamespace,
 			ClusterName:              workloadClusterName,
 			KubernetesVersion:        kubernetesVersion,
 			ControlPlaneMachineCount: controlPlaneMachineCount,
@@ -396,41 +415,35 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 
 		if input.PreWaitForCluster != nil {
 			By("Running PreWaitForCluster steps against the management cluster")
-			input.PreWaitForCluster(managementClusterProxy, testNamespace.Name, workloadClusterName)
+			input.PreWaitForCluster(managementClusterProxy, workloadClusterNamespace, workloadClusterName)
 		}
 
-		workloadCluster := framework.DiscoveryAndWaitForCluster(ctx, framework.DiscoveryAndWaitForClusterInput{
-			Getter:    managementClusterProxy.GetClient(),
-			Namespace: testNamespace.Name,
-			Name:      workloadClusterName,
-		}, input.E2EConfig.GetIntervals(specName, "wait-cluster")...)
-
-		expectedMachineCount := *controlPlaneMachineCount + calculateExpectedWorkerCount(ctx, managementClusterProxy.GetClient(), input.E2EConfig, specName, workloadCluster)
+		coreCAPIStorageVersion := getCoreCAPIStorageVersion(ctx, managementClusterProxy.GetClient())
 
-		// Build GroupVersionKind for Machine resources
-		machineListGVK := schema.GroupVersionKind{
-			Group: machineCRD.Spec.Group,
-			Kind:  machineCRD.Spec.Names.ListKind,
-		}
+		// Note: We have to use unstructured here as the Cluster could be e.g. v1alpha3 / v1alpha4 / v1beta1.
+		workloadClusterUnstructured := discoveryAndWaitForCluster(ctx, discoveryAndWaitForClusterInput{
+			Client:                 managementClusterProxy.GetClient(),
+			CoreCAPIStorageVersion: coreCAPIStorageVersion,
+			Namespace:              workloadClusterNamespace,
+			Name:                   workloadClusterName,
+		}, input.E2EConfig.GetIntervals(specName, "wait-cluster")...)
 
-		// Pick the storage version
-		for _, version := range machineCRD.Spec.Versions {
-			if version.Storage {
-				machineListGVK.Version = version.Name
-				break
-			}
-		}
+		expectedMachineCount := *controlPlaneMachineCount + calculateExpectedWorkerCount(ctx, managementClusterProxy.GetClient(), workloadClusterUnstructured, coreCAPIStorageVersion)
 
 		By("Waiting for the machines to exist")
 		Eventually(func() (int64, error) {
 			var n int64
 			machineList := &unstructured.UnstructuredList{}
-			machineList.SetGroupVersionKind(machineListGVK)
+			machineList.SetGroupVersionKind(schema.GroupVersionKind{
+				Group:   clusterv1.GroupVersion.Group,
+				Version: coreCAPIStorageVersion,
+				Kind:    "MachineList",
+			})
 			if err := managementClusterProxy.GetClient().List(
 				ctx,
 				machineList,
-				client.InNamespace(workloadCluster.Namespace),
-				client.MatchingLabels{clusterv1.ClusterNameLabel: workloadCluster.Name},
+				client.InNamespace(workloadClusterNamespace),
+				client.MatchingLabels{clusterv1.ClusterNameLabel: workloadClusterName},
 			); err == nil {
 				for _, m := range machineList.Items {
 					_, found, err := unstructured.NestedMap(m.Object, "status", "nodeRef")
@@ -440,108 +453,169 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 				}
 			}
 			return n, nil
-		}, input.E2EConfig.GetIntervals(specName, "wait-worker-nodes")...).Should(Equal(expectedMachineCount), "Timed out waiting for all machines to be exist")
+		}, input.E2EConfig.GetIntervals(specName, "wait-worker-nodes")...).Should(Equal(expectedMachineCount), "Timed out waiting for all Machines to exist")
 
 		By("THE MANAGEMENT CLUSTER WITH OLDER VERSION OF PROVIDERS WORKS!")
 
-		if input.PreUpgrade != nil {
-			By("Running Pre-upgrade steps against the management cluster")
-			input.PreUpgrade(managementClusterProxy)
-		}
-
-		// Get the workloadCluster before the management cluster is upgraded to make sure that the upgrade did not trigger
-		// any unexpected rollouts.
-		preUpgradeMachineList := &unstructured.UnstructuredList{}
-		preUpgradeMachineList.SetGroupVersionKind(machineListGVK)
-		err = managementClusterProxy.GetClient().List(
-			ctx,
-			preUpgradeMachineList,
-			client.InNamespace(workloadCluster.Namespace),
-			client.MatchingLabels{clusterv1.ClusterNameLabel: workloadCluster.Name},
-		)
-		Expect(err).ToNot(HaveOccurred())
-		// Check if the user want a custom upgrade
-		isCustomUpgrade := input.CoreProvider != "" ||
-			len(input.BootstrapProviders) > 0 ||
-			len(input.ControlPlaneProviders) > 0 ||
-			len(input.InfrastructureProviders) > 0 ||
-			len(input.IPAMProviders) > 0 ||
-			len(input.RuntimeExtensionProviders) > 0 ||
-			len(input.AddonProviders) > 0
-
-		if isCustomUpgrade {
-			By("Upgrading providers to custom versions")
-			clusterctl.UpgradeManagementClusterAndWait(ctx, clusterctl.UpgradeManagementClusterAndWaitInput{
-				ClusterctlConfigPath:      input.ClusterctlConfigPath,
-				ClusterctlVariables:       input.UpgradeClusterctlVariables,
-				ClusterProxy:              managementClusterProxy,
-				CoreProvider:              input.CoreProvider,
-				BootstrapProviders:        input.BootstrapProviders,
-				ControlPlaneProviders:     input.ControlPlaneProviders,
-				InfrastructureProviders:   input.InfrastructureProviders,
-				IPAMProviders:             input.IPAMProviders,
-				RuntimeExtensionProviders: input.RuntimeExtensionProviders,
-				AddonProviders:            input.AddonProviders,
-				LogFolder:                 filepath.Join(input.ArtifactFolder, "clusters", cluster.Name),
-			}, input.E2EConfig.GetIntervals(specName, "wait-controllers")...)
-		} else {
-			By("Upgrading providers to the latest version available")
-			clusterctl.UpgradeManagementClusterAndWait(ctx, clusterctl.UpgradeManagementClusterAndWaitInput{
-				ClusterctlConfigPath: input.ClusterctlConfigPath,
-				ClusterctlVariables:  input.UpgradeClusterctlVariables,
-				ClusterProxy:         managementClusterProxy,
-				Contract:             clusterv1.GroupVersion.Version,
-				LogFolder:            filepath.Join(input.ArtifactFolder, "clusters", cluster.Name),
-			}, input.E2EConfig.GetIntervals(specName, "wait-controllers")...)
-		}
-
-		By("THE MANAGEMENT CLUSTER WAS SUCCESSFULLY UPGRADED!")
-
-		if input.PostUpgrade != nil {
-			By("Running Post-upgrade steps against the management cluster")
-			input.PostUpgrade(managementClusterProxy, testNamespace.Name, managementClusterName)
-		}
+		for i, upgrade := range input.Upgrades {
+			Byf("[%d] Starting upgrade", i)
+			if input.PreUpgrade != nil {
+				Byf("[%d] Running Pre-upgrade steps against the management cluster", i)
+				input.PreUpgrade(managementClusterProxy)
+			}
 
-		// After the upgrade check that there were no unexpected rollouts.
-		log.Logf("Verify there are no unexpected rollouts")
-		Consistently(func() bool {
-			postUpgradeMachineList := &unstructured.UnstructuredList{}
-			postUpgradeMachineList.SetGroupVersionKind(clusterv1.GroupVersion.WithKind("MachineList"))
-			err = managementClusterProxy.GetClient().List(
+			// Get the workloadCluster before the management cluster is upgraded to make sure that the upgrade did not trigger
+			// any unexpected rollouts.
+			preUpgradeMachineList := &unstructured.UnstructuredList{}
+			preUpgradeMachineList.SetGroupVersionKind(schema.GroupVersionKind{
+				Group:   clusterv1.GroupVersion.Group,
+				Version: coreCAPIStorageVersion,
+				Kind:    "MachineList",
+			})
+			err := managementClusterProxy.GetClient().List(
 				ctx,
-				postUpgradeMachineList,
-				client.InNamespace(workloadCluster.Namespace),
-				client.MatchingLabels{clusterv1.ClusterNameLabel: workloadCluster.Name},
+				preUpgradeMachineList,
+				client.InNamespace(workloadClusterNamespace),
+				client.MatchingLabels{clusterv1.ClusterNameLabel: workloadClusterName},
 			)
 			Expect(err).ToNot(HaveOccurred())
-			return validateMachineRollout(preUpgradeMachineList, postUpgradeMachineList)
-		}, "3m", "30s").Should(BeTrue(), "Machines should remain the same after the upgrade")
-
-		if workloadCluster.Spec.Topology != nil {
-			// Cluster is using ClusterClass, scale up via topology.
-			framework.ScaleAndWaitMachineDeploymentTopology(ctx, framework.ScaleAndWaitMachineDeploymentTopologyInput{
-				ClusterProxy:              managementClusterProxy,
-				Cluster:                   workloadCluster,
-				Replicas:                  2,
-				WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
-			})
-		} else {
-			// Cluster is not using ClusterClass, scale up via MachineDeployment.
-			testMachineDeployments := framework.GetMachineDeploymentsByCluster(ctx, framework.GetMachineDeploymentsByClusterInput{
-				Lister:      managementClusterProxy.GetClient(),
-				ClusterName: workloadCluster.Name,
-				Namespace:   workloadCluster.Namespace,
-			})
-			framework.ScaleAndWaitMachineDeployment(ctx, framework.ScaleAndWaitMachineDeploymentInput{
-				ClusterProxy:              managementClusterProxy,
-				Cluster:                   workloadCluster,
-				MachineDeployment:         testMachineDeployments[0],
-				Replicas:                  2,
-				WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
+
+			clusterctlUpgradeBinaryPath := ""
+			clusterctlUpgradeConfigPath := input.ClusterctlConfigPath
+			if upgrade.WithBinary != "" {
+				// Download the clusterctl version to be used to upgrade the management cluster
+				upgradeClusterctlBinaryURL := strings.NewReplacer("{OS}", runtime.GOOS, "{ARCH}", runtime.GOARCH).Replace(upgrade.WithBinary)
+				Byf("[%d] Downloading clusterctl binary from %s", i, upgradeClusterctlBinaryURL)
+				clusterctlUpgradeBinaryPath, clusterctlUpgradeConfigPath = setupClusterctl(ctx, upgradeClusterctlBinaryURL, input.ClusterctlConfigPath)
+				defer os.Remove(clusterctlBinaryPath) //nolint:gocritic // Resource leakage is not a concern here.
+			}
+
+			// Check if the user want a custom upgrade
+			isCustomUpgrade := upgrade.CoreProvider != "" ||
+				len(upgrade.BootstrapProviders) > 0 ||
+				len(upgrade.ControlPlaneProviders) > 0 ||
+				len(upgrade.InfrastructureProviders) > 0 ||
+				len(upgrade.IPAMProviders) > 0 ||
+				len(upgrade.RuntimeExtensionProviders) > 0 ||
+				len(upgrade.AddonProviders) > 0
+
+			if isCustomUpgrade {
+				Byf("[%d] Upgrading providers to custom versions", i)
+				clusterctl.UpgradeManagementClusterAndWait(ctx, clusterctl.UpgradeManagementClusterAndWaitInput{
+					ClusterctlBinaryPath:      clusterctlUpgradeBinaryPath, // use specific version of clusterctl to upgrade the management cluster (if set)
+					ClusterctlConfigPath:      clusterctlUpgradeConfigPath,
+					ClusterctlVariables:       input.UpgradeClusterctlVariables,
+					ClusterProxy:              managementClusterProxy,
+					CoreProvider:              upgrade.CoreProvider,
+					BootstrapProviders:        upgrade.BootstrapProviders,
+					ControlPlaneProviders:     upgrade.ControlPlaneProviders,
+					InfrastructureProviders:   upgrade.InfrastructureProviders,
+					IPAMProviders:             upgrade.IPAMProviders,
+					RuntimeExtensionProviders: upgrade.RuntimeExtensionProviders,
+					AddonProviders:            upgrade.AddonProviders,
+					LogFolder:                 filepath.Join(input.ArtifactFolder, "clusters", cluster.Name),
+				}, input.E2EConfig.GetIntervals(specName, "wait-controllers")...)
+			} else {
+				Byf("[%d] Upgrading providers to the latest version available", i)
+				clusterctl.UpgradeManagementClusterAndWait(ctx, clusterctl.UpgradeManagementClusterAndWaitInput{
+					ClusterctlBinaryPath: clusterctlUpgradeBinaryPath, // use specific version of clusterctl to upgrade the management cluster (if set)
+					ClusterctlConfigPath: clusterctlUpgradeConfigPath,
+					ClusterctlVariables:  input.UpgradeClusterctlVariables,
+					ClusterProxy:         managementClusterProxy,
+					Contract:             upgrade.Contract,
+					LogFolder:            filepath.Join(input.ArtifactFolder, "clusters", cluster.Name),
+				}, input.E2EConfig.GetIntervals(specName, "wait-controllers")...)
+			}
+
+			Byf("[%d] THE MANAGEMENT CLUSTER WAS SUCCESSFULLY UPGRADED!", i)
+
+			// We have to get the core CAPI storage version again as the upgrade might have stopped serving v1alpha3/v1alpha4.
+			coreCAPIStorageVersion = getCoreCAPIStorageVersion(ctx, managementClusterProxy.GetClient())
+
+			// Note: Currently we only support v1beta1 core CAPI apiVersion after upgrades.
+			// This seems a reasonable simplification as we don't want to test upgrades to v1alpha3 / v1alpha4.
+			workloadCluster := framework.DiscoveryAndWaitForCluster(ctx, framework.DiscoveryAndWaitForClusterInput{
+				Getter:    managementClusterProxy.GetClient(),
+				Namespace: workloadClusterNamespace,
+				Name:      workloadClusterName,
+			}, input.E2EConfig.GetIntervals(specName, "wait-cluster")...)
+
+			if input.PostUpgrade != nil {
+				Byf("[%d] Running Post-upgrade steps against the management cluster", i)
+				input.PostUpgrade(managementClusterProxy, workloadClusterNamespace, managementClusterName)
+			}
+
+			// After the upgrade check that there were no unexpected rollouts.
+			Byf("[%d] Verify there are no unexpected rollouts", i)
+			Consistently(func() bool {
+				postUpgradeMachineList := &unstructured.UnstructuredList{}
+				postUpgradeMachineList.SetGroupVersionKind(schema.GroupVersionKind{
+					Group:   clusterv1.GroupVersion.Group,
+					Version: coreCAPIStorageVersion,
+					Kind:    "MachineList",
+				})
+				err = managementClusterProxy.GetClient().List(
+					ctx,
+					postUpgradeMachineList,
+					client.InNamespace(workloadCluster.GetNamespace()),
+					client.MatchingLabels{clusterv1.ClusterNameLabel: workloadCluster.GetName()},
+				)
+				Expect(err).ToNot(HaveOccurred())
+				return validateMachineRollout(preUpgradeMachineList, postUpgradeMachineList)
+			}, "3m", "30s").Should(BeTrue(), "Machines should remain the same after the upgrade")
+
+			// Scale up to 2 and back down to 1 so we can repeat this multiple times.
+			Byf("[%d] Scale MachineDeployment to ensure the providers work", i)
+			if workloadCluster.Spec.Topology != nil {
+				// Cluster is using ClusterClass, scale up via topology.
+				framework.ScaleAndWaitMachineDeploymentTopology(ctx, framework.ScaleAndWaitMachineDeploymentTopologyInput{
+					ClusterProxy:              managementClusterProxy,
+					Cluster:                   workloadCluster,
+					Replicas:                  2,
+					WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
+				})
+				framework.ScaleAndWaitMachineDeploymentTopology(ctx, framework.ScaleAndWaitMachineDeploymentTopologyInput{
+					ClusterProxy:              managementClusterProxy,
+					Cluster:                   workloadCluster,
+					Replicas:                  1,
+					WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
+				})
+			} else {
+				// Cluster is not using ClusterClass, scale up via MachineDeployment.
+				testMachineDeployments := framework.GetMachineDeploymentsByCluster(ctx, framework.GetMachineDeploymentsByClusterInput{
+					Lister:      managementClusterProxy.GetClient(),
+					ClusterName: workloadClusterName,
+					Namespace:   workloadClusterNamespace,
+				})
+				framework.ScaleAndWaitMachineDeployment(ctx, framework.ScaleAndWaitMachineDeploymentInput{
+					ClusterProxy:              managementClusterProxy,
+					Cluster:                   workloadCluster,
+					MachineDeployment:         testMachineDeployments[0],
+					Replicas:                  2,
+					WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
+				})
+				framework.ScaleAndWaitMachineDeployment(ctx, framework.ScaleAndWaitMachineDeploymentInput{
+					ClusterProxy:              managementClusterProxy,
+					Cluster:                   workloadCluster,
+					MachineDeployment:         testMachineDeployments[0],
+					Replicas:                  1,
+					WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
+				})
+			}
+
+			Byf("[%d] Verify client-side SSA still works", i)
+			clusterUpdate := &unstructured.Unstructured{}
+			clusterUpdate.SetGroupVersionKind(clusterv1.GroupVersion.WithKind("Cluster"))
+			clusterUpdate.SetNamespace(workloadCluster.Namespace)
+			clusterUpdate.SetName(workloadCluster.Name)
+			clusterUpdate.SetLabels(map[string]string{
+				fmt.Sprintf("test-label-upgrade-%d", i): "test-label-value",
 			})
-		}
+			err = managementClusterProxy.GetClient().Patch(ctx, clusterUpdate, client.Apply, client.FieldOwner("e2e-test-client"))
+			Expect(err).ToNot(HaveOccurred())
 
-		By("THE UPGRADED MANAGEMENT CLUSTER WORKS!")
+			Byf("[%d] THE UPGRADED MANAGEMENT CLUSTER WORKS!", i)
+		}
 
 		By("PASSED!")
 	})
@@ -552,19 +626,8 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 			dumpAllResources(ctx, managementClusterProxy, input.ArtifactFolder, testNamespace, managementClusterResources.Cluster)
 
 			if !input.SkipCleanup {
-				switch {
-				case discovery.ServerSupportsVersion(managementClusterProxy.GetClientSet().DiscoveryClient, clusterv1.GroupVersion) == nil:
-					Byf("Deleting all %s clusters in namespace %s in management cluster %s", clusterv1.GroupVersion, testNamespace.Name, managementClusterName)
-					framework.DeleteAllClustersAndWait(ctx, framework.DeleteAllClustersAndWaitInput{
-						Client:    managementClusterProxy.GetClient(),
-						Namespace: testNamespace.Name,
-					}, input.E2EConfig.GetIntervals(specName, "wait-delete-cluster")...)
-				default:
-					log.Logf("Management Cluster does not appear to support CAPI resources.")
-				}
-
-				Byf("Deleting cluster %s", klog.KRef(testNamespace.Name, managementClusterName))
-				framework.DeleteAllClustersAndWait(ctx, framework.DeleteAllClustersAndWaitInput{
+				Byf("Deleting all clusters in namespace %s in management cluster %s", testNamespace.Name, managementClusterName)
+				deleteAllClustersAndWait(ctx, deleteAllClustersAndWaitInput{
 					Client:    managementClusterProxy.GetClient(),
 					Namespace: testNamespace.Name,
 				}, input.E2EConfig.GetIntervals(specName, "wait-delete-cluster")...)
@@ -594,6 +657,18 @@ func ClusterctlUpgradeSpec(ctx context.Context, inputGetter func() ClusterctlUpg
 	})
 }
 
+func setupClusterctl(ctx context.Context, clusterctlBinaryURL, clusterctlConfigPath string) (string, string) {
+	clusterctlBinaryPath := downloadToTmpFile(ctx, clusterctlBinaryURL)
+
+	err := os.Chmod(clusterctlBinaryPath, 0744) //nolint:gosec
+	Expect(err).ToNot(HaveOccurred(), "failed to chmod temporary file")
+
+	// Adjusts the clusterctlConfigPath in case the clusterctl version <= v1.3 (thus using a config file with only the providers supported in those versions)
+	clusterctlConfigPath = clusterctl.AdjustConfigPathForBinary(clusterctlBinaryPath, clusterctlConfigPath)
+
+	return clusterctlBinaryPath, clusterctlConfigPath
+}
+
 func downloadToTmpFile(ctx context.Context, url string) string {
 	tmpFile, err := os.CreateTemp("", "clusterctl")
 	Expect(err).ToNot(HaveOccurred(), "failed to get temporary file")
@@ -614,52 +689,186 @@ func downloadToTmpFile(ctx context.Context, url string) string {
 	return tmpFile.Name()
 }
 
-func calculateExpectedWorkerCount(ctx context.Context, c client.Client, e2eConfig *clusterctl.E2EConfig, specName string, cluster *clusterv1.Cluster) int64 {
+func getCoreCAPIStorageVersion(ctx context.Context, c client.Client) string {
+	clusterCRD := &apiextensionsv1.CustomResourceDefinition{}
+	if err := c.Get(ctx, client.ObjectKey{Name: "clusters.cluster.x-k8s.io"}, clusterCRD); err != nil {
+		Expect(err).ToNot(HaveOccurred(), "failed to retrieve a machine CRD")
+	}
+	// Pick the storage version
+	for _, version := range clusterCRD.Spec.Versions {
+		if version.Storage {
+			return version.Name
+		}
+	}
+	Fail("Cluster CRD has no storage version")
+	return ""
+}
+
+// discoveryAndWaitForClusterInput is the input type for DiscoveryAndWaitForCluster.
+type discoveryAndWaitForClusterInput struct {
+	Client                 client.Client
+	CoreCAPIStorageVersion string
+	Namespace              string
+	Name                   string
+}
+
+// discoveryAndWaitForCluster discovers a cluster object in a namespace and waits for the cluster infrastructure to be provisioned.
+func discoveryAndWaitForCluster(ctx context.Context, input discoveryAndWaitForClusterInput, intervals ...interface{}) *unstructured.Unstructured {
+	Expect(ctx).NotTo(BeNil(), "ctx is required for discoveryAndWaitForCluster")
+	Expect(input.Client).ToNot(BeNil(), "Invalid argument. input.Client can't be nil when calling discoveryAndWaitForCluster")
+	Expect(input.Namespace).ToNot(BeNil(), "Invalid argument. input.Namespace can't be empty when calling discoveryAndWaitForCluster")
+	Expect(input.Name).ToNot(BeNil(), "Invalid argument. input.Name can't be empty when calling discoveryAndWaitForCluster")
+
+	cluster := &unstructured.Unstructured{}
+	cluster.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   clusterv1.GroupVersion.Group,
+		Version: input.CoreCAPIStorageVersion,
+		Kind:    "Cluster",
+	})
+	Eventually(func(g Gomega) {
+		key := client.ObjectKey{
+			Namespace: input.Namespace,
+			Name:      input.Name,
+		}
+		g.Expect(input.Client.Get(ctx, key, cluster)).To(Succeed())
+
+		clusterPhase, ok, err := unstructured.NestedString(cluster.Object, "status", "phase")
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(ok).To(BeTrue(), "could not get status.phase field")
+		g.Expect(clusterPhase).To(Equal(string(clusterv1.ClusterPhaseProvisioned)), "Timed out waiting for Cluster %s to provision")
+	}, intervals...).Should(Succeed(), "Failed to get Cluster object %s", klog.KRef(input.Namespace, input.Name))
+
+	return cluster
+}
+
+func calculateExpectedWorkerCount(ctx context.Context, c client.Client, unstructuredCluster *unstructured.Unstructured, coreCAPIStorageVersion string) int64 {
 	var expectedWorkerCount int64
 
-	if cluster.Spec.Topology != nil {
-		if cluster.Spec.Topology.Workers != nil {
-			for _, md := range cluster.Spec.Topology.Workers.MachineDeployments {
-				if md.Replicas == nil {
-					continue
+	// Convert v1beta1 unstructured Cluster to clusterv1.Cluster
+	// Only v1beta1 Cluster support ClusterClass (i.e. have cluster.spec.topology).
+	if unstructuredCluster.GroupVersionKind().Version == clusterv1.GroupVersion.Version {
+		cluster := &clusterv1.Cluster{}
+		Expect(apiruntime.DefaultUnstructuredConverter.FromUnstructured(unstructuredCluster.Object, cluster)).To(Succeed())
+
+		if cluster.Spec.Topology != nil {
+			if cluster.Spec.Topology.Workers != nil {
+				for _, md := range cluster.Spec.Topology.Workers.MachineDeployments {
+					if md.Replicas == nil {
+						continue
+					}
+					expectedWorkerCount += int64(*md.Replicas)
 				}
-				expectedWorkerCount += int64(*md.Replicas)
-			}
-			for _, mp := range cluster.Spec.Topology.Workers.MachinePools {
-				if mp.Replicas == nil {
-					continue
+				for _, mp := range cluster.Spec.Topology.Workers.MachinePools {
+					if mp.Replicas == nil {
+						continue
+					}
+					expectedWorkerCount += int64(*mp.Replicas)
 				}
-				expectedWorkerCount += int64(*mp.Replicas)
 			}
+			return expectedWorkerCount
 		}
-		return expectedWorkerCount
 	}
 
-	machineDeployments := framework.DiscoveryAndWaitForMachineDeployments(ctx, framework.DiscoveryAndWaitForMachineDeploymentsInput{
-		Lister:  c,
-		Cluster: cluster,
-	}, e2eConfig.GetIntervals(specName, "wait-worker-nodes")...)
-	for _, md := range machineDeployments {
-		if md.Spec.Replicas == nil {
+	byClusterOptions := []client.ListOption{
+		client.InNamespace(unstructuredCluster.GetNamespace()),
+		client.MatchingLabels{clusterv1.ClusterNameLabel: unstructuredCluster.GetName()},
+	}
+
+	machineDeploymentList := &unstructured.UnstructuredList{}
+	machineDeploymentList.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   clusterv1.GroupVersion.Group,
+		Version: coreCAPIStorageVersion,
+		Kind:    "MachineDeploymentList",
+	})
+	Eventually(func() error {
+		return c.List(ctx, machineDeploymentList, byClusterOptions...)
+	}, 3*time.Minute, 3*time.Second).Should(Succeed(), "Failed to list MachineDeployments object for Cluster %s", klog.KObj(unstructuredCluster))
+	for _, md := range machineDeploymentList.Items {
+		replicas, ok, err := unstructured.NestedInt64(md.Object, "spec", "replicas")
+		Expect(err).ToNot(HaveOccurred())
+		if !ok {
 			continue
 		}
-		expectedWorkerCount += int64(*md.Spec.Replicas)
+		expectedWorkerCount += replicas
 	}
 
-	machinePools := framework.DiscoveryAndWaitForMachinePools(ctx, framework.DiscoveryAndWaitForMachinePoolsInput{
-		Getter:  c,
-		Lister:  c,
-		Cluster: cluster,
-	}, e2eConfig.GetIntervals(specName, "wait-machine-pool-nodes")...)
-	for _, mp := range machinePools {
-		if mp.Spec.Replicas == nil {
+	machinePoolList := &unstructured.UnstructuredList{}
+	machinePoolList.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   clusterv1.GroupVersion.Group,
+		Version: coreCAPIStorageVersion,
+		Kind:    "MachinePoolList",
+	})
+	Eventually(func() error {
+		return c.List(ctx, machinePoolList, byClusterOptions...)
+	}, 3*time.Minute, 3*time.Second).Should(Succeed(), "Failed to list MachinePool object for Cluster %s", klog.KObj(unstructuredCluster))
+	for _, md := range machinePoolList.Items {
+		replicas, ok, err := unstructured.NestedInt64(md.Object, "spec", "replicas")
+		Expect(err).ToNot(HaveOccurred())
+		if !ok {
 			continue
 		}
-		expectedWorkerCount += int64(*mp.Spec.Replicas)
+		expectedWorkerCount += replicas
 	}
+
 	return expectedWorkerCount
 }
 
+// deleteAllClustersAndWaitInput is the input type for deleteAllClustersAndWait.
+type deleteAllClustersAndWaitInput struct {
+	Client    client.Client
+	Namespace string
+}
+
+// deleteAllClustersAndWait deletes all cluster resources in the given namespace and waits for them to be gone.
+func deleteAllClustersAndWait(ctx context.Context, input deleteAllClustersAndWaitInput, intervals ...interface{}) {
+	Expect(ctx).NotTo(BeNil(), "ctx is required for deleteAllClustersAndWaitOldAPI")
+	Expect(input.Client).ToNot(BeNil(), "Invalid argument. input.Client can't be nil when calling deleteAllClustersAndWaitOldAPI")
+	Expect(input.Namespace).ToNot(BeEmpty(), "Invalid argument. input.Namespace can't be empty when calling deleteAllClustersAndWaitOldAPI")
+
+	coreCAPIStorageVersion := getCoreCAPIStorageVersion(ctx, input.Client)
+
+	clusterList := &unstructured.UnstructuredList{}
+	clusterList.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   clusterv1.GroupVersion.Group,
+		Version: coreCAPIStorageVersion,
+		Kind:    "ClusterList",
+	})
+	Expect(input.Client.List(ctx, clusterList, client.InNamespace(input.Namespace))).To(Succeed(), "Failed to list clusters in namespace %s", input.Namespace)
+
+	// Enforce restart of kube-controller-manager by stealing its lease.
+	// Note: Due to a known issue in the kube-controller-manager we have to restart it
+	// in case the kube-controller-manager internally caches ownerRefs of apiVersions
+	// which now don't exist anymore (e.g. v1alpha3/v1alpha4).
+	// Alternatives to this would be:
+	// * some other way to restart the kube-controller-manager (e.g. control plane node rollout)
+	// * removing ownerRefs from (at least) MachineDeployments
+	Eventually(func(g Gomega) {
+		kubeControllerManagerLease := &coordinationv1.Lease{}
+		g.Expect(input.Client.Get(ctx, client.ObjectKey{Namespace: metav1.NamespaceSystem, Name: "kube-controller-manager"}, kubeControllerManagerLease)).To(Succeed())
+		// As soon as the kube-controller-manager detects it doesn't own the lease anymore it will restart.
+		// Once the current lease times out the kube-controller-manager will become leader again.
+		kubeControllerManagerLease.Spec.HolderIdentity = ptr.To("e2e-test-client")
+		g.Expect(input.Client.Update(ctx, kubeControllerManagerLease)).To(Succeed())
+	}, 3*time.Minute, 3*time.Second).Should(Succeed(), "failed to steal lease from kube-controller-manager to trigger restart")
+
+	for _, c := range clusterList.Items {
+		Byf("Deleting cluster %s", c.GetName())
+		Expect(input.Client.Delete(ctx, c.DeepCopy())).To(Succeed())
+	}
+
+	for _, c := range clusterList.Items {
+		Byf("Waiting for cluster %s to be deleted", c.GetName())
+		Eventually(func() bool {
+			cluster := c.DeepCopy()
+			key := client.ObjectKey{
+				Namespace: c.GetNamespace(),
+				Name:      c.GetName(),
+			}
+			return apierrors.IsNotFound(input.Client.Get(ctx, key, cluster))
+		}, intervals...).Should(BeTrue())
+	}
+}
+
 // validateMachineRollout compares preMachineList and postMachineList to detect a rollout.
 // Note: we are using unstructured lists because the Machines have different apiVersions.
 func validateMachineRollout(preMachineList, postMachineList *unstructured.UnstructuredList) bool {
diff --git a/test/e2e/clusterctl_upgrade_test.go b/test/e2e/clusterctl_upgrade_test.go
index 628eee2c2368..fdf65a7462bd 100644
--- a/test/e2e/clusterctl_upgrade_test.go
+++ b/test/e2e/clusterctl_upgrade_test.go
@@ -25,6 +25,8 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"k8s.io/utils/ptr"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 )
 
 var (
@@ -34,6 +36,60 @@ var (
 	providerDockerPrefix  = "docker:v%s"
 )
 
+var _ = Describe("When testing clusterctl upgrades (v0.4=>v1.6=>current)", func() {
+	// Get v0.4 latest stable release
+	version04 := "0.4"
+	stableRelease04, err := GetStableReleaseOfMinor(ctx, version04)
+	Expect(err).ToNot(HaveOccurred(), "Failed to get stable version for minor release : %s", version04)
+
+	// Get v1.6 latest stable release
+	version16 := "1.6"
+	stableRelease16, err := GetStableReleaseOfMinor(ctx, version16)
+	Expect(err).ToNot(HaveOccurred(), "Failed to get stable version for minor release : %s", version16)
+
+	ClusterctlUpgradeSpec(ctx, func() ClusterctlUpgradeSpecInput {
+		return ClusterctlUpgradeSpecInput{
+			E2EConfig:              e2eConfig,
+			ClusterctlConfigPath:   clusterctlConfigPath,
+			BootstrapClusterProxy:  bootstrapClusterProxy,
+			ArtifactFolder:         artifactFolder,
+			SkipCleanup:            skipCleanup,
+			InfrastructureProvider: ptr.To("docker"),
+			// Configuration for the initial provider deployment.
+			InitWithBinary: fmt.Sprintf(clusterctlDownloadURL, stableRelease04),
+			// We have to pin the providers because with `InitWithProvidersContract` the test would
+			// use the latest version for the contract (which is v1.3.X for v1beta1).
+			InitWithCoreProvider:            fmt.Sprintf(providerCAPIPrefix, stableRelease04),
+			InitWithBootstrapProviders:      []string{fmt.Sprintf(providerKubeadmPrefix, stableRelease04)},
+			InitWithControlPlaneProviders:   []string{fmt.Sprintf(providerKubeadmPrefix, stableRelease04)},
+			InitWithInfrastructureProviders: []string{fmt.Sprintf(providerDockerPrefix, stableRelease04)},
+			// We have to set this to an empty array as clusterctl v0.4 doesn't support
+			// runtime extension providers. If we don't do this the test will automatically
+			// try to deploy the latest version of our test-extension from docker.yaml.
+			InitWithRuntimeExtensionProviders: []string{},
+			// Configuration for the provider upgrades.
+			Upgrades: []ClusterctlUpgradeSpecInputUpgrade{
+				{ // Upgrade to 1.6.
+					WithBinary:              fmt.Sprintf(clusterctlDownloadURL, stableRelease16),
+					CoreProvider:            fmt.Sprintf(providerCAPIPrefix, stableRelease16),
+					BootstrapProviders:      []string{fmt.Sprintf(providerKubeadmPrefix, stableRelease16)},
+					ControlPlaneProviders:   []string{fmt.Sprintf(providerKubeadmPrefix, stableRelease16)},
+					InfrastructureProviders: []string{fmt.Sprintf(providerDockerPrefix, stableRelease16)},
+				},
+				{ // Upgrade to latest v1beta1.
+					Contract: clusterv1.GroupVersion.Version,
+				},
+			},
+			// NOTE: If this version is changed here the image and SHA must also be updated in all DockerMachineTemplates in `test/data/infrastructure-docker/v0.4/bases.
+			//  Note: Both InitWithKubernetesVersion and WorkloadKubernetesVersion should be the highest mgmt cluster version supported by the source Cluster API version.
+			InitWithKubernetesVersion: "v1.23.17",
+			WorkloadKubernetesVersion: "v1.23.17",
+			MgmtFlavor:                "topology",
+			WorkloadFlavor:            "",
+		}
+	})
+})
+
 var _ = Describe("When testing clusterctl upgrades (v1.0=>current)", func() {
 	// Get v1.0 latest stable release
 	version := "1.0"
diff --git a/test/e2e/config/docker.yaml b/test/e2e/config/docker.yaml
index fa4a3d5078cb..1e6e433ffd7c 100644
--- a/test/e2e/config/docker.yaml
+++ b/test/e2e/config/docker.yaml
@@ -35,6 +35,15 @@ providers:
 - name: cluster-api
   type: CoreProvider
   versions:
+  - name: "{go://sigs.k8s.io/cluster-api@v0.4}" # latest published release in the v1alpha4 series; this is used for v1alpha4 --> v1beta1 clusterctl upgrades test only.
+    value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v0.4}/core-components.yaml"
+    type: "url"
+    contract: v1alpha4
+    replacements:
+    - old: --metrics-addr=127.0.0.1:8080
+      new: --metrics-addr=:8080
+    files:
+    - sourcePath: "../data/shared/v0.4/metadata.yaml"
   - name: "{go://sigs.k8s.io/cluster-api@v1.0}" # supported release in the v1beta1 series; this is used for v1beta1 --> main clusterctl upgrades test only.
     value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v1.0}/core-components.yaml"
     type: "url"
@@ -73,6 +82,15 @@ providers:
 - name: kubeadm
   type: BootstrapProvider
   versions:
+  - name: "{go://sigs.k8s.io/cluster-api@v0.4}" # latest published release in the v1alpha4 series; this is used for v1alpha4 --> v1beta1 clusterctl upgrades test only.
+    value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v0.4}/bootstrap-components.yaml"
+    type: "url"
+    contract: v1alpha4
+    replacements:
+    - old: --metrics-addr=127.0.0.1:8080
+      new: --metrics-addr=:8080
+    files:
+    - sourcePath: "../data/shared/v0.4/metadata.yaml"
   - name: "{go://sigs.k8s.io/cluster-api@v1.0}" # supported release in the v1beta1 series; this is used for v1beta1 --> main clusterctl upgrades test only.
     value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v1.0}/bootstrap-components.yaml"
     type: "url"
@@ -111,6 +129,15 @@ providers:
 - name: kubeadm
   type: ControlPlaneProvider
   versions:
+  - name: "{go://sigs.k8s.io/cluster-api@v0.4}" # latest published release in the v1alpha4 series; this is used for v1alpha4 --> v1beta1 clusterctl upgrades test only.
+    value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v0.4}/control-plane-components.yaml"
+    type: "url"
+    contract: v1alpha4
+    replacements:
+    - old: --metrics-addr=127.0.0.1:8080
+      new: --metrics-addr=:8080
+    files:
+    - sourcePath: "../data/shared/v0.4/metadata.yaml"
   - name: "{go://sigs.k8s.io/cluster-api@v1.0}" # supported release in the v1beta1 series; this is used for v1beta1 --> main clusterctl upgrades test only.
     value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v1.0}/control-plane-components.yaml"
     type: "url"
@@ -149,6 +176,16 @@ providers:
 - name: docker
   type: InfrastructureProvider
   versions:
+  - name: "{go://sigs.k8s.io/cluster-api@v0.4}" # latest published release in the v1alpha4 series; this is used for v1alpha4 --> v1beta1 clusterctl upgrades test only.
+    value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v0.4}/infrastructure-components-development.yaml"
+    type: "url"
+    contract: v1alpha4
+    replacements:
+    - old: --metrics-addr=127.0.0.1:8080
+      new: --metrics-addr=:8080
+    files:
+    - sourcePath: "../data/shared/v0.4/metadata.yaml"
+    - sourcePath: "../data/infrastructure-docker/v0.4/cluster-template.yaml"
   - name: "{go://sigs.k8s.io/cluster-api@v1.0}" # supported release in the v1beta1 series; this is used for v1beta1 --> main clusterctl upgrades test only.
     value: "https://github.com/kubernetes-sigs/cluster-api/releases/download/{go://sigs.k8s.io/cluster-api@v1.0}/infrastructure-components-development.yaml"
     type: "url"
diff --git a/test/e2e/data/infrastructure-docker/v0.4/bases/cluster-with-kcp.yaml b/test/e2e/data/infrastructure-docker/v0.4/bases/cluster-with-kcp.yaml
new file mode 100644
index 000000000000..371789cf5745
--- /dev/null
+++ b/test/e2e/data/infrastructure-docker/v0.4/bases/cluster-with-kcp.yaml
@@ -0,0 +1,87 @@
+---
+# DockerCluster object referenced by the Cluster object
+apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+kind: DockerCluster
+metadata:
+  name: '${CLUSTER_NAME}'
+---
+# Cluster object with
+# - Reference to the KubeadmControlPlane object
+# - the label cni=${CLUSTER_NAME}-crs-0, so the cluster can be selected by the ClusterResourceSet.
+apiVersion: cluster.x-k8s.io/v1alpha4
+kind: Cluster
+metadata:
+  name: '${CLUSTER_NAME}'
+  labels:
+    cni: "${CLUSTER_NAME}-crs-0"
+spec:
+  clusterNetwork:
+    services:
+      cidrBlocks: ['${DOCKER_SERVICE_CIDRS}']
+    pods:
+      cidrBlocks: ['${DOCKER_POD_CIDRS}']
+    serviceDomain: '${DOCKER_SERVICE_DOMAIN}'
+  infrastructureRef:
+    apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+    kind: DockerCluster
+    name: '${CLUSTER_NAME}'
+  controlPlaneRef:
+    kind: KubeadmControlPlane
+    apiVersion: controlplane.cluster.x-k8s.io/v1alpha4
+    name: "${CLUSTER_NAME}-control-plane"
+---
+# DockerMachineTemplate object referenced by the KubeadmControlPlane object
+apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+kind: DockerMachineTemplate
+metadata:
+  name: "${CLUSTER_NAME}-control-plane"
+spec:
+  template:
+    spec:
+      # NOTE: If the Kubernetes version is changed in `clusterctl_upgrade_test.go` the image and SHA must be updated here.
+      customImage: "kindest/node:v1.23.17@sha256:f77f8cf0b30430ca4128cc7cfafece0c274a118cd0cdb251049664ace0dee4ff"
+      extraMounts:
+        - containerPath: "/var/run/docker.sock"
+          hostPath: "/var/run/docker.sock"
+---
+# KubeadmControlPlane referenced by the Cluster object with
+# - the label kcp-adoption.step2, because it should be created in the second step of the kcp-adoption test.
+kind: KubeadmControlPlane
+apiVersion: controlplane.cluster.x-k8s.io/v1alpha4
+metadata:
+  name: "${CLUSTER_NAME}-control-plane"
+  labels:
+    kcp-adoption.step2: ""
+spec:
+  replicas: ${CONTROL_PLANE_MACHINE_COUNT}
+  machineTemplate:
+    infrastructureRef:
+      kind: DockerMachineTemplate
+      apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+      name: "${CLUSTER_NAME}-control-plane"
+  kubeadmConfigSpec:
+    clusterConfiguration:
+      controllerManager:
+        extraArgs: {enable-hostpath-provisioner: 'true'}
+      apiServer:
+        # host.docker.internal is required by kubetest when running on MacOS because of the way ports are proxied.
+        certSANs: [localhost, 127.0.0.1, 0.0.0.0, host.docker.internal]
+    initConfiguration:
+      nodeRegistration:
+        criSocket: unix:///var/run/containerd/containerd.sock
+        kubeletExtraArgs:
+          # We have to pin the cgroupDriver to cgroupfs for Kubernetes < v1.24 because kind does not support systemd for those versions, but kubeadm >= 1.21 defaults to systemd.
+          # This cluster is used in tests where the Kubernetes version is < 1.24
+          cgroup-driver: cgroupfs
+          eviction-hard: 'nodefs.available<0%,nodefs.inodesFree<0%,imagefs.available<0%'
+          fail-swap-on: "false"
+    joinConfiguration:
+      nodeRegistration:
+        criSocket: unix:///var/run/containerd/containerd.sock
+        kubeletExtraArgs:
+          # We have to pin the cgroupDriver to cgroupfs for Kubernetes < v1.24 because kind does not support systemd for those versions, but kubeadm >= 1.21 defaults to systemd.
+          # This cluster is used in tests where the Kubernetes version is < 1.24
+          cgroup-driver: cgroupfs
+          eviction-hard: 'nodefs.available<0%,nodefs.inodesFree<0%,imagefs.available<0%'
+          fail-swap-on: "false"
+  version: "${KUBERNETES_VERSION}"
diff --git a/test/e2e/data/infrastructure-docker/v0.4/bases/crs.yaml b/test/e2e/data/infrastructure-docker/v0.4/bases/crs.yaml
new file mode 100644
index 000000000000..7f8f9f9d46e1
--- /dev/null
+++ b/test/e2e/data/infrastructure-docker/v0.4/bases/crs.yaml
@@ -0,0 +1,24 @@
+---
+# ConfigMap object referenced by the ClusterResourceSet object and with
+# the CNI resource defined in the test config file
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: "cni-${CLUSTER_NAME}-crs-0"
+data: ${CNI_RESOURCES}
+binaryData:
+---
+# ClusterResourceSet object with
+# a selector that targets all the Cluster with label cni=${CLUSTER_NAME}-crs-0
+apiVersion: addons.cluster.x-k8s.io/v1alpha4
+kind: ClusterResourceSet
+metadata:
+  name:  "${CLUSTER_NAME}-crs-0"
+spec:
+  strategy: ApplyOnce
+  clusterSelector:
+    matchLabels:
+      cni: "${CLUSTER_NAME}-crs-0"
+  resources:
+    - name: "cni-${CLUSTER_NAME}-crs-0"
+      kind: ConfigMap
diff --git a/test/e2e/data/infrastructure-docker/v0.4/bases/md.yaml b/test/e2e/data/infrastructure-docker/v0.4/bases/md.yaml
new file mode 100644
index 000000000000..e7c19d3c4497
--- /dev/null
+++ b/test/e2e/data/infrastructure-docker/v0.4/bases/md.yaml
@@ -0,0 +1,57 @@
+---
+# DockerMachineTemplate referenced by the MachineDeployment and with
+# - extraMounts for the docker sock, thus allowing self-hosting test
+apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+kind: DockerMachineTemplate
+metadata:
+  name: "${CLUSTER_NAME}-md-0"
+spec:
+  template:
+    spec:
+      # NOTE: If the Kubernetes version is changed in `clusterctl_upgrade_test.go` the image and SHA must be updated here.
+      customImage: "kindest/node:v1.23.17@sha256:f77f8cf0b30430ca4128cc7cfafece0c274a118cd0cdb251049664ace0dee4ff"
+      extraMounts:
+        - containerPath: "/var/run/docker.sock"
+          hostPath: "/var/run/docker.sock"
+---
+# KubeadmConfigTemplate referenced by the MachineDeployment
+apiVersion: bootstrap.cluster.x-k8s.io/v1alpha4
+kind: KubeadmConfigTemplate
+metadata:
+  name: "${CLUSTER_NAME}-md-0"
+spec:
+  template:
+    spec:
+      joinConfiguration:
+        nodeRegistration:
+          criSocket: unix:///var/run/containerd/containerd.sock
+          kubeletExtraArgs:
+            # We have to pin the cgroupDriver to cgroupfs for Kubernetes < v1.24 because kind does not support systemd for those versions, but kubeadm >= 1.21 defaults to systemd.
+            # This cluster is used in tests where the Kubernetes version is < 1.24
+            cgroup-driver: cgroupfs
+            eviction-hard: 'nodefs.available<0%,nodefs.inodesFree<0%,imagefs.available<0%'
+            fail-swap-on: "false"
+---
+# MachineDeployment object
+apiVersion: cluster.x-k8s.io/v1alpha4
+kind: MachineDeployment
+metadata:
+  name: "${CLUSTER_NAME}-md-0"
+spec:
+  clusterName: "${CLUSTER_NAME}"
+  replicas: ${WORKER_MACHINE_COUNT}
+  selector:
+    matchLabels:
+  template:
+    spec:
+      clusterName: "${CLUSTER_NAME}"
+      version: "${KUBERNETES_VERSION}"
+      bootstrap:
+        configRef:
+          name: "${CLUSTER_NAME}-md-0"
+          apiVersion: bootstrap.cluster.x-k8s.io/v1alpha4
+          kind: KubeadmConfigTemplate
+      infrastructureRef:
+        name: "${CLUSTER_NAME}-md-0"
+        apiVersion: infrastructure.cluster.x-k8s.io/v1alpha4
+        kind: DockerMachineTemplate
diff --git a/test/e2e/data/infrastructure-docker/v0.4/cluster-template/kustomization.yaml b/test/e2e/data/infrastructure-docker/v0.4/cluster-template/kustomization.yaml
new file mode 100644
index 000000000000..c7805717ecc1
--- /dev/null
+++ b/test/e2e/data/infrastructure-docker/v0.4/cluster-template/kustomization.yaml
@@ -0,0 +1,4 @@
+bases:
+- ../bases/cluster-with-kcp.yaml
+- ../bases/md.yaml
+- ../bases/crs.yaml
\ No newline at end of file
diff --git a/test/e2e/data/shared/v0.4/metadata.yaml b/test/e2e/data/shared/v0.4/metadata.yaml
new file mode 100644
index 000000000000..318ea96c6eda
--- /dev/null
+++ b/test/e2e/data/shared/v0.4/metadata.yaml
@@ -0,0 +1,12 @@
+apiVersion: clusterctl.cluster.x-k8s.io/v1alpha3
+kind: Metadata
+releaseSeries:
+  - major: 0
+    minor: 4
+    contract: v1alpha4
+  - major: 0
+    minor: 3
+    contract: v1alpha3
+  - major: 0
+    minor: 2
+    contract: v1alpha2
\ No newline at end of file
diff --git a/test/framework/clusterctl/client.go b/test/framework/clusterctl/client.go
index 13606b2d8564..f01f7c15bacf 100644
--- a/test/framework/clusterctl/client.go
+++ b/test/framework/clusterctl/client.go
@@ -109,7 +109,7 @@ func InitWithBinary(_ context.Context, binary string, input InitInput) {
 }
 
 func calculateClusterCtlInitArgs(input InitInput) []string {
-	args := []string{"init", "--config", input.ClusterctlConfigPath, "--kubeconfig", input.KubeconfigPath}
+	args := []string{"init", "--config", input.ClusterctlConfigPath, "--kubeconfig", input.KubeconfigPath, "--wait-providers"}
 	if input.CoreProvider != "" {
 		args = append(args, "--core", input.CoreProvider)
 	}
@@ -163,37 +163,8 @@ func Upgrade(ctx context.Context, input UpgradeInput) {
 		input.ClusterctlConfigPath = outputPath
 	}
 
-	// Check if the user want a custom upgrade
-	isCustomUpgrade := input.CoreProvider != "" ||
-		len(input.BootstrapProviders) > 0 ||
-		len(input.ControlPlaneProviders) > 0 ||
-		len(input.InfrastructureProviders) > 0 ||
-		len(input.IPAMProviders) > 0 ||
-		len(input.RuntimeExtensionProviders) > 0 ||
-		len(input.AddonProviders) > 0
-
-	Expect((input.Contract != "" && !isCustomUpgrade) || (input.Contract == "" && isCustomUpgrade)).To(BeTrue(), `Invalid arguments. Either the input.Contract parameter or at least one of the following providers has to be set:
-		input.CoreProvider, input.BootstrapProviders, input.ControlPlaneProviders, input.InfrastructureProviders, input.IPAMProviders, input.RuntimeExtensionProviders, input.AddonProviders`)
-
-	if isCustomUpgrade {
-		log.Logf("clusterctl upgrade apply --core %s --bootstrap %s --control-plane %s --infrastructure %s --ipam %s --runtime-extension %s --addon %s --config %s --kubeconfig %s",
-			input.CoreProvider,
-			strings.Join(input.BootstrapProviders, ","),
-			strings.Join(input.ControlPlaneProviders, ","),
-			strings.Join(input.InfrastructureProviders, ","),
-			strings.Join(input.IPAMProviders, ","),
-			strings.Join(input.RuntimeExtensionProviders, ","),
-			strings.Join(input.AddonProviders, ","),
-			input.ClusterctlConfigPath,
-			input.KubeconfigPath,
-		)
-	} else {
-		log.Logf("clusterctl upgrade apply --contract %s --config %s --kubeconfig %s",
-			input.Contract,
-			input.ClusterctlConfigPath,
-			input.KubeconfigPath,
-		)
-	}
+	args := calculateClusterCtlUpgradeArgs(input)
+	log.Logf("clusterctl %s", strings.Join(args, " "))
 
 	upgradeOpt := clusterctlclient.ApplyUpgradeOptions{
 		Kubeconfig: clusterctlclient.Kubeconfig{
@@ -218,6 +189,79 @@ func Upgrade(ctx context.Context, input UpgradeInput) {
 	Expect(err).ToNot(HaveOccurred(), "failed to run clusterctl upgrade")
 }
 
+// UpgradeWithBinary calls clusterctl upgrade apply with the list of providers defined in the local repository.
+func UpgradeWithBinary(ctx context.Context, binary string, input UpgradeInput) {
+	if len(input.ClusterctlVariables) > 0 {
+		outputPath := filepath.Join(filepath.Dir(input.ClusterctlConfigPath), fmt.Sprintf("clusterctl-upgrade-config-%s.yaml", input.ClusterName))
+		Expect(CopyAndAmendClusterctlConfig(ctx, CopyAndAmendClusterctlConfigInput{
+			ClusterctlConfigPath: input.ClusterctlConfigPath,
+			OutputPath:           outputPath,
+			Variables:            input.ClusterctlVariables,
+		})).To(Succeed(), "Failed to CopyAndAmendClusterctlConfig")
+		input.ClusterctlConfigPath = outputPath
+	}
+
+	args := calculateClusterCtlUpgradeArgs(input)
+	log.Logf("clusterctl %s", strings.Join(args, " "))
+
+	cmd := exec.Command(binary, args...) //nolint:gosec // We don't care about command injection here.
+
+	out, err := cmd.CombinedOutput()
+	_ = os.WriteFile(filepath.Join(input.LogFolder, "clusterctl-upgrade.log"), out, 0644) //nolint:gosec // this is a log file to be shared via prow artifacts
+	var stdErr string
+	if err != nil {
+		var exitErr *exec.ExitError
+		if errors.As(err, &exitErr) {
+			stdErr = string(exitErr.Stderr)
+		}
+	}
+	Expect(err).ToNot(HaveOccurred(), "failed to run clusterctl upgrade apply:\nstdout:\n%s\nstderr:\n%s", string(out), stdErr)
+}
+
+func calculateClusterCtlUpgradeArgs(input UpgradeInput) []string {
+	args := []string{"upgrade", "apply", "--config", input.ClusterctlConfigPath, "--kubeconfig", input.KubeconfigPath, "--wait-providers"}
+
+	// Check if the user want a custom upgrade
+	isCustomUpgrade := input.CoreProvider != "" ||
+		len(input.BootstrapProviders) > 0 ||
+		len(input.ControlPlaneProviders) > 0 ||
+		len(input.InfrastructureProviders) > 0 ||
+		len(input.IPAMProviders) > 0 ||
+		len(input.RuntimeExtensionProviders) > 0 ||
+		len(input.AddonProviders) > 0
+
+	Expect((input.Contract != "" && !isCustomUpgrade) || (input.Contract == "" && isCustomUpgrade)).To(BeTrue(), `Invalid arguments. Either the input.Contract parameter or at least one of the following providers has to be set:
+		input.CoreProvider, input.BootstrapProviders, input.ControlPlaneProviders, input.InfrastructureProviders, input.IPAMProviders, input.RuntimeExtensionProviders, input.AddonProviders`)
+
+	if isCustomUpgrade {
+		if input.CoreProvider != "" {
+			args = append(args, "--core", input.CoreProvider)
+		}
+		if len(input.BootstrapProviders) > 0 {
+			args = append(args, "--bootstrap", strings.Join(input.BootstrapProviders, ","))
+		}
+		if len(input.ControlPlaneProviders) > 0 {
+			args = append(args, "--control-plane", strings.Join(input.ControlPlaneProviders, ","))
+		}
+		if len(input.InfrastructureProviders) > 0 {
+			args = append(args, "--infrastructure", strings.Join(input.InfrastructureProviders, ","))
+		}
+		if len(input.IPAMProviders) > 0 {
+			args = append(args, "--ipam", strings.Join(input.IPAMProviders, ","))
+		}
+		if len(input.RuntimeExtensionProviders) > 0 {
+			args = append(args, "--runtime-extension", strings.Join(input.RuntimeExtensionProviders, ","))
+		}
+		if len(input.AddonProviders) > 0 {
+			args = append(args, "--addon", strings.Join(input.AddonProviders, ","))
+		}
+	} else {
+		args = append(args, "--contract", input.Contract)
+	}
+
+	return args
+}
+
 // DeleteInput is the input for Delete.
 type DeleteInput struct {
 	LogFolder            string
diff --git a/test/framework/clusterctl/clusterctl_helpers.go b/test/framework/clusterctl/clusterctl_helpers.go
index ec56339ad43a..508b657cd7f8 100644
--- a/test/framework/clusterctl/clusterctl_helpers.go
+++ b/test/framework/clusterctl/clusterctl_helpers.go
@@ -144,6 +144,7 @@ type UpgradeManagementClusterAndWaitInput struct {
 	RuntimeExtensionProviders []string
 	AddonProviders            []string
 	LogFolder                 string
+	ClusterctlBinaryPath      string
 }
 
 // UpgradeManagementClusterAndWait upgrades provider a management cluster using clusterctl, and waits for the cluster to be ready.
@@ -165,7 +166,7 @@ func UpgradeManagementClusterAndWait(ctx context.Context, input UpgradeManagemen
 
 	Expect(os.MkdirAll(input.LogFolder, 0750)).To(Succeed(), "Invalid argument. input.LogFolder can't be created for UpgradeManagementClusterAndWait")
 
-	Upgrade(ctx, UpgradeInput{
+	upgradeInput := UpgradeInput{
 		ClusterctlConfigPath:      input.ClusterctlConfigPath,
 		ClusterctlVariables:       input.ClusterctlVariables,
 		ClusterName:               input.ClusterProxy.GetName(),
@@ -179,7 +180,13 @@ func UpgradeManagementClusterAndWait(ctx context.Context, input UpgradeManagemen
 		RuntimeExtensionProviders: input.RuntimeExtensionProviders,
 		AddonProviders:            input.AddonProviders,
 		LogFolder:                 input.LogFolder,
-	})
+	}
+
+	if input.ClusterctlBinaryPath != "" {
+		UpgradeWithBinary(ctx, input.ClusterctlBinaryPath, upgradeInput)
+	} else {
+		Upgrade(ctx, upgradeInput)
+	}
 
 	client := input.ClusterProxy.GetClient()
 
diff --git a/test/framework/convenience.go b/test/framework/convenience.go
index 3e0f62608692..07fa4422641d 100644
--- a/test/framework/convenience.go
+++ b/test/framework/convenience.go
@@ -20,6 +20,7 @@ import (
 	"reflect"
 
 	appsv1 "k8s.io/api/apps/v1"
+	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -75,6 +76,10 @@ func TryAddDefaultSchemes(scheme *runtime.Scheme) {
 
 	// Add rbac to the scheme.
 	_ = rbacv1.AddToScheme(scheme)
+
+	// Add coordination to the schema
+	// Note: This is e.g. used to trigger kube-controller-manager restarts by stealing its lease.
+	_ = coordinationv1.AddToScheme(scheme)
 }
 
 // ObjectToKind returns the Kind without the package prefix. Pass in a pointer to a struct