feat: add encryption/decryption integration test

apiclient/harvester_api/api.py

@@ -84,6 +84,7 @@ def load_managers(self, version="0.0.0"):
         self.lhreplicas = mgrs.LonghornReplicaManager.for_version(version)(self, version)
         self.lhvolumes = mgrs.LonghornVolumeManager.for_version(version)(self, version)
         self.lhbackupvolumes = mgrs.LonghornBackupVolumeManager.for_version(version)(self, version)
+        self.secrets = mgrs.SecretManager.for_version(version)(self, version)
 
     def _get(self, path, **kwargs):
         url = self.get_url(path)

apiclient/harvester_api/managers/__init__.py

@@ -12,6 +12,7 @@
 from .clusternetworks import ClusterNetworkManager
 from .volumesnapshots import VolumeSnapshotManager
 from .virtualmachines import VirtualMachineManager
+from .secrets import SecretManager
 from .addons import AddonManager
 # Not available in dashboard
 from .internals import VersionManager, UpgradeManager, DEFAULT_HARVESTER_NAMESPACE
@@ -32,6 +33,7 @@
  "VolumeSnapshotManager",
  "ClusterNetworkManager",
  "VirtualMachineManager",
+ "SecretManager",
  "StorageClassManager",
  "AddonManager",
  "BackupManager", "VirtualMachineSnapshotManager",

apiclient/harvester_api/managers/images.py

@@ -77,3 +77,57 @@ def delete(self, name, namespace=DEFAULT_NAMESPACE, *, raw=False):
 
     def download(self, name, namespace=DEFAULT_NAMESPACE):
         return self._get(self.DOWNLOAD_fmt.format(uid=name, ns=namespace), raw=True)
+
+    def create_by_encrypt(
+        self,
+        source_image_name,
+        new_image_name,
+        storage_class_name,
+        namespace=DEFAULT_NAMESPACE,
+    ):
+        data = self.create_data(
+            name=new_image_name,
+            url=None,
+            desc="",
+            stype="clone",
+            namespace=namespace,
+            image_checksum=None,
+            display_name=new_image_name,
+            storageclass=storage_class_name
+        )
+
+        # Add securityParameters
+        data["spec"]["securityParameters"] = {
+            "cryptoOperation": "encrypt",
+            "sourceImageName": source_image_name,
+            "sourceImageNamespace": namespace,
+        }
+
+        return self.create("", namespace=namespace, json=data)
+
+    def create_by_decrypt(
+        self,
+        source_image_name,
+        new_image_name,
+        storage_class_name,
+        namespace=DEFAULT_NAMESPACE,
+    ):
+        data = self.create_data(
+            name=new_image_name,
+            url=None,
+            desc="",
+            stype="clone",
+            namespace=namespace,
+            image_checksum=None,
+            display_name=new_image_name,
+            storageclass=storage_class_name
+        )
+
+        # Add securityParameters
+        data["spec"]["securityParameters"] = {
+            "cryptoOperation": "decrypt",
+            "sourceImageName": source_image_name,
+            "sourceImageNamespace": namespace,
+        }
+
+        return self.create("", namespace=namespace, json=data)

apiclient/harvester_api/managers/secrets.py

@@ -0,0 +1,34 @@
+import base64
+from .base import BaseManager, DEFAULT_NAMESPACE
+
+
+class SecretManager(BaseManager):
+    PATH_fmt = "/api/v1/namespaces/{ns}/secrets/{name}"
+    CREATE_fmt = "v1/harvester/secrets"
+
+    def create_data(self, name, namespace, data, annotations=None):
+        annotations = annotations or {}
+        encoded_data = {
+            k: base64.b64encode(v.encode('utf-8')).decode('utf-8')
+            for k, v in data.items()
+        }
+
+        return {
+            "type": "secret",
+            "metadata": {
+                "namespace": namespace,
+                "name": name,
+                "annotations": annotations
+            },
+            "data": encoded_data
+        }
+
+    def create(self, name, data, namespace=DEFAULT_NAMESPACE, annotations=None, **kwargs):
+        secret_data = self.create_data(name, namespace, data, annotations=annotations)
+        return self._create(self.CREATE_fmt, json=secret_data, **kwargs)
+
+    def get(self, name, namespace=DEFAULT_NAMESPACE, *, raw=False):
+        return self._get(self.PATH_fmt.format(ns=namespace, name=name), raw=raw)
+
+    def delete(self, name, namespace=DEFAULT_NAMESPACE, *, raw=False):
+        return self._delete(self.PATH_fmt.format(ns=namespace, name=name), raw=raw)

apiclient/harvester_api/managers/storageclasses.py

@@ -1,4 +1,4 @@
-from .base import BaseManager
+from .base import BaseManager, merge_dict
 
 DEFAULT_STORAGE_CLASS_ANNOTATION = "storageclass.kubernetes.io/is-default-class"
 
@@ -23,7 +23,7 @@ def get_default(self):
         else:
             return code, data
 
-    def create_data(self, name, replicas):
+    def create_data(self, name, replicas, **options):
         data = {
             "type": f"{self.API_VERSION}",
             "metadata": {
@@ -39,12 +39,11 @@ def create_data(self, name, replicas):
             "reclaimPolicy": "Delete",
             "volumeBindingMode": "Immediate"
         }
+        return merge_dict(options, data)
 
-        return data
-
-    def create(self, name, replicas=3, *, raw=False):
+    def create(self, name, replicas=3, *, raw=False, **options):
         path = self.CREATE_PATH_fmt.format(SC_API=self.API_VERSION)
-        data = self.create_data(name, replicas)
+        data = self.create_data(name, replicas, **options)
         return self._create(path, json=data, raw=raw)
 
     def set_default(self, name, *, raw=False):