feat: on_spot_discontinue on os_volume and delete_permanently on instance action (#74)

huksley · Ruslan Gainutdinov · web-flow · commit a69239ac30f1 · 2026-02-05T09:58:35.000+02:00
* feat: add on_spot_discontinue on os_volume and delete_permanently on instance action

* fix: tests

* fix: reformat

* fix: formatting

* fix: add test for spot

* fix: format

* fix: copilot review fixes

* fix: fix

* fix: simpler check

---------

Co-authored-by: Ruslan Gainutdinov &lt;ruslan@datacrunch.io&gt;
diff --git a/tests/integration_tests/test_instances.py b/tests/integration_tests/test_instances.py
@@ -1,4 +1,6 @@
+import logging
 import os
+import time
 
 import pytest
 
@@ -7,6 +9,9 @@
 
 IN_GITHUB_ACTIONS = os.getenv('GITHUB_ACTIONS') == 'true'
 
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger()
+
 
 @pytest.mark.skipif(IN_GITHUB_ACTIONS, reason="Test doesn't work in Github Actions.")
 @pytest.mark.withoutresponses
@@ -19,20 +24,37 @@ def test_create_instance(self, verda_client: VerdaClient):
         instance = verda_client.instances.create(
             hostname='test-instance',
             location=Locations.FIN_03,
-            instance_type='CPU.4V',
-            description='test instance',
-            image='ubuntu-18.04',
+            instance_type='CPU.4V.16G',
+            description='test cpu instance',
+            image='ubuntu-22.04',
             ssh_key_ids=[ssh_key.id],
+            os_volume={'name': 'test-os-volume-cpu', 'size': 55},
         )
 
         # assert instance is created
         assert instance.id is not None
         assert instance.status == verda_client.constants.instance_status.PROVISIONING
 
-        # delete instance
-        verda_client.instances.action(instance.id, 'delete')
+        while instance.status != verda_client.constants.instance_status.RUNNING:
+            time.sleep(2)
+            logger.debug('Waiting for instance to be running... %s', instance.status)
+            instance = verda_client.instances.get_by_id(instance.id)
+
+        logger.debug('Instance is running... %s', instance.status)
+        logger.debug('Instance ID: %s', instance.id)
+        logger.debug('Instance OS Volume ID: %s', instance.os_volume_id)
+        logger.debug('Instance IP: %s', instance.ip)
+
+        # assert os volume is created
+        assert instance.os_volume_id is not None
 
-        # permanently delete all volumes in trash
-        trash = verda_client.volumes.get_in_trash()
-        for volume in trash:
-            verda_client.volumes.delete(volume.id, is_permanent=True)
+        # get os volume
+        os_volume = verda_client.volumes.get_by_id(instance.os_volume_id)
+        assert os_volume.id is not None
+        assert os_volume.name == 'test-os-volume-cpu'
+        assert os_volume.size == 55
+
+        # delete instance
+        verda_client.instances.action(
+            instance.id, 'delete', volume_ids=[instance.os_volume_id], delete_permanently=True
+        )
diff --git a/tests/integration_tests/test_spot_instances.py b/tests/integration_tests/test_spot_instances.py
@@ -0,0 +1,59 @@
+import logging
+import os
+import time
+
+import pytest
+
+from verda import VerdaClient
+from verda.constants import Locations
+from verda.instances import OSVolume
+
+IN_GITHUB_ACTIONS = os.getenv('GITHUB_ACTIONS') == 'true'
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger()
+
+
+@pytest.mark.skipif(IN_GITHUB_ACTIONS, reason="Test doesn't work in Github Actions.")
+@pytest.mark.withoutresponses
+class TestInstances:
+    def test_create_spot(self, verda_client: VerdaClient):
+        # get ssh key
+        ssh_key = verda_client.ssh_keys.get()[0]
+
+        # create instance
+        instance = verda_client.instances.create(
+            hostname='test-instance',
+            location=Locations.FIN_03,
+            instance_type='CPU.4V.16G',
+            description='test cpu instance',
+            image='ubuntu-22.04',
+            is_spot=True,
+            ssh_key_ids=[ssh_key.id],
+            os_volume=OSVolume(
+                name='test-os-volume-spot', size=56, on_spot_discontinue='delete_permanently'
+            ),
+        )
+
+        # assert instance is created
+        assert instance.id is not None
+        assert instance.status == verda_client.constants.instance_status.PROVISIONING
+
+        while instance.status != verda_client.constants.instance_status.RUNNING:
+            time.sleep(2)
+            logger.debug('Waiting for instance to be running... %s', instance.status)
+            instance = verda_client.instances.get_by_id(instance.id)
+
+        logger.debug('Instance is running... %s', instance.status)
+        logger.debug('Instance ID: %s', instance.id)
+        logger.debug('Instance OS Volume ID: %s', instance.os_volume_id)
+        logger.debug('Instance IP: %s', instance.ip)
+
+        # assert os volume is created
+        assert instance.os_volume_id is not None
+
+        # get os volume
+        os_volume = verda_client.volumes.get_by_id(instance.os_volume_id)
+        assert os_volume.id is not None
+        assert os_volume.name == 'test-os-volume-spot'
+        assert os_volume.size == 56
diff --git a/tests/unit_tests/instances/test_instances.py b/tests/unit_tests/instances/test_instances.py
@@ -1,9 +1,11 @@
+import json
+
 import pytest
-import responses  # https://github.com/getsentry/responses
+import 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 +268,33 @@ 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_with_spot_volume_policy(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)
+
+        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=os_volume,
+        )
+
+        # assert
+        request_body = responses.calls[0].request.body.decode('utf-8')
+        body = json.loads(request_body)
+        assert body['os_volume']['name'] == os_volume.name
+        assert body['os_volume']['size'] == 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 +369,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
diff --git a/verda/instances/__init__.py b/verda/instances/__init__.py
@@ -1 +1,8 @@
-from ._instances import Contract, Instance, InstancesService, Pricing
+from ._instances import (
+    Contract,
+    Instance,
+    InstancesService,
+    OnSpotDiscontinue,
+    OSVolume,
+    Pricing,
+)
diff --git a/verda/instances/_instances.py b/verda/instances/_instances.py
@@ -3,14 +3,35 @@
 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 +144,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 | dict | None = None,
         is_spot: bool = False,
         contract: Contract | None = None,
         pricing: Pricing | None = None,
@@ -170,7 +191,7 @@ def create(
             'hostname': hostname,
             'description': description,
             'location_code': location,
-            'os_volume': os_volume,
+            'os_volume': os_volume.to_dict() if isinstance(os_volume, OSVolume) else os_volume,
             'volumes': volumes or [],
             'existing_volumes': existing_volumes or [],
             'is_spot': is_spot,
@@ -204,21 +225,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
