Introduce storage volume snapshots

doc/source/storage-pools.rst

@@ -13,18 +13,22 @@ Storage Pool objects
 object that is returned from `GET /1.0/storage-pools/<name>` and then the
 associated methods that are then available at the same endpoint.
 
-There are also :py:class:`~pylxd.models.storage_pool.StorageResource` and
-:py:class:`~pylxd.models.storage_pool.StorageVolume` objects that represent the
-storage resources endpoint for a pool at `GET
-/1.0/storage-pools/<pool>/resources` and a storage volume on a pool at `GET
-/1.0/storage-pools/<pool>/volumes/<type>/<name>`.  Note that these should be
-accessed from the storage pool object.  For example:
+There are also :py:class:`~pylxd.models.storage_pool.StorageResource`,
+:py:class:`~pylxd.models.storage_pool.StorageVolume` and
+:py:class:`~pylxd.models.storage_pool.StorageVolumeSnapshot` objects that
+represent, respectively, the storage resources endpoint for a pool at
+`GET /1.0/storage-pools/<pool>/resources`, a storage volume on a pool at
+`GET /1.0/storage-pools/<pool>/volumes/<type>/<name>` and a custom volume snapshot
+at `GET /1.0/storage-pools/<pool>/volumes/<type>/<volume>/snapshots/<snapshot>`.
+Note that these should be accessed from the storage pool object.  For example:
 
 .. code:: python
 
     client = pylxd.Client()
     storage_pool = client.storage_pools.get('poolname')
+    resources = storage_pool.resources.get()
     storage_volume = storage_pool.volumes.get('custom', 'volumename')
+    snapshot = storage_volume.snapshots.get('snap0')
 
 
 .. note:: For more details of the LXD documentation concerning storage pools
@@ -137,10 +141,47 @@ following methods are available:
     changes to the LXD server.
   - `delete` - delete a storage volume object.  Note that the object is,
     therefore, stale after this action.
+  - `restore_from` - Restore the volume from a snapshot using the snapshot name.
 
 .. note:: `raw_put` and `raw_patch` are availble (but not documented) to allow
         putting and patching without syncing the object back.
 
+Storage Volume Snapshots
+------------------------
+
+Storage Volume Snapshots are represented as `StorageVolumeSnapshot` objects and
+stored in `StorageVolume` objects and represent snapshots of custom storage volumes.
+On the `pylxd` API they are accessed from a storage volume object that, in turn,
+is accessed from a storage pool object:
+
+.. code:: Python
+
+    storage_pool = client.storage_pools.get('pool1')
+    volumes = storage_pool.volumes.all()
+    custom_volume = storage_pool.volumes.get('custom', 'vol1')
+    a_snapshot = custom_volume.snapshots.get('snap0')
+
+Methods available on `<storage_volume_object>.snapshots`
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The following methods are accessed from the `snapshots` attribute on the `StorageVolume` object:
+
+  - `all` - Get all the snapshots from the storage volume.
+  - `get` - Get a single snapshot using its name.
+  - `create` - Take a snapshot on the current stage of the storage volume. The new snapshot's
+  name and expiration date can be set, default name is in the format "snapX".
+  - `exists` - Returns True if a storage volume snapshot with the given name exists, returns False otherwise.
+
+Methods available on the storage snapshot object
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Once in possession of a `StorageVolumeSnapshot` object from the `pylxd` API via `volume.snapshots.get()`,
+the following methods are available:
+
+  - `restore` - Restore the volume from the snapshot.
+  - `delete` - Delete the snapshot. Note that the object is, therefore, stale after this action.
+  - `rename` - Renames the snapshot. The endpoints that reference this snapshot will change accordingly.
+
 .. links
 
 .. _LXD Storage Pools: https://documentation.ubuntu.com/lxd/en/latest/storage/

integration/test_storage.py

@@ -12,8 +12,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import random
-import string
 import unittest
 
 import pylxd.exceptions as exceptions
@@ -25,26 +23,7 @@ def setUp(self):
         super().setUp()
 
         if not self.client.has_api_extension("storage"):
-            self.skipTest("Required LXD API extension not available!")
-
-    def create_storage_pool(self):
-        # create a storage pool in the form of 'xxx1' as a dir.
-        name = "".join(random.sample(string.ascii_lowercase, 3)) + "1"
-        self.lxd.storage_pools.post(
-            json={
-                "config": {},
-                "driver": "dir",
-                "name": name,
-            }
-        )
-        return name
-
-    def delete_storage_pool(self, name):
-        # delete the named storage pool
-        try:
-            self.lxd.storage_pools[name].delete()
-        except exceptions.NotFound:
-            pass
+            self.skipTest("Required 'storage' LXD API extension not available!")
 
 
 class TestStoragePools(StorageTestCase):
