Added support for Block Storage Encryption (#453)

* Implemented changes for Linode Disk Encryption

* Added more test cases

* Added LA note

Author: ezilber-akamai

linode_api4/groups/volume.py

@@ -45,7 +45,9 @@ def create(self, label, region=None, linode=None, size=20, **kwargs):
                      tags included do not exist, they will be created as part of
                      this operation.
         :type tags: list[str]
-
+        :param encryption: Whether the new Volume should opt in or out of disk encryption.
+        :type encryption: str
+            Note: Block Storage Disk Encryption is not currently available to all users.
         :returns: The new Volume.
         :rtype: Volume
         """

linode_api4/objects/linode.py

@@ -679,6 +679,7 @@ class Instance(Base):
         "has_user_data": Property(),
         "disk_encryption": Property(),
         "lke_cluster_id": Property(),
+        "capabilities": Property(unordered=True),
     }
 
     @property

linode_api4/objects/volume.py

@@ -46,6 +46,7 @@ class Volume(Base):
         "filesystem_path": Property(),
         "hardware_type": Property(),
         "linode_label": Property(),
+        "encryption": Property(),
     }
 
     def attach(self, to_linode, config=None):

test/fixtures/volumes.json

@@ -41,9 +41,24 @@
       "filesystem_path": "this/is/a/file/path",
       "hardware_type": "nvme",
       "linode_label": "some_label"
+    },
+    {
+      "id": 4,
+      "label": "block4",
+      "created": "2017-08-04T03:00:00",
+      "region": "ap-west-1a",
+      "linode_id": null,
+      "size": 40,
+      "updated": "2017-08-04T04:00:00",
+      "status": "active",
+      "tags": ["something"],
+      "filesystem_path": "this/is/a/file/path",
+      "hardware_type": "hdd",
+      "linode_label": null,
+      "encryption": "enabled"
     }
   ],
-  "results": 3,
+  "results": 4,
   "pages": 1,
   "page": 1
 }

test/integration/conftest.py

@@ -295,6 +295,35 @@ def test_volume(test_linode_client):
                 raise e
 
 
+@pytest.fixture(scope="session")
+def test_volume_with_encryption(test_linode_client):
+    client = test_linode_client
+    timestamp = str(time.time_ns())
+    region = get_region(client, {"Block Storage Encryption"})
+    label = "TestSDK-" + timestamp
+
+    volume = client.volume_create(
+        label=label, region=region, encryption="enabled"
+    )
+
+    yield volume
+
+    timeout = 100  # give 100s for volume to be detached before deletion
+
+    start_time = time.time()
+
+    while time.time() - start_time < timeout:
+        try:
+            res = volume.delete()
+            if res:
+                break
+            else:
+                time.sleep(3)
+        except ApiError as e:
+            if time.time() - start_time > timeout:
+                raise e
+
+
 @pytest.fixture
 def test_tag(test_linode_client):
     client = test_linode_client

test/integration/models/linode/test_linode.py

@@ -123,6 +123,25 @@ def linode_for_disk_tests(test_linode_client, e2e_test_firewall):
     linode_instance.delete()
 
 
+@pytest.fixture
+def linode_with_block_storage_encryption(test_linode_client, e2e_test_firewall):
+    client = test_linode_client
+    chosen_region = get_region(client, {"Linodes", "Block Storage Encryption"})
+    label = get_test_label()
+
+    linode_instance, password = client.linode.instance_create(
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/alpine3.19",
+        label=label + "block-storage-encryption",
+        firewall=e2e_test_firewall,
+    )
+
+    yield linode_instance
+
+    linode_instance.delete()
+
+
 @pytest.fixture
 def create_linode_for_long_running_tests(test_linode_client, e2e_test_firewall):
     client = test_linode_client
@@ -440,6 +459,13 @@ def test_linode_with_disk_encryption_disabled(linode_with_disk_encryption):
     )
 
 
+def test_linode_with_block_storage_encryption(
+    linode_with_block_storage_encryption,
+):
+    linode = linode_with_block_storage_encryption
+    assert "Block Storage Encryption" in linode.capabilities
+
+
 def wait_for_disk_status(disk: Disk, timeout):
     start_time = time.time()
     while True:

test/integration/models/volume/test_volume.py

@@ -60,6 +60,15 @@ def test_get_volume(test_linode_client, test_volume):
     assert volume.id == test_volume.id
 
 
+def test_get_volume_with_encryption(
+    test_linode_client, test_volume_with_encryption
+):
+    volume = test_linode_client.load(Volume, test_volume_with_encryption.id)
+
+    assert volume.id == test_volume_with_encryption.id
+    assert volume.encryption == "enabled"
+
+
 def test_update_volume_tag(test_linode_client, test_volume):
     volume = test_volume
     tag_1 = "volume_test_tag1"

test/unit/linode_client_test.py

@@ -150,7 +150,7 @@ def test_image_create(self):
     def test_get_volumes(self):
         v = self.client.volumes()
 
-        self.assertEqual(len(v), 3)
+        self.assertEqual(len(v), 4)
         self.assertEqual(v[0].label, "block1")
         self.assertEqual(v[0].region.id, "us-east-1a")
         self.assertEqual(v[1].label, "block2")

test/unit/objects/volume_test.py

@@ -31,6 +31,10 @@ def test_get_volume(self):
         self.assertEqual(volume.hardware_type, "hdd")
         self.assertEqual(volume.linode_label, None)
 
+    def test_get_volume_with_encryption(self):
+        volume = Volume(self.client, 4)
+        self.assertEqual(volume.encryption, "enabled")
+
     def test_update_volume_tags(self):
         """
         Tests that updating tags on an entity send the correct request