feat: add on_spot_discontinue on os_volume and delete_permanently on instance action

Author: Ruslan Gainutdinov

tests/unit_tests/instances/test_instances.py

@@ -1,9 +1,10 @@
+import json
 import pytest
 import responses  # https://github.com/getsentry/responses
 
 from verda.constants import Actions, ErrorCodes, Locations
 from verda.exceptions import APIException
-from verda.instances import Instance, InstancesService
+from verda.instances import Instance, InstancesService, OSVolume
 
 INVALID_REQUEST = ErrorCodes.INVALID_REQUEST
 INVALID_REQUEST_MESSAGE = 'Your existence is invalid'
@@ -266,6 +267,31 @@ def test_create_spot_instance_successful(self, instances_service, endpoint):
         assert responses.assert_call_count(endpoint, 1) is True
         assert responses.assert_call_count(url, 1) is True
 
+    def test_create_spot_instance(self, instances_service, endpoint):
+        # arrange
+        responses.add(responses.POST, endpoint, body=INSTANCE_ID, status=200)
+        url = endpoint + '/' + INSTANCE_ID
+        responses.add(responses.GET, url, json=PAYLOAD[0], status=200)
+
+        SPOT_INSTANCE_OS_VOLUME = OSVolume(name='spot-instance-os-volume', size=50, on_spot_discontinue='delete_permanently')
+
+        # act
+        instances_service.create(
+            instance_type=INSTANCE_TYPE,
+            image=INSTANCE_IMAGE,
+            ssh_key_ids=[SSH_KEY_ID],
+            hostname=INSTANCE_HOSTNAME,
+            description=INSTANCE_DESCRIPTION,
+            os_volume=SPOT_INSTANCE_OS_VOLUME,
+        )
+
+        # assert
+        request_body = responses.calls[0].request.body.decode('utf-8')
+        body = json.loads(request_body)
+        assert body['os_volume']['name'] == SPOT_INSTANCE_OS_VOLUME.name
+        assert body['os_volume']['size'] == SPOT_INSTANCE_OS_VOLUME.size
+        assert body['os_volume']['on_spot_discontinue'] == 'delete_permanently'
+
     def test_create_instance_attached_os_volume_successful(self, instances_service, endpoint):
         # arrange - add response mock
         # create instance
@@ -340,6 +366,28 @@ def test_action_successful(self, instances_service, endpoint):
         assert result is None
         assert responses.assert_call_count(url, 1) is True
 
+    def test_action_with_delete_permanently_sends_payload(self, instances_service, endpoint):
+        # arrange
+        url = endpoint
+        responses.add(responses.PUT, url, status=202)
+        volume_ids = [OS_VOLUME_ID]
+
+        # act
+        instances_service.action(
+            id_list=[INSTANCE_ID],
+            action=Actions.DELETE,
+            volume_ids=volume_ids,
+            delete_permanently=True,
+        )
+
+        # assert
+        request_body = responses.calls[0].request.body.decode('utf-8')
+        body = json.loads(request_body)
+        assert body['id'] == [INSTANCE_ID]
+        assert body['action'] == Actions.DELETE
+        assert body['volume_ids'] == volume_ids
+        assert body['delete_permanently'] is True
+
     def test_action_failed(self, instances_service, endpoint):
         # arrange - add response mock
         url = endpoint

verda/instances/__init__.py

@@ -1 +1,8 @@
-from ._instances import Contract, Instance, InstancesService, Pricing
+from ._instances import (
+    Contract,
+    Instance,
+    InstancesService,
+    OnSpotDiscontinue,
+    Pricing,
+    OSVolume,
+)

verda/instances/_instances.py

@@ -3,14 +3,33 @@
 from dataclasses import dataclass
 from typing import Literal
 
-from dataclasses_json import dataclass_json
+from dataclasses_json import Undefined, dataclass_json
 
 from verda.constants import InstanceStatus, Locations
 
 INSTANCES_ENDPOINT = '/instances'
 
 Contract = Literal['LONG_TERM', 'PAY_AS_YOU_GO', 'SPOT']
 Pricing = Literal['DYNAMIC_PRICE', 'FIXED_PRICE']
+OnSpotDiscontinue = Literal['keep_detached', 'move_to_trash', 'delete_permanently']
+
+@dataclass_json(undefined=Undefined.EXCLUDE)
+@dataclass
+class OSVolume:
+    """Represents an operating system volume.
+    
+    Attributes:
+        name: Name of the volume.
+        size: Size of the volume in GB.
+        on_spot_discontinue: What to do with the volume on spot discontinue.
+            - keep_detached: Keep the volume detached.
+            - move_to_trash: Move the volume to trash.
+            - delete_permanently: Delete the volume permanently.
+            Defaults to keep_detached.
+    """
+    name: str
+    size: int
+    on_spot_discontinue: OnSpotDiscontinue | None = None
 
 
 @dataclass_json
@@ -123,7 +142,7 @@ def create(
         startup_script_id: str | None = None,
         volumes: list[dict] | None = None,
         existing_volumes: list[str] | None = None,
-        os_volume: dict | None = None,
+        os_volume: OSVolume | None = None,
         is_spot: bool = False,
         contract: Contract | None = None,
         pricing: Pricing | None = None,
@@ -204,21 +223,32 @@ def action(
         id_list: list[str] | str,
         action: str,
         volume_ids: list[str] | None = None,
+        delete_permanently: bool = False,
     ) -> None:
         """Performs an action on one or more instances.
 
         Args:
             id_list: Single instance ID or list of instance IDs to act upon.
             action: Action to perform on the instances.
             volume_ids: Optional list of volume IDs to delete.
+            delete_permanently: When deleting (or discontinuing), delete the
+                given volume IDs permanently. Only applicable when volume_ids
+                is also provided.
 
         Raises:
             HTTPError: If the action fails or other API error occurs.
         """
         if type(id_list) is str:
             id_list = [id_list]
 
-        payload = {'id': id_list, 'action': action, 'volume_ids': volume_ids}
+        payload = {
+            'id': id_list,
+            'action': action,
+            'volume_ids': volume_ids,
+        }
+
+        if delete_permanently:
+            payload['delete_permanently'] = True
 
         self._http_client.put(INSTANCES_ENDPOINT, json=payload)
         return