@@ -126,36 +105,12 @@ def test_get(self):
 class TestStorageVolume(StorageTestCase):
     """Tests for :py:class:`pylxd.models.storage_pools.StorageVolume"""
 
-    # note create and delete are tested in every method
-
-    def create_storage_volume(self, pool):
-        # note 'pool' needs to be storage_pool object or a string
-        if isinstance(pool, str):
-            pool = self.client.storage_pools.get(pool)
-        vol_input = {
-            "config": {},
-            "type": "custom",
-            # "pool": name,
-            "name": "vol1",
-        }
-        volume = pool.volumes.create(vol_input)
-        return volume
-
-    def delete_storage_volume(self, pool, volume):
-        # pool is either string or storage_pool object
-        # volume is either a string of storage_pool object
-        if isinstance(volume, str):
-            if isinstance(pool, str):
-                pool = self.client.storage_pools.get(pool)
-            volume = pool.volumes.get("custom", volume)
-        volume.delete()
-
     def test_create_and_get_and_delete(self):
         pool_name = self.create_storage_pool()
         self.addCleanup(self.delete_storage_pool, pool_name)
-
         storage_pool = self.client.storage_pools.get(pool_name)
-        volume = self.create_storage_volume(storage_pool)
+
+        volume = self.create_storage_volume(pool_name, "vol1")
         vol_copy = storage_pool.volumes.get("custom", "vol1")
         self.assertEqual(vol_copy.name, volume.name)
         volume.delete()
@@ -173,3 +128,80 @@ def test_patch(self):
         # as we're not using ZFS (and can't in these integration tests) we
         # can't really patch anything on a dir volume.
         pass
+
+
+class TestStorageVolumeSnapshot(StorageTestCase):
+    """Tests for :py:class:`pylxd.models.storage_pool.StorageVolumeSnapshot"""
+
+    def setUp(self):
+        super().setUp()
+
+        if not self.client.has_api_extension("storage_api_volume_snapshots"):
+            self.skipTest(
+                "Required 'storage_api_volume_snapshots' LXD API extension not available!"
+            )
+
+    def test_create_get_restore_delete_volume_snapshot(self):
+        # Create pool and volume
+        pool = self.create_storage_pool()
+        self.addCleanup(self.delete_storage_pool, pool)
+
+        volume = self.create_storage_volume(pool, "vol1")
+        self.addCleanup(self.delete_storage_volume, pool, "vol1")
+
+        # Create a few snapshots
+        first_snapshot = volume.snapshots.create()
+        self.assertEqual(first_snapshot.name, "snap0")
+
+        second_snapshot = volume.snapshots.create()
+        self.assertEqual(second_snapshot.name, "snap1")
+
+        # Try restoring the volume from one of the snapshots
+        first_snapshot.restore()
+
+        # Create new snapshot with defined name and expiration date
+        custom_snapshot_name = "custom-snapshot"
+        custom_snapshot_expiry_date = "2183-06-16T00:00:00Z"
+
+        custom_snapshot = volume.snapshots.create(
+            name=custom_snapshot_name, expires_at=custom_snapshot_expiry_date
+        )
+        self.assertEqual(custom_snapshot.name, custom_snapshot_name)
+        self.assertEqual(custom_snapshot.expires_at, custom_snapshot_expiry_date)
+
+        # Get all snapshots from the volume
+        all_snapshots = volume.snapshots.all()
+        self.assertEqual(len(all_snapshots), 3)
+
+        for snapshot_name in ["snap0", "snap1", custom_snapshot_name]:
+            self.assertIn(snapshot_name, all_snapshots)
+
+        # Delete a snapshot
+        second_snapshot.delete()
+
+        self.assertFalse(volume.snapshots.exists(second_snapshot.name))
+
+        self.assertRaises(exceptions.NotFound, volume.snapshots.get, "snap1")
+
+        all_snapshots = volume.snapshots.all()
+        self.assertEqual(len(all_snapshots), 2)
+
+        for snapshot_name in ["snap0", custom_snapshot_name]:
+            self.assertIn(snapshot_name, all_snapshots)
+
+        self.assertFalse("snap1" in all_snapshots)
+
+        # Test getting all snapshots with recursion
+        all_snapshots = volume.snapshots.all(use_recursion=True)
+        self.assertIn(first_snapshot, all_snapshots)
+        self.assertIn(custom_snapshot, all_snapshots)
+
+        # Change snapshot values
+        first_snapshot.rename("first")
+        self.assertFalse(volume.snapshots.exists("snap0"))
+        self.assertRaises(exceptions.NotFound, volume.snapshots.get, "snap0")
+
+        new_description = "first snapshot"
+        first_snapshot.description = new_description
+        first_snapshot.save(wait=True)
+        self.assertEqual(volume.snapshots.get("first").description, new_description)

