Merge "volume: Add Snapshot.manage, unmanage to volume v2 API"

openstack/block_storage/v2/_proxy.py

@@ -222,6 +222,30 @@ class Proxy(proxy.Proxy):
         )
         return self.reset_snapshot_status(snapshot, status)
 
+    def manage_snapshot(self, **attrs):
+        """Creates a snapshot by using existing storage rather than
+        allocating new storage.
+
+        :param dict attrs: Keyword arguments which will be used to create
+            a :class:`~openstack.block_storage.v2.snapshot.Snapshot`,
+            comprised of the properties on the Snapshot class.
+
+        :returns: The results of snapshot creation
+        :rtype: :class:`~openstack.block_storage.v2.snapshot.Snapshot`
+        """
+        return _snapshot.Snapshot.manage(self, **attrs)
+
+    def unmanage_snapshot(self, snapshot):
+        """Unmanage a snapshot from block storage provisioning.
+
+        :param snapshot: Either the ID of a snapshot or a
+            :class:`~openstack.block_storage.v2.snapshot.Snapshot`.
+
+        :returns: None
+        """
+        snapshot_obj = self._get_resource(_snapshot.Snapshot, snapshot)
+        snapshot_obj.unmanage(self)
+
     # ========== Types ==========
 
     def get_type(self, type):

openstack/block_storage/v2/snapshot.py

@@ -54,12 +54,10 @@ class Snapshot(resource.Resource, metadata.MetadataMixin):
     #: The ID of the volume this snapshot was taken of.
     volume_id = resource.Body("volume_id")
 
-    def _action(self, session, body, microversion=None):
+    def _action(self, session, body):
         """Preform backup actions given the message body."""
         url = utils.urljoin(self.base_path, self.id, 'action')
-        resp = session.post(
-            url, json=body, microversion=self._max_microversion
-        )
+        resp = session.post(url, json=body)
         exceptions.raise_from_response(resp)
         return resp
 
@@ -76,5 +74,37 @@ class Snapshot(resource.Resource, metadata.MetadataMixin):
         )
         self.reset_status(session, status)
 
+    @classmethod
+    def manage(
+        cls,
+        session,
+        volume_id,
+        ref,
+        name=None,
+        description=None,
+        metadata=None,
+    ):
+        """Manage a snapshot under block storage provisioning."""
+        url = '/os-snapshot-manage'
+        body = {
+            'snapshot': {
+                'volume_id': volume_id,
+                'ref': ref,
+                'name': name,
+                'description': description,
+                'metadata': metadata,
+            }
+        }
+        resp = session.post(url, json=body)
+        exceptions.raise_from_response(resp)
+        snapshot = Snapshot()
+        snapshot._translate_response(resp)
+        return snapshot
+
+    def unmanage(self, session):
+        """Unmanage a snapshot from block storage provisioning."""
+        body = {'os-unmanage': None}
+        self._action(session, body)
+
 
 SnapshotDetail = Snapshot

openstack/tests/unit/block_storage/v2/test_proxy.py

@@ -415,6 +415,33 @@ class TestSnapshot(TestVolumeProxy):
             expected_args=[self.proxy, "new_status"],
         )
 
