Add snapshot manage unmanage support

Add support for volume snapshot manage and volume snapshot unmanage
operations.

Change-Id: I25710af7b7a85a07e133aa3df94d2d751e253e84

doc/source/user/proxies/block_storage_v3.rst

@@ -118,7 +118,7 @@ Snapshot Operations
   :members: create_snapshot, delete_snapshot, update_snapshot, get_snapshot,
             find_snapshot, snapshots, get_snapshot_metadata,
             set_snapshot_metadata, delete_snapshot_metadata, reset_snapshot,
-            set_snapshot_status
+            set_snapshot_status, manage_snapshot, unmanage_snapshot
 
 Stats Operations
 ^^^^^^^^^^^^^^^^

openstack/block_storage/v3/_proxy.py

@@ -253,6 +253,30 @@ class Proxy(_base_proxy.BaseBlockStorageProxy):
         snapshot = self._get_resource(_snapshot.Snapshot, snapshot)
         snapshot.set_status(self, status, progress)
 
+    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.v3.snapshot.Snapshot`,
+            comprised of the properties on the Snapshot class.
+
+        :returns: The results of snapshot creation
+        :rtype: :class:`~openstack.block_storage.v3.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.v3.snapshot.Snapshot`.
+
+        :returns: None
+        """
+        snapshot_obj = self._get_resource(_snapshot.Snapshot, snapshot)
+        snapshot_obj.unmanage(self)
+
     # ====== TYPES ======
     def get_type(self, type):
         """Get a single type

openstack/block_storage/v3/snapshot.py

@@ -65,6 +65,8 @@ class Snapshot(resource.Resource, metadata.MetadataMixin):
     #: The ID of the volume this snapshot was taken of.
     volume_id = resource.Body("volume_id")
 
+    _max_microversion = '3.8'
+
     def _action(self, session, body, microversion=None):
         """Preform backup actions given the message body."""
         url = utils.urljoin(self.base_path, self.id, 'action')
@@ -91,5 +93,39 @@ class Snapshot(resource.Resource, metadata.MetadataMixin):
             body['os-update_snapshot_status']['progress'] = progress
         self._action(session, body)
 
+    @classmethod
+    def manage(
+        cls,
+        session,
+        volume_id,
+        ref,
+        name=None,
+        description=None,
+        metadata=None,
+    ):
+        """Manage a snapshot under block storage provisioning."""
+        url = '/manageable_snapshots'
+        if not utils.supports_microversion(session, '3.8'):
+            url = '/os-snapshot-manage'
+        body = {
+            'snapshot': {
+                'volume_id': volume_id,
+                'ref': ref,
+                'name': name,
+                'description': description,
+                'metadata': metadata,
+            }
+        }
+        resp = session.post(url, json=body, microversion=cls._max_microversion)
+        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/v3/test_proxy.py

@@ -785,6 +785,24 @@ 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):

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

@@ -9,6 +9,7 @@
 # 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 +19,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",
@@ -25,7 +27,7 @@ SNAPSHOT = {
     "created_at": "2015-03-09T12:14:57.233772",
     "updated_at": None,
     "metadata": {},
-    "volume_id": "5aa119a8-d25b-45a7-8d1b-88e127885635",
+    "volume_id": FAKE_VOLUME_ID,
     "size": 1,
     "id": FAKE_ID,
     "name": "snap-001",
@@ -130,3 +132,82 @@ class TestSnapshotActions(base.TestCase):
         self.sess.post.assert_called_with(
             url, json=body, microversion=sot._max_microversion
         )
+
+    @mock.patch(
+        'openstack.utils.supports_microversion',
+        autospec=True,
+        return_value=True,
+    )
+    def test_manage(self, mock_mv):
+        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 = '/manageable_snapshots'
+        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, microversion=sot._max_microversion
+        )
+
+    @mock.patch(
+        'openstack.utils.supports_microversion',
+        autospec=True,
+        return_value=False,
+    )
+    def test_manage_pre_38(self, mock_mv):
+        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, microversion=sot._max_microversion
+        )
+
+    def test_unmanage(self):
+        sot = snapshot.Snapshot(**SNAPSHOT)
+
+        self.assertIsNone(sot.unmanage(self.sess))
+
+        url = 'snapshots/%s/action' % FAKE_ID
+        body = {'os-unmanage': None}
+        self.sess.post.assert_called_with(
+            url, json=body, microversion=sot._max_microversion
+        )

releasenotes/notes/add-volume-snapshot-manage-unmanage-support-fc0be2a3fb4427d1.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Added support for volume snapshot manage and volume
+    snapshot unmanage.