integration/testing.py

@@ -157,6 +157,42 @@ def delete_network(self, name):
         except exceptions.NotFound:
             pass
 
+    def create_storage_pool(self):
+        # create a storage pool in the form of 'xxx1' as a dir.
+        name = "".join(random.sample(string.ascii_lowercase, 3)) + "1"
+        self.lxd.storage_pools.post(
+            json={
+                "config": {},
+                "driver": "dir",
+                "name": name,
+            }
+        )
+        return name
+
+    def delete_storage_pool(self, name):
+        # delete the named storage pool
+        try:
+            self.lxd.storage_pools[name].delete()
+        except exceptions.NotFound:
+            pass
+
+    def create_storage_volume(self, pool_name, volume_name):
+        pool = self.client.storage_pools.get(pool_name)
+        vol_json = {
+            "config": {},
+            "type": "custom",
+            "name": volume_name,
+        }
+        return pool.volumes.create(vol_json)
+
+    def delete_storage_volume(self, pool_name, volume_name):
+        try:
+            pool = self.client.storage_pools.get(pool_name)
+            pool.volumes.get("custom", volume_name).delete()
+            return True
+        except exceptions.NotFound:
+            return False
+
     def assertCommon(self, response):
         """Assert common LXD responses.
 

pylxd/managers.py

@@ -73,6 +73,18 @@ class StoragePoolManager(BaseManager):
     manager_for = "pylxd.models.StoragePool"
 
 
+class StorageResourcesManager(BaseManager):
+    manager_for = "pylxd.models.StorageResources"
+
+
+class StorageVolumeManager(BaseManager):
+    manager_for = "pylxd.models.StorageVolume"
+
+
+class StorageVolumeSnapshotManager(BaseManager):
+    manager_for = "pylxd.models.StorageVolumeSnapshot"
+
+
 class ClusterMemberManager(BaseManager):
     manager_for = "pylxd.models.ClusterMember"
 

pylxd/models/__init__.py

@@ -7,7 +7,12 @@
 from pylxd.models.operation import Operation
 from pylxd.models.profile import Profile
 from pylxd.models.project import Project
-from pylxd.models.storage_pool import StoragePool, StorageResources, StorageVolume
+from pylxd.models.storage_pool import (
+    StoragePool,
+    StorageResources,
+    StorageVolume,
+    StorageVolumeSnapshot,
+)
 from pylxd.models.virtual_machine import VirtualMachine
 
 __all__ = [
@@ -27,5 +32,6 @@
     "StoragePool",
     "StorageResources",
     "StorageVolume",
+    "StorageVolumeSnapshot",
     "VirtualMachine",
 ]

pylxd/models/_model.py

@@ -173,6 +173,21 @@ def __iter__(self):
         for attr in self.__attributes__.keys():
             yield attr, getattr(self, attr)
 
+    def __eq__(self, other):
+        if other.__class__ != self.__class__:
+            return False
+
+        for attr in self.__attributes__.keys():
+            if not hasattr(self, attr) and not hasattr(other, attr):
+                continue
+            try:
+                if self.__getattribute__(attr) != other.__getattribute__(attr):
+                    return False
+            except AttributeError:
+                return False
+
+        return True
+
     @property
     def dirty(self):
         return len(self.__dirty__) > 0
@@ -249,6 +264,20 @@ def marshall(self, skip_readonly=True):
                     marshalled[key] = val
         return marshalled
 
+    def post(self, json=None, wait=False):
+        """Access the POST method directly for the object.
+
+        :param wait: If wait is True, then wait here until the operation
+            completes.
+        :type wait: bool
+        :param json: Dictionary that the represents the request body used on the POST method.
+        :type wait: dict
+        :raises: :class:`pylxd.exception.LXDAPIException` on error
+        """
+        response = self.api.post(json=json)
+        if response.json()["type"] == "async" and wait:
+            self.client.operations.wait_for_operation(response.json()["operation"])
+
     def put(self, put_object, wait=False):
         """Access the PUT method directly for the object.