Add functional tests to cover the import populator flow

This commit updates the import tests to cover the new import populator flow.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

pkg/controller/common/util.go

@@ -1356,7 +1356,7 @@ func GetMetricsURL(pod *corev1.Pod) (string, error) {
 	return url, nil
 }
 
-// GetProgressReport fetches the progress report from the passed URL according to an specific regular expression
+// GetProgressReportFromURL fetches the progress report from the passed URL according to an specific regular expression
 func GetProgressReportFromURL(url string, regExp *regexp.Regexp, httpClient *http.Client) (string, error) {
 	resp, err := httpClient.Get(url)
 	if err != nil {

tests/BUILD.bazel

@@ -54,6 +54,7 @@ go_test(
         "//pkg/controller:go_default_library",
         "//pkg/controller/common:go_default_library",
         "//pkg/controller/datavolume:go_default_library",
+        "//pkg/controller/populators:go_default_library",
         "//pkg/feature-gates:go_default_library",
         "//pkg/image:go_default_library",
         "//pkg/operator/resources/utils:go_default_library",

tests/framework/pvc.go

@@ -42,6 +42,19 @@ func (f *Framework) CreateBoundPVCFromDefinition(def *k8sv1.PersistentVolumeClai
 	return pvc
 }
 
+// CreateScheduledPVCFromDefinition is a wrapper around utils.CreatePVCFromDefinition that also triggeres
+// the scheduler to dynamically provision a pvc with WaitForFirstConsumer storage class by
+// executing f.ForceBindIfWaitForFirstConsumer(pvc)
+func (f *Framework) CreateScheduledPVCFromDefinition(def *k8sv1.PersistentVolumeClaim) *k8sv1.PersistentVolumeClaim {
+	pvc, err := utils.CreatePVCFromDefinition(f.K8sClient, f.Namespace.Name, def)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	pvc, err = utils.WaitForPVC(f.K8sClient, pvc.Namespace, pvc.Name)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	f.ForceSchedulingIfWaitForFirstConsumerPopulationPVC(pvc)
+	return pvc
+}
+
 // DeletePVC is a wrapper around utils.DeletePVC
 func (f *Framework) DeletePVC(pvc *k8sv1.PersistentVolumeClaim) error {
 	return utils.DeletePVC(f.K8sClient, f.Namespace.Name, pvc.Name)
@@ -82,6 +95,13 @@ func (f *Framework) ForceBindIfWaitForFirstConsumer(targetPvc *k8sv1.PersistentV
 	}
 }
 
+// ForceSchedulingIfWaitForFirstConsumerPopulationPVC creates a Pod with the passed in PVC mounted under /dev/pvc, which forces the PVC to be scheduled for provisioning.
+func (f *Framework) ForceSchedulingIfWaitForFirstConsumerPopulationPVC(targetPvc *k8sv1.PersistentVolumeClaim) {
+	if f.IsBindingModeWaitForFirstConsumer(targetPvc.Spec.StorageClassName) {
+		createConsumerPodForPopulationPVC(targetPvc, f)
+	}
+}
+
 func createConsumerPod(targetPvc *k8sv1.PersistentVolumeClaim, f *Framework) {
 	fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: creating \"consumer-pod\" to force binding PVC: %s\n", targetPvc.Name)
 	namespace := targetPvc.Namespace
@@ -101,6 +121,25 @@ func createConsumerPod(targetPvc *k8sv1.PersistentVolumeClaim, f *Framework) {
 	utils.DeletePodNoGrace(f.K8sClient, executorPod, namespace)
 }
 
+func createConsumerPodForPopulationPVC(targetPvc *k8sv1.PersistentVolumeClaim, f *Framework) {
+	fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: creating \"consumer-pod\" to get 'selected-node' annotation on PVC: %s\n", targetPvc.Name)
+	namespace := targetPvc.Namespace
+
+	err := utils.WaitForPersistentVolumeClaimPhase(f.K8sClient, targetPvc.Namespace, k8sv1.ClaimPending, targetPvc.Name)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	podName := naming.GetResourceName("consumer-pod", targetPvc.Name)
+	executorPod, err := f.CreateNoopPodWithPVC(podName, namespace, targetPvc)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	selectedNode, status, err := utils.WaitForPVCAnnotation(f.K8sClient, namespace, targetPvc, controller.AnnSelectedNodeName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	gomega.Expect(status).To(gomega.BeTrue())
+	gomega.Expect(selectedNode).ToNot(gomega.BeEmpty())
+
+	utils.DeletePodNoGrace(f.K8sClient, executorPod, namespace)
+}
+
 // VerifyPVCIsEmpty verifies a passed in PVC is empty, returns true if the PVC is empty, false if it is not. Optionaly, specify node for the pod.
 func VerifyPVCIsEmpty(f *Framework, pvc *k8sv1.PersistentVolumeClaim, node string) (bool, error) {
 	var err error

tests/import_test.go

@@ -33,6 +33,7 @@ import (
 	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
 	"kubevirt.io/containerized-data-importer/pkg/common"
 	controller "kubevirt.io/containerized-data-importer/pkg/controller/common"
+	"kubevirt.io/containerized-data-importer/pkg/controller/populators"
 	"kubevirt.io/containerized-data-importer/tests"
 	"kubevirt.io/containerized-data-importer/tests/framework"
 	"kubevirt.io/containerized-data-importer/tests/utils"
@@ -1439,6 +1440,303 @@ var _ = Describe("Preallocation", func() {
 	})
 })
 
+var _ = Describe("Import populator", func() {
+	f := framework.NewFramework(namespacePrefix)
+
+	var (
+		err                error
+		pvc                *v1.PersistentVolumeClaim
+		pvcPrime           *v1.PersistentVolumeClaim
+		tinyCoreIsoURL     = func() string { return fmt.Sprintf(utils.TinyCoreIsoURL, f.CdiInstallNs) }
+		tinyCoreArchiveURL = func() string { return fmt.Sprintf(utils.TarArchiveURL, f.CdiInstallNs) }
+		trustedRegistryURL = func() string { return fmt.Sprintf(utils.TrustedRegistryURL, f.DockerPrefix) }
+		imageioURL         = func() string { return fmt.Sprintf(utils.ImageioURL, f.CdiInstallNs) }
+		vcenterURL         = func() string { return fmt.Sprintf(utils.VcenterURL, f.CdiInstallNs) }
+	)
+
+	// importPopulationPVCDefinition creates a PVC with import datasourceref
+	importPopulationPVCDefinition := func() *v1.PersistentVolumeClaim {
+		pvcDef := utils.NewPVCDefinition("import-populator-pvc-test", "1Gi", nil, nil)
+		apiGroup := controller.AnnAPIGroup
+		pvcDef.Spec.DataSourceRef = &v1.TypedLocalObjectReference{
+			APIGroup: &apiGroup,
+			Kind:     cdiv1.ImportSourceRef,
+			Name:     "import-populator-test",
+		}
+		return pvcDef
+	}
+
+	// importPopulatorCR creates an import source CR
+	importPopulatorCR := func(namespace string, contentType cdiv1.DataVolumeContentType, preallocation bool) *cdiv1.ImportSource {
+		return &cdiv1.ImportSource{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "import-populator-test",
+				Namespace: namespace,
+			},
+			Spec: cdiv1.ImportSourceSpec{
+				ContentType:   contentType,
+				Preallocation: &preallocation,
+			},
+		}
+	}
+
+	// ImporSource creation functions
+
+	createHTTPImportPopulatorCR := func(contentType cdiv1.DataVolumeContentType, preallocation bool) error {
+		By("Creating Import Populator CR with HTTP source")
+		url := tinyCoreArchiveURL()
+		if contentType == cdiv1.DataVolumeKubeVirt {
+			url = tinyCoreIsoURL()
+		}
+		importPopulatorCR := importPopulatorCR(f.Namespace.Name, contentType, preallocation)
+		importPopulatorCR.Spec.HTTP = &cdiv1.DataVolumeSourceHTTP{
+			URL: url,
+		}
+		_, err := f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Create(
+			context.TODO(), importPopulatorCR, metav1.CreateOptions{})
+		return err
+	}
+
+	createRegistryImportPopulatorCR := func(contentType cdiv1.DataVolumeContentType, preallocation bool) error {
+		By("Creating Import Populator CR with Registry source")
+		registryURL := trustedRegistryURL()
+		pullMethod := cdiv1.RegistryPullNode
+		importPopulatorCR := importPopulatorCR(f.Namespace.Name, contentType, preallocation)
+		importPopulatorCR.Spec.Registry = &cdiv1.DataVolumeSourceRegistry{
+			URL:        &registryURL,
+			PullMethod: &pullMethod,
+		}
+		_, err := f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Create(
+			context.TODO(), importPopulatorCR, metav1.CreateOptions{})
+		return err
+	}
+
+	createImageIOImportPopulatorCR := func(contentType cdiv1.DataVolumeContentType, preallocation bool) error {
+		By("Creating Import Populator CR with ImageIO source")
+		cm, err := utils.CopyImageIOCertConfigMap(f.K8sClient, f.Namespace.Name, f.CdiInstallNs)
+		Expect(err).To(BeNil())
+		stringData := map[string]string{
+			common.KeyAccess: "admin@internal",
+			common.KeySecret: "12345",
+		}
+		tests.CreateImageIoDefaultInventory(f)
+		s, _ := utils.CreateSecretFromDefinition(f.K8sClient, utils.NewSecretDefinition(nil, stringData, nil, f.Namespace.Name, "mysecret"))
+		importPopulatorCR := importPopulatorCR(f.Namespace.Name, contentType, preallocation)
+		importPopulatorCR.Spec.Imageio = &cdiv1.DataVolumeSourceImageIO{
+			URL:           imageioURL(),
+			SecretRef:     s.Name,
+			CertConfigMap: cm,
+			DiskID:        "123",
+		}
+		_, err = f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Create(
+			context.TODO(), importPopulatorCR, metav1.CreateOptions{})
+		return err
+	}
+
+	createVDDKImportPopulatorCR := func(contentType cdiv1.DataVolumeContentType, preallocation bool) error {
+		By("Creating Import Populator CR with VDDK source")
+		// Find vcenter-simulator pod
+		pod, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, "vcenter-deployment", "app=vcenter")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(pod).ToNot(BeNil())
+
+		// Get test VM UUID
+		id, err := f.RunKubectlCommand("exec", "-n", pod.Namespace, pod.Name, "--", "cat", "/tmp/vmid")
+		Expect(err).To(BeNil())
+		vmid, err := uuid.Parse(strings.TrimSpace(id))
+		Expect(err).To(BeNil())
+
+		// Get disk name
+		disk, err := f.RunKubectlCommand("exec", "-n", pod.Namespace, pod.Name, "--", "cat", "/tmp/vmdisk")
+		Expect(err).To(BeNil())
+		disk = strings.TrimSpace(disk)
+		Expect(err).To(BeNil())
+
+		// Create VDDK login secret
+		stringData := map[string]string{
+			common.KeyAccess: "user",
+			common.KeySecret: "pass",
+		}
+		backingFile := disk
+		secretRef := "vddksecret"
+		thumbprint := "testprint"
+		s, _ := utils.CreateSecretFromDefinition(f.K8sClient, utils.NewSecretDefinition(nil, stringData, nil, f.Namespace.Name, secretRef))
+
+		importPopulatorCR := importPopulatorCR(f.Namespace.Name, contentType, preallocation)
+		importPopulatorCR.Spec.VDDK = &cdiv1.DataVolumeSourceVDDK{
+			BackingFile: backingFile,
+			SecretRef:   s.Name,
+			Thumbprint:  thumbprint,
+			URL:         vcenterURL(),
+			UUID:        vmid.String(),
+		}
+		_, err = f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Create(
+			context.TODO(), importPopulatorCR, metav1.CreateOptions{})
+		return err
+	}
+
+	createBlankImportPopulatorCR := func(contentType cdiv1.DataVolumeContentType, preallocation bool) error {
+		By("Creating Import Populator CR with blank source")
+		importPopulatorCR := importPopulatorCR(f.Namespace.Name, contentType, preallocation)
+		importPopulatorCR.Spec.Blank = &cdiv1.DataVolumeBlankImage{}
+		_, err := f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Create(
+			context.TODO(), importPopulatorCR, metav1.CreateOptions{})
+		return err
+	}
+
+	verifyCleanup := func(pvc *v1.PersistentVolumeClaim) {
+		if pvc != nil {
+			Eventually(func() bool {
+				// Make sure the pvc doesn't exist. The after each should have called delete.
+				_, err := f.FindPVC(pvc.Name)
+				return err != nil
+			}, timeout, pollingInterval).Should(BeTrue())
+		}
+	}
+
+	BeforeEach(func() {
+		verifyCleanup(pvc)
+	})
+
+	AfterEach(func() {
+		By("Deleting verifier pod")
+		err := utils.DeleteVerifierPod(f.K8sClient, f.Namespace.Name)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = f.CdiClient.CdiV1beta1().ImportSources(f.Namespace.Name).Delete(context.TODO(), "import-populator-source", metav1.DeleteOptions{})
+		if err != nil && !k8serrors.IsNotFound(err) {
+			Expect(err).ToNot(HaveOccurred())
+		}
+
+		By("Delete import population PVC")
+		err = f.DeletePVC(pvc)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	DescribeTable("should import fileSystem PVC", func(expectedMD5 string, importSourceFunc func(cdiv1.DataVolumeContentType, bool) error, preallocation bool) {
+		pvc = importPopulationPVCDefinition()
+		pvc = f.CreateScheduledPVCFromDefinition(pvc)
+		err = importSourceFunc(cdiv1.DataVolumeKubeVirt, preallocation)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify PVC prime was created")
+		pvcPrime, err = utils.WaitForPVC(f.K8sClient, pvc.Namespace, populators.PVCPrimeName(pvc))
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify target PVC is bound")
+		err = utils.WaitForPersistentVolumeClaimPhase(f.K8sClient, pvc.Namespace, v1.ClaimBound, pvc.Name)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify content")
+		md5, err := f.GetMD5(f.Namespace, pvc, utils.DefaultImagePath, utils.MD5PrefixSize)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(md5).To(Equal(expectedMD5))
+		By("Verifying the image is sparse")
+		Expect(f.VerifySparse(f.Namespace, pvc, utils.DefaultImagePath)).To(BeTrue())
+
+		if preallocation {
+			ok, err := f.VerifyImagePreallocated(f.Namespace, pvc)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(ok).To(BeTrue())
+		}
+
+		if utils.DefaultStorageCSIRespectsFsGroup {
+			// CSI storage class, it should respect fsGroup
+			By("Checking that disk image group is qemu")
+			Expect(f.GetDiskGroup(f.Namespace, pvc, false)).To(Equal("107"))
+		}
+
+		By("Verifying permissions are 660")
+		Expect(f.VerifyPermissions(f.Namespace, pvc)).To(BeTrue(), "Permissions on disk image are not 660")
+
+		By("Wait for PVC prime to be deleted")
+		Eventually(func() bool {
+			// Make sure pvcPrime was deleted after upload population
+			_, err := f.FindPVC(pvcPrime.Name)
+			return err != nil && k8serrors.IsNotFound(err)
+		}, timeout, pollingInterval).Should(BeTrue())
+	},
+		Entry("with HTTP image and preallocation", utils.TinyCoreMD5, createHTTPImportPopulatorCR, true),
+		Entry("with HTTP image withou preallocation", utils.TinyCoreMD5, createHTTPImportPopulatorCR, false),
+		Entry("with Registry image and preallocation", utils.TinyCoreMD5, createRegistryImportPopulatorCR, true),
+		Entry("with Registry image without preallocation", utils.TinyCoreMD5, createRegistryImportPopulatorCR, false),
+		Entry("with ImageIO image with preallocation", utils.ImageioMD5, createImageIOImportPopulatorCR, true),
+		Entry("with ImageIO image without preallocation", utils.ImageioMD5, createImageIOImportPopulatorCR, false),
+		Entry("with VDDK image with preallocation", utils.VcenterMD5, createVDDKImportPopulatorCR, true),
+		Entry("with VDDK image without preallocation", utils.VcenterMD5, createVDDKImportPopulatorCR, false),
+		Entry("with Blank image with preallocation", utils.BlankMD5, createBlankImportPopulatorCR, true),
+		Entry("with Blank image without preallocation", utils.BlankMD5, createBlankImportPopulatorCR, false),
+	)
+
+	DescribeTable("should import Block PVC", func(expectedMD5 string, importSourceFunc func(cdiv1.DataVolumeContentType, bool) error) {
+		if !f.IsBlockVolumeStorageClassAvailable() {
+			Skip("Storage Class for block volume is not available")
+		}
+
+		pvc = importPopulationPVCDefinition()
+		volumeMode := v1.PersistentVolumeBlock
+		pvc.Spec.VolumeMode = &volumeMode
+		pvc = f.CreateScheduledPVCFromDefinition(pvc)
+		err = importSourceFunc(cdiv1.DataVolumeKubeVirt, true)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify PVC prime was created")
+		pvcPrime, err = utils.WaitForPVC(f.K8sClient, pvc.Namespace, populators.PVCPrimeName(pvc))
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify target PVC is bound")
+		err = utils.WaitForPersistentVolumeClaimPhase(f.K8sClient, pvc.Namespace, v1.ClaimBound, pvc.Name)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify content")
+		md5, err := f.GetMD5(f.Namespace, pvc, utils.DefaultPvcMountPath, utils.MD5PrefixSize)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(md5).To(Equal(expectedMD5))
+		By("Verifying the image is sparse")
+		Expect(f.VerifySparse(f.Namespace, pvc, utils.DefaultPvcMountPath)).To(BeTrue())
+
+		By("Wait for PVC prime to be deleted")
+		Eventually(func() bool {
+			// Make sure pvcPrime was deleted after upload population
+			_, err := f.FindPVC(pvcPrime.Name)
+			return err != nil && k8serrors.IsNotFound(err)
+		}, timeout, pollingInterval).Should(BeTrue())
+	},
+		Entry("with HTTP image", utils.TinyCoreMD5, createHTTPImportPopulatorCR),
+		Entry("with Registry image", utils.TinyCoreMD5, createRegistryImportPopulatorCR),
+		Entry("with ImageIO image", utils.ImageioMD5, createImageIOImportPopulatorCR),
+		Entry("with VDDK image", utils.VcenterMD5, createVDDKImportPopulatorCR),
+		Entry("with Blank image", utils.BlankMD5, createBlankImportPopulatorCR),
+	)
+
+	It("should import archive", func() {
+		pvc = importPopulationPVCDefinition()
+		pvc = f.CreateScheduledPVCFromDefinition(pvc)
+		err = createHTTPImportPopulatorCR(cdiv1.DataVolumeArchive, true)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify PVC prime was created")
+		pvcPrime, err = utils.WaitForPVC(f.K8sClient, pvc.Namespace, populators.PVCPrimeName(pvc))
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify target PVC is bound")
+		err = utils.WaitForPersistentVolumeClaimPhase(f.K8sClient, pvc.Namespace, v1.ClaimBound, pvc.Name)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify content")
+		same, err := f.VerifyTargetPVCArchiveContent(f.Namespace, pvc, "3")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(same).To(BeTrue())
+
+		By("Wait for PVC prime to be deleted")
+		Eventually(func() bool {
+			// Make sure pvcPrime was deleted after upload population
+			_, err := f.FindPVC(pvcPrime.Name)
+			return err != nil && k8serrors.IsNotFound(err)
+		}, timeout, pollingInterval).Should(BeTrue())
+	})
+})
+
 func generateRegistryOnlySidecar() *unstructured.Unstructured {
 	return &unstructured.Unstructured{
 		Object: map[string]interface{}{