release 1.3.0

Author: tamirse

CHANGELOG.rst

@@ -1,6 +1,11 @@
 Changelog
 =========
 
+v1.3.0 (2023-05-25)
+-------------------
+
+* Added support for volume cloning
+
 v1.2.0 (2023-04-24)
 -------------------
 

datacrunch/__version__.py

@@ -1 +1 @@
-VERSION = '1.2.0'
+VERSION = '1.3.0'

datacrunch/constants.py

@@ -15,6 +15,7 @@ class VolumeActions:
     RENAME = 'rename'
     INCREASE_SIZE = 'increase-size'
     DELETE = 'delete'
+    CLONE = 'clone'
 
     def __init__(self):
         return
@@ -40,6 +41,7 @@ class VolumeStatus:
     DETACHED = "detached"
     DELETING = "deleting"
     DELETED = "deleted"
+    CLONING = 'cloning'
 
     def __init__(self):
         return

datacrunch/volumes/volumes.py

@@ -46,7 +46,6 @@ def __init__(self,
         :param ssh_key_ids: list of ssh keys ids
         :type ssh_key_ids: List[str]
         """
-
         self._id = id
         self._status = status
         self._name = name
@@ -166,6 +165,7 @@ def __str__(self) -> str:
         """
         return stringify_class_object_properties(self)
 
+
 class VolumesService:
     """A service for interacting with the volumes endpoint"""
 
@@ -193,7 +193,8 @@ def get(self, status: str = None) -> List[Volume]:
             target=volume_dict['target'] if 'target' in volume_dict else None,
             location=volume_dict['location'],
             instance_id=volume_dict['instance_id'] if 'instance_id' in volume_dict else None,
-            ssh_key_ids=volume_dict['ssh_key_ids'] if 'ssh_key_ids' in volume_dict else [],
+            ssh_key_ids=volume_dict['ssh_key_ids'] if 'ssh_key_ids' in volume_dict else [
+            ],
         ), volumes_dict))
         return volumes
 
@@ -218,7 +219,8 @@ def get_by_id(self, id: str) -> Volume:
             target=volume_dict['target'] if 'target' in volume_dict else None,
             location=volume_dict['location'],
             instance_id=volume_dict['instance_id'] if 'instance_id' in volume_dict else None,
-            ssh_key_ids=volume_dict['ssh_key_ids'] if 'ssh_key_ids' in volume_dict else [],
+            ssh_key_ids=volume_dict['ssh_key_ids'] if 'ssh_key_ids' in volume_dict else [
+            ],
         )
         return volume
 
@@ -288,6 +290,40 @@ def detach(self, id_list: Union[List[str], str]) -> None:
         self._http_client.put(VOLUMES_ENDPOINT, json=payload)
         return
 