+    def test_snapshot_manage(self):
+        kwargs = {
+            "volume_id": "fake_id",
+            "remote_source": "fake_volume",
+            "snapshot_name": "fake_snap",
+            "description": "test_snap",
+            "property": {"k": "v"},
+        }
+        self._verify(
+            "openstack.block_storage.v2.snapshot.Snapshot.manage",
+            self.proxy.manage_snapshot,
+            method_kwargs=kwargs,
+            method_result=snapshot.Snapshot(id="fake_id"),
+            expected_args=[self.proxy],
+            expected_kwargs=kwargs,
+            expected_result=snapshot.Snapshot(id="fake_id"),
+        )
+
+    def test_snapshot_unmanage(self):
+        self._verify(
+            "openstack.block_storage.v2.snapshot.Snapshot.unmanage",
+            self.proxy.unmanage_snapshot,
+            method_args=["value"],
+            expected_args=[self.proxy],
+            expected_result=None,
+        )
+
     def test_get_snapshot_metadata(self):
         self._verify(
             "openstack.block_storage.v2.snapshot.Snapshot.fetch_metadata",

openstack/tests/unit/block_storage/v2/test_snapshot.py

@@ -9,6 +9,8 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
+
+import copy
 from unittest import mock
 
 from keystoneauth1 import adapter
@@ -18,6 +20,7 @@ from openstack.tests.unit import base
 
 
 FAKE_ID = "ffa9bc5e-1172-4021-acaf-cdcd78a9584d"
+FAKE_VOLUME_ID = "5aa119a8-d25b-45a7-8d1b-88e127885635"
 
 SNAPSHOT = {
     "status": "creating",
@@ -99,6 +102,40 @@ class TestSnapshotActions(base.TestCase):
 
         url = f'snapshots/{FAKE_ID}/action'
         body = {'os-reset_status': {'status': 'new_status'}}
-        self.sess.post.assert_called_with(
-            url, json=body, microversion=sot._max_microversion
+        self.sess.post.assert_called_with(url, json=body)
+
+    def test_manage(self):
+        resp = mock.Mock()
+        resp.body = {'snapshot': copy.deepcopy(SNAPSHOT)}
+        resp.json = mock.Mock(return_value=resp.body)
+        resp.headers = {}
+        resp.status_code = 202
+
+        self.sess.post = mock.Mock(return_value=resp)
+
+        sot = snapshot.Snapshot.manage(
+            self.sess, volume_id=FAKE_VOLUME_ID, ref=FAKE_ID
         )
+
+        self.assertIsNotNone(sot)
+
+        url = '/os-snapshot-manage'
+        body = {
+            'snapshot': {
+                'volume_id': FAKE_VOLUME_ID,
+                'ref': FAKE_ID,
+                'name': None,
+                'description': None,
+                'metadata': None,
+            }
+        }
+        self.sess.post.assert_called_with(url, json=body)
+
+    def test_unmanage(self):
+        sot = snapshot.Snapshot(**SNAPSHOT)
+
+        self.assertIsNone(sot.unmanage(self.sess))
+
+        url = f'snapshots/{FAKE_ID}/action'
+        body = {'os-unmanage': None}
+        self.sess.post.assert_called_with(url, json=body)

openstack/tests/unit/block_storage/v3/test_proxy.py

@@ -873,7 +873,7 @@ class TestSnapshot(TestVolumeProxy):
             expected_args=[self.proxy, "new_status"],
         )
 
-    def test_set_status(self):
+    def test_snapshot_set_status(self):
         self._verify(
             "openstack.block_storage.v3.snapshot.Snapshot.set_status",
             self.proxy.set_snapshot_status,
@@ -881,7 +881,7 @@ class TestSnapshot(TestVolumeProxy):
             expected_args=[self.proxy, "new_status", None],
         )
 
-    def test_set_status_percentage(self):
+    def test_snapshot_set_status_percentage(self):
         self._verify(
             "openstack.block_storage.v3.snapshot.Snapshot.set_status",
             self.proxy.set_snapshot_status,
@@ -889,6 +889,33 @@ class TestSnapshot(TestVolumeProxy):
             expected_args=[self.proxy, "new_status", "per"],
         )
 
+    def test_snapshot_manage(self):
+        kwargs = {
+            "volume_id": "fake_id",
+            "remote_source": "fake_volume",
+            "snapshot_name": "fake_snap",
+            "description": "test_snap",
+            "property": {"k": "v"},
+        }
+        self._verify(
+            "openstack.block_storage.v3.snapshot.Snapshot.manage",
+            self.proxy.manage_snapshot,
+            method_kwargs=kwargs,
+            method_result=snapshot.Snapshot(id="fake_id"),
+            expected_args=[self.proxy],
+            expected_kwargs=kwargs,
+            expected_result=snapshot.Snapshot(id="fake_id"),
+        )
+
+    def test_snapshot_unmanage(self):
+        self._verify(
+            "openstack.block_storage.v3.snapshot.Snapshot.unmanage",
+            self.proxy.unmanage_snapshot,
+            method_args=["value"],
+            expected_args=[self.proxy],
+            expected_result=None,
+        )
+
     def test_get_snapshot_metadata(self):
         self._verify(
             "openstack.block_storage.v3.snapshot.Snapshot.fetch_metadata",
@@ -922,24 +949,6 @@ class TestSnapshot(TestVolumeProxy):
             expected_args=[self.proxy, "key"],
         )
 
-    def test_manage_snapshot(self):
-        kwargs = {
-            "volume_id": "fake_id",
-            "remote_source": "fake_volume",
-            "snapshot_name": "fake_snap",
-            "description": "test_snap",
-            "property": {"k": "v"},
-        }
-        self._verify(
-            "openstack.block_storage.v3.snapshot.Snapshot.manage",
-            self.proxy.manage_snapshot,
-            method_kwargs=kwargs,
-            method_result=snapshot.Snapshot(id="fake_id"),
-            expected_args=[self.proxy],
-            expected_kwargs=kwargs,
-            expected_result=snapshot.Snapshot(id="fake_id"),
-        )
-
 
 class TestType(TestVolumeProxy):
     def test_type_get(self):