diff --git a/openstack/block_storage/v2/_proxy.py b/openstack/block_storage/v2/_proxy.py index 529b50a5f..ead6a3e0d 100644 --- a/openstack/block_storage/v2/_proxy.py +++ b/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): diff --git a/openstack/block_storage/v2/snapshot.py b/openstack/block_storage/v2/snapshot.py index 4d726c651..6e3929187 100644 --- a/openstack/block_storage/v2/snapshot.py +++ b/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 diff --git a/openstack/tests/unit/block_storage/v2/test_proxy.py b/openstack/tests/unit/block_storage/v2/test_proxy.py index 649bd8475..37124fc4b 100644 --- a/openstack/tests/unit/block_storage/v2/test_proxy.py +++ b/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", diff --git a/openstack/tests/unit/block_storage/v2/test_snapshot.py b/openstack/tests/unit/block_storage/v2/test_snapshot.py index beca9660f..501317893 100644 --- a/openstack/tests/unit/block_storage/v2/test_snapshot.py +++ b/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) diff --git a/openstack/tests/unit/block_storage/v3/test_proxy.py b/openstack/tests/unit/block_storage/v3/test_proxy.py index e9ceb4808..cf17c3654 100644 --- a/openstack/tests/unit/block_storage/v3/test_proxy.py +++ b/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):