+    def clone(self, id: str, name: str = None, type: str = None) -> Volume:
+        """Clone a volume or multiple volumes
+
+        :param id: volume id or list of volume ids
+        :type id: str or List[str]
+        :param name: new volume name
+        :type name: str
+        :param type: volume type
+        :type type: str, optional
+        :return: the new volume object, or a list of volume objects if cloned mutliple volumes
+        :rtype: Volume or List[Volume]
+        """
+        payload = {
+            "id": id,
+            "action": VolumeActions.CLONE,
+            "name": name,
+            "type": type
+        }
+
+        # clone volume(s)
+        volume_ids_array = self._http_client.put(
+            VOLUMES_ENDPOINT, json=payload).json()
+
+        # map the IDs into Volume objects
+        volumes_array = list(
+            map(lambda volume_id: self.get_by_id(volume_id), volume_ids_array))
+
+        # if the array has only one element, return that element
+        if len(volumes_array) == 1:
+            return volumes_array[0]
+
+        # otherwise return the volumes array
+        return volumes_array
+
     def rename(self, id_list: Union[List[str], str], name: str) -> None:
         """Rename multiple volumes or single volume
 

examples/storage_volumes.py

@@ -19,10 +19,12 @@
 all_volumes = datacrunch.volumes.get()
 
 # Get all attached volumes
-all_attached_volumes = datacrunch.volumes.get(status=datacrunch.constants.volume_status.ATTACHED)
+all_attached_volumes = datacrunch.volumes.get(
+    status=datacrunch.constants.volume_status.ATTACHED)
 
 # Get volume by id
-random_volume = datacrunch.volumes.get_by_id("0c41e387-3dd8-495f-a285-e861527f2f3d")
+random_volume = datacrunch.volumes.get_by_id(
+    "0c41e387-3dd8-495f-a285-e861527f2f3d")
 
 # Create a 200 GB detached NVMe volume
 nvme_volume = datacrunch.volumes.create(type=NVMe,
@@ -51,5 +53,14 @@
 # increase volume size
 datacrunch.volumes.increase_size(nvme_volume_id, 300)
 
+# clone volume
+datacrunch.volumes.clone(nvme_volume_id)
+
+# clone volume and give it a new name and storage type (from NVMe to HDD)
+datacrunch.volumes.clone(nvme_volume_id, name="my-cloned-volume", type=HDD)
+
+# clone multiple volumes at once
+datacrunch.volumes.clone([nvme_volume_id, hdd_volume_id])
+
 # delete volumes
 datacrunch.volumes.delete([nvme_volume_id, hdd_volume_id])

tests/unit_tests/volumes/test_volumes.py

@@ -29,6 +29,9 @@
 HDD_VOL_SIZE = 100
 HDD_VOL_CREATED_AT = "2021-06-02T12:56:49.582Z"
 
+RANDOM_VOL_ID = '07d864ee-ba86-451e-85b3-34ef551bd4a2'
+RANDOM_VOL2_ID = '72c5c082-7fe7-4d13-bd9e-f529c97d63b3'
+
 NVME_VOLUME = {
     "id": NVME_VOL_ID,
     "status": NVME_VOL_STATUS,
@@ -69,7 +72,23 @@ def volumes_service(self, http_client):
     def endpoint(self, http_client):
         return http_client._base_url + "/volumes"
 
-    def test_get_instances(self, volumes_service, endpoint):
+    def test_initialize_a_volume(self):
+        volume = Volume(RANDOM_VOL_ID, VolumeStatus.DETACHED, HDD_VOL_NAME, HDD_VOL_SIZE,
+                        HDD, False, HDD_VOL_CREATED_AT)
+
+        assert volume.id == RANDOM_VOL_ID
+        assert volume.status == VolumeStatus.DETACHED
+        assert volume.instance_id == None
+        assert volume.name == HDD_VOL_NAME
+        assert volume.size == HDD_VOL_SIZE
+        assert volume.type == HDD
+        assert volume.location == FIN1
+        assert volume.is_os_volume == False
+        assert volume.created_at == HDD_VOL_CREATED_AT
+        assert volume.target == None
+        assert volume.ssh_key_ids == []
+
+    def test_get_volumes(self, volumes_service, endpoint):
         # arrange - add response mock
         responses.add(
             responses.GET,
@@ -219,7 +238,8 @@ def test_create_volume_successful(self, volumes_service, endpoint):
         )
 
         # act
-        volume = volumes_service.create(VolumeTypes.NVMe, NVME_VOL_NAME, NVME_VOL_SIZE)
+        volume = volumes_service.create(
+            VolumeTypes.NVMe, NVME_VOL_NAME, NVME_VOL_SIZE)
 
         # assert
         assert volume.id == NVME_VOL_ID
@@ -236,7 +256,8 @@ def test_create_volume_failed(self, volumes_service, endpoint):
 
         # act
         with pytest.raises(APIException) as excinfo:
-            volumes_service.create(VolumeTypes.NVMe, NVME_VOL_NAME, 100000000000000000000000)
+            volumes_service.create(
+                VolumeTypes.NVMe, NVME_VOL_NAME, 100000000000000000000000)
 
         # assert
         assert excinfo.value.code == INVALID_REQUEST
@@ -481,3 +502,139 @@ def test_delete_volume_failed(self, volumes_service, endpoint):
         assert excinfo.value.code == INVALID_REQUEST
         assert excinfo.value.message == INVALID_REQUEST_MESSAGE
         assert responses.assert_call_count(endpoint, 1) is True
+
+    def test_clone_volume_with_input_name_successful(self, volumes_service, endpoint):
+        # arrange
+        CLONED_VOLUME_NAME = "cloned-volume"
+
+        # mock response for cloning the volume
+        responses.add(
+            responses.PUT,
+            endpoint,
+            status=202,
+            json=[RANDOM_VOL_ID],
+            match=[
+                responses.json_params_matcher({
+                    "id": NVME_VOL_ID,
+                    "action": VolumeActions.CLONE,
+                    "name": CLONED_VOLUME_NAME,
+                    "type": None
+                })
+            ]
+        )
+
+        # mock object for the cloned volume
+        CLONED_VOL_GET_MOCK = NVME_VOLUME
+        CLONED_VOL_GET_MOCK['id'] = RANDOM_VOL_ID
+        CLONED_VOL_GET_MOCK['name'] = CLONED_VOLUME_NAME
+        CLONED_VOL_GET_MOCK['status'] = VolumeStatus.CLONING
+
+        # mock response for getting the cloned volume
+        responses.add(
+            responses.GET,
+            endpoint + "/" + RANDOM_VOL_ID,
+            status=200,
+            json=CLONED_VOL_GET_MOCK,
+        )
+
+        # act
+        cloned_volume = volumes_service.clone(NVME_VOL_ID, CLONED_VOLUME_NAME)
+
+        # assert
+        assert responses.assert_call_count(endpoint, 1) is True
+        assert cloned_volume.name == CLONED_VOLUME_NAME
+
+    def test_clone_volume_without_input_name_successful(self, volumes_service: VolumesService, endpoint):
+        # arrange
+        CLONED_VOLUME_NAME = "CLONE-" + NVME_VOL_NAME
+
+        # mock response for cloning the volume
+        responses.add(
+            responses.PUT,
+            endpoint,
+            status=202,
+            json=[RANDOM_VOL_ID],
+            match=[
+                responses.json_params_matcher({
+                    "id": NVME_VOL_ID,
+                    "action": VolumeActions.CLONE,
+                    "name": None,
+                    "type": None
+                })
+            ]
+        )
+
+        # mock object for the cloned volume
+        CLONED_VOL_GET_MOCK = NVME_VOLUME
+        CLONED_VOL_GET_MOCK['id'] = RANDOM_VOL_ID
+        CLONED_VOL_GET_MOCK['name'] = CLONED_VOLUME_NAME
+        CLONED_VOL_GET_MOCK['status'] = VolumeStatus.CLONING
+
+        # mock response for getting the cloned volume
+        responses.add(
+            responses.GET,
+            endpoint + "/" + RANDOM_VOL_ID,
+            status=200,
+            json=CLONED_VOL_GET_MOCK,
+        )
+
+        # act
+        cloned_volume = volumes_service.clone(NVME_VOL_ID)
+
+        # assert
+        assert responses.assert_call_count(endpoint, 1) is True
+        assert cloned_volume.name == CLONED_VOLUME_NAME
+
+    def test_clone_two_volumes_successful(self, volumes_service: VolumesService, endpoint):
+        # arrange
+        CLONED_VOL1_NAME = "CLONE-" + NVME_VOL_NAME
+        CLONED_VOL2_NAME = "CLONE-" + HDD_VOL_NAME
+
+        # mock response for cloning the volumes
+        responses.add(
+            responses.PUT,
+            endpoint,
+            status=202,
+            json=[RANDOM_VOL_ID, RANDOM_VOL2_ID],
+            match=[
+                responses.json_params_matcher({
+                    "id": [NVME_VOL_ID, HDD_VOL_ID],
+                    "action": VolumeActions.CLONE,
+                    "name": None,
+                    "type": None
+                })
+            ]
+        )
+
+        # mock object for the cloned volumes
+        CLONED_VOL1_GET_MOCK = NVME_VOLUME
+        CLONED_VOL1_GET_MOCK['id'] = RANDOM_VOL_ID
+        CLONED_VOL1_GET_MOCK['name'] = CLONED_VOL1_NAME
+        CLONED_VOL1_GET_MOCK['status'] = VolumeStatus.CLONING
+
+        CLONED_VOL2_GET_MOCK = HDD_VOLUME
+        CLONED_VOL2_GET_MOCK['id'] = RANDOM_VOL2_ID
+        CLONED_VOL2_GET_MOCK['name'] = CLONED_VOL2_NAME
+        CLONED_VOL2_GET_MOCK['status'] = VolumeStatus.CLONING
+
+        # mock response for getting the cloned volumes
+        responses.add(
+            responses.GET,
+            endpoint + "/" + RANDOM_VOL_ID,
+            status=200,
+            json=CLONED_VOL1_GET_MOCK,
+        )
+        responses.add(
+            responses.GET,
+            endpoint + "/" + RANDOM_VOL2_ID,
+            status=200,
+            json=CLONED_VOL2_GET_MOCK,
+        )
+
+        # act
+        cloned_volume = volumes_service.clone([NVME_VOL_ID, HDD_VOL_ID])
+
+        # assert
+        assert responses.assert_call_count(endpoint, 1) is True
+        assert cloned_volume[0].name == CLONED_VOL1_NAME
+        assert cloned_volume[1].name == CLONED_VOL2_NAME