qa: add Windows import/export tests

We're adding a few tests that exercise the rbd import/export
functionality on Windows. This also covers block devices,
which are actually rbd images mapped using rbd-wnbd.

Signed-off-by: Lucian Petrut <lpetrut@cloudbasesolutions.com>

qa/workunits/windows/py_tests/internal/rbd_image.py

@@ -46,6 +46,28 @@ def create(cls,
 
         return RbdImage(name, size_mb, is_shared)
 
+    @classmethod
+    @Tracer.trace
+    def import_image(cls, name: str, path: str, export_format: int = 1):
+        LOG.info("Importing image: %s. Path: %s.", name, path)
+        cmd = ["rbd", "import", "--export-format",
+               str(export_format), path, name]
+        utils.execute(*cmd)
+
+        cmd = ["rbd", "info", name, "--format=json"]
+        result = utils.ps_execute(*cmd)
+        size = json.loads(result.stdout)["size"]
+        size_mb = size / (1 << 20)
+
+        return RbdImage(name, size_mb, False)
+
+    @Tracer.trace
+    def export_image(self, path: str, export_format: int = 1):
+        LOG.info("Exporting image: %s. Path: %s.", self.name, path)
+        cmd = ["rbd", "export", "--export-format",
+               str(export_format), self.name, path]
+        utils.execute(*cmd)
+
     @Tracer.trace
     def get_disk_number(self,
                         timeout: int = 60,

qa/workunits/windows/py_tests/internal/unittest_base.py

@@ -0,0 +1,27 @@
+import logging
+import os
+import unittest
+
+from py_tests.internal import utils
+
+_logging_configured = False
+
+
+class TestBase(unittest.TestCase):
+    @staticmethod
+    def _setup_logging(debug=False, verbose=False):
+        log_level = logging.WARNING
+        if verbose:
+            log_level = logging.INFO
+        if debug:
+            log_level = logging.DEBUG
+        utils.setup_logging(log_level)
+
+    @classmethod
+    def setUpClass(cls):
+        global _logging_configured
+        if not _logging_configured:
+            cls._setup_logging(
+                debug=utils.str2bool(os.getenv('CEPHTEST_DEBUG')),
+                verbose=utils.str2bool(os.getenv('CEPHTEST_VERBOSE')))
+            _logging_configured = True

qa/workunits/windows/py_tests/internal/utils.py

@@ -7,8 +7,10 @@
 
 import collections
 import functools
+import hashlib
 import logging
 import math
+import os
 import subprocess
 import time
 import typing
@@ -117,3 +119,37 @@ def array_stats(array: list):
         'std_dev': std_dev,
         'count': len(array)
     }
+
+
+def generate_random_file(path, size, chunk_size=8192):
+    if size % chunk_size:
+        raise exception.CephTestException(
+            f"The file size ({size}) is not a multiple of the "
+            f"chunk size ({chunk_size}).")
+    with open(path, 'rb+') as f:
+        for chunk_idx in range(size // chunk_size):
+            f.write(os.urandom(chunk_size))
+
+
+def checksum(path, algorithm='md5', chunk_size=1048576):
+    total_read = 0
+    hash_func = getattr(hashlib, algorithm)
+    file_hash = hash_func()
+
+    with open(path, "rb") as f:
+        try:
+            while chunk := f.read(chunk_size):
+                file_hash.update(chunk)
+                total_read += len(chunk)
+        except PermissionError:
+            # Windows throws a permission error when reading past
+            # the disk boundary.
+            if not total_read:
+                raise
+
+    return file_hash.hexdigest()
+
+
+def str2bool(val):
+    val = val or ''
+    return val.lower() in ['y', 'yes', 'true', 't', '1', 'enabled']

qa/workunits/windows/py_tests/rbd/test_import_export.py

@@ -0,0 +1,84 @@
+import os
+import tempfile
+import uuid
+
+from py_tests.internal import exception
+from py_tests.internal import utils
+from py_tests.internal import rbd_image
+from py_tests.internal import unittest_base
+
+
+class TestImportExport(unittest_base.TestBase):
+    _image_size = 32 << 20
+    _image_prefix = 'cephImportTest-'
+
+    @classmethod
+    def _generate_img_name(cls):
+        return cls._image_prefix + str(uuid.uuid4()).split('-')[-1]
+
+    def test_bdev_export_import(self):
+        fd, import_path = tempfile.mkstemp(prefix=self._image_prefix)
+        os.close(fd)
+        self.addCleanup(os.remove, import_path)
+        utils.generate_random_file(import_path, self._image_size)
+
+        # Map the rbd image and then import the resulting block device
+        img = rbd_image.RbdImage.import_image(self._generate_img_name(), import_path)
+        self.addCleanup(img.cleanup)
+        img.map()
+
+        bdev_imported_img = rbd_image.RbdImage.import_image(
+            self._generate_img_name(), img.path)
+        self.addCleanup(bdev_imported_img.cleanup)
+
+        # Create a new image, map it and then export our image there.
+        bdev_exported_img = rbd_image.RbdImage.create(
+            self._generate_img_name(),
+            size_mb=self._image_size >> 20)
+        self.addCleanup(bdev_exported_img.cleanup)
+        bdev_exported_img.map()
+
+        bdev_imported_img.export_image(bdev_exported_img.path)
+        bdev_imported_img.map()
+
+        # validate mapped disk size
+        self.assertEqual(self._image_size, img.get_disk_size())
+        self.assertEqual(self._image_size, bdev_imported_img.get_disk_size())
+        self.assertEqual(self._image_size, bdev_exported_img.get_disk_size())
+
+        # validate contents
+        file_checksum = utils.checksum(import_path)
+        img_checksum = utils.checksum(img.path)
+        bdev_exported_checksum = utils.checksum(bdev_exported_img.path)
+        bdev_imported_checksum = utils.checksum(bdev_imported_img.path)
+
+        self.assertEqual(file_checksum, img_checksum)
+        self.assertEqual(file_checksum, bdev_exported_checksum)
+        self.assertEqual(file_checksum, bdev_imported_checksum)
+
+    def test_bdev_v2_import_export(self):
+        # v2 exports are not allowed with Windows block devices
+        img = rbd_image.RbdImage.create(
+            self._generate_img_name(),
+            size_mb=self._image_size >> 20)
+        self.addCleanup(img.cleanup)
+        img.map()
+
+        self.assertRaises(
+            exception.CommandFailed,
+            rbd_image.RbdImage.import_image,
+            self._generate_img_name(),
+            img.path,
+            export_format=2)
+
+        img2 = rbd_image.RbdImage.create(
+            self._generate_img_name(),
+            size_mb=self._image_size >> 20)
+        self.addCleanup(img2.cleanup)
+        img2.map()
+
+        self.assertRaises(
+            exception.CommandFailed,
+            img.export_image,
+            img2.path,
+            export_format=2)

qa/workunits/windows/run-tests.ps1

@@ -40,3 +40,6 @@ safe_exec python.exe -m py_tests.rbd_wnbd.service_restart_test `
     --test-name=RbdFsFioTest --iterations=3 --image-count=8 --concurrency=8 --image-size-mb=64
 safe_exec python.exe -m py_tests.rbd_wnbd.service_restart_test `
     --test-name=RbdFsStampTest --iterations=3 --image-count=8 --concurrency=8 --image-size-mb=64
+
+# Automatically detect and run Python tests that use the unittest framework.
+safe_exec python.exe -m unittest discover -d "$scriptLocation"