openstack/python-cinderclient
Code Issues Proposed changes

Merge "Fix backwards compat for volume transfer < 3.55"

Browse Source
This commit is contained in:
Zuul
2018-10-02 17:33:26 +00:00
committed by Gerrit Code Review
parent ef95b503f2 b3f3cafd1e
commit e9658c6ef6
3 changed files with 131 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cinderclient/tests/unit/v3/test_shell.py
View File

@@ -1338,7 +1338,7 @@ class ShellTest(utils.TestCase):
expected = {'transfer': {'volume_id': 1234, expected = {'transfer': {'volume_id': 1234,
'name': None, 'name': None,
}} }}
self.assert_called('POST', '/volume-transfers', body=expected) self.assert_called('POST', '/os-volume-transfer', body=expected)
def test_create_transfer_no_snaps(self): def test_create_transfer_no_snaps(self):
self.run_command('--os-volume-api-version 3.55 transfer-create ' self.run_command('--os-volume-api-version 3.55 transfer-create '

96
cinderclient/tests/unit/v3/test_volume_transfers.py
View File

@@ -17,55 +17,93 @@ from cinderclient import api_versions
from cinderclient.tests.unit import utils from cinderclient.tests.unit import utils
from cinderclient.tests.unit.v3 import fakes from cinderclient.tests.unit.v3 import fakes
TRANSFER_URL = 'os-volume-transfer'
TRANSFER_355_URL = 'volume-transfers'
# Create calls need the right version of faked client # Create calls need the right version of faked client
v3cs = fakes.FakeClient(api_versions.APIVersion('3.55')) v355cs = fakes.FakeClient(api_versions.APIVersion('3.55'))
# Other calls fall back to default behavior # Other calls fall back to API extension behavior
cs = fakes.FakeClient() v3cs = fakes.FakeClient(api_versions.APIVersion('3.0'))
class VolumeTransfersTest(utils.TestCase): class VolumeTransfersTest(utils.TestCase):
def test_create(self): def test_create(self):
vol = v3cs.transfers.create('1234') vol = v3cs.transfers.create('1234')
v3cs.assert_called('POST', '/volume-transfers', v3cs.assert_called('POST', '/%s' % TRANSFER_URL,
body={'transfer': {'volume_id': '1234', 'name': None, body={'transfer': {'volume_id': '1234',
'no_snapshots': False}}) 'name': None}})
self._assert_request_id(vol)
def test_create_355(self):
vol = v355cs.transfers.create('1234')
v355cs.assert_called('POST', '/%s' % TRANSFER_355_URL,
body={'transfer': {'volume_id': '1234',
'name': None,
'no_snapshots': False}})
self._assert_request_id(vol) self._assert_request_id(vol)
def test_create_without_snapshots(self): def test_create_without_snapshots(self):
vol = v3cs.transfers.create('1234', no_snapshots=True) vol = v355cs.transfers.create('1234', no_snapshots=True)
v3cs.assert_called('POST', '/volume-transfers', v355cs.assert_called('POST', '/%s' % TRANSFER_355_URL,
body={'transfer': {'volume_id': '1234', 'name': None, body={'transfer': {'volume_id': '1234',
'no_snapshots': True}}) 'name': None,
'no_snapshots': True}})
self._assert_request_id(vol)
def _test_get(self, client, expected_url):
transfer_id = '5678'
vol = client.transfers.get(transfer_id)
client.assert_called('GET', '/%s/%s' % (expected_url, transfer_id))
self._assert_request_id(vol) self._assert_request_id(vol)
def test_get(self): def test_get(self):
transfer_id = '5678' self._test_get(v3cs, TRANSFER_URL)
vol = cs.transfers.get(transfer_id)
cs.assert_called('GET', '/os-volume-transfer/%s' % transfer_id)
self._assert_request_id(vol)
def test_list(self): def test_get_355(self):
lst = cs.transfers.list() self._test_get(v355cs, TRANSFER_355_URL)
cs.assert_called('GET', '/os-volume-transfer/detail')
def _test_list(self, client, expected_url):
lst = client.transfers.list()
client.assert_called('GET', '/%s/detail' % expected_url)
self._assert_request_id(lst) self._assert_request_id(lst)
def test_delete(self): def test_list(self):
b = cs.transfers.list()[0] self._test_list(v3cs, TRANSFER_URL)
def test_list_355(self):
self._test_list(v355cs, TRANSFER_355_URL)
def _test_delete(self, client, expected_url):
url = '/%s/5678' % expected_url
b = client.transfers.list()[0]
vol = b.delete() vol = b.delete()
cs.assert_called('DELETE', '/os-volume-transfer/5678') client.assert_called('DELETE', url)
self._assert_request_id(vol) self._assert_request_id(vol)
vol = cs.transfers.delete('5678') vol = client.transfers.delete('5678')
self._assert_request_id(vol) self._assert_request_id(vol)
cs.assert_called('DELETE', '/os-volume-transfer/5678') client.assert_called('DELETE', url)
vol = cs.transfers.delete(b) vol = client.transfers.delete(b)
cs.assert_called('DELETE', '/os-volume-transfer/5678') client.assert_called('DELETE', url)
self._assert_request_id(vol)
def test_delete(self):
self._test_delete(v3cs, TRANSFER_URL)
def test_delete_355(self):
self._test_delete(v355cs, TRANSFER_355_URL)
def _test_accept(self, client, expected_url):
transfer_id = '5678'
auth_key = '12345'
vol = client.transfers.accept(transfer_id, auth_key)
client.assert_called(
'POST',
'/%s/%s/accept' % (expected_url, transfer_id))
self._assert_request_id(vol) self._assert_request_id(vol)
def test_accept(self): def test_accept(self):
transfer_id = '5678' self._test_accept(v3cs, TRANSFER_URL)
auth_key = '12345'
vol = cs.transfers.accept(transfer_id, auth_key) def test_accept_355(self):
cs.assert_called('POST', '/os-volume-transfer/%s/accept' % transfer_id) self._test_accept(v355cs, TRANSFER_355_URL)
self._assert_request_id(vol)

67
cinderclient/v3/volume_transfers.py
View File

@@ -13,10 +13,10 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
""" """Volume transfer interface (v3 extension)."""
Volume transfer interface (v3 extension).
"""
from cinderclient import base
from cinderclient import utils
from cinderclient.v2 import volume_transfers from cinderclient.v2 import volume_transfers
@@ -33,4 +33,63 @@ class VolumeTransferManager(volume_transfers.VolumeTransferManager):
'name': name}} 'name': name}}
if self.api_version.matches('3.55'): if self.api_version.matches('3.55'):
body['transfer']['no_snapshots'] = no_snapshots body['transfer']['no_snapshots'] = no_snapshots
return self._create('/volume-transfers', body, 'transfer') return self._create('/volume-transfers', body, 'transfer')
return self._create('/os-volume-transfer', body, 'transfer')
def accept(self, transfer_id, auth_key):
"""Accept a volume transfer.
:param transfer_id: The ID of the transfer to accept.
:param auth_key: The auth_key of the transfer.
:rtype: :class:`VolumeTransfer`
"""
body = {'accept': {'auth_key': auth_key}}
if self.api_version.matches('3.55'):
return self._create('/volume-transfers/%s/accept' % transfer_id,
body, 'transfer')
return self._create('/os-volume-transfer/%s/accept' % transfer_id,
body, 'transfer')
def get(self, transfer_id):
"""Show details of a volume transfer.
:param transfer_id: The ID of the volume transfer to display.
:rtype: :class:`VolumeTransfer`
"""
if self.api_version.matches('3.55'):
return self._get("/volume-transfers/%s" % transfer_id, "transfer")
return self._get("/os-volume-transfer/%s" % transfer_id, "transfer")
def list(self, detailed=True, search_opts=None):
"""Get a list of all volume transfer.
:param detailed: Get detailed object information.
:param search_opts: Filtering options.
:rtype: list of :class:`VolumeTransfer`
"""
query_string = utils.build_query_param(search_opts)
detail = ""
if detailed:
detail = "/detail"
if self.api_version.matches('3.55'):
return self._list("/volume-transfers%s%s" % (detail, query_string),
"transfers")
return self._list("/os-volume-transfer%s%s" % (detail, query_string),
"transfers")
def delete(self, transfer_id):
"""Delete a volume transfer.
:param transfer_id: The :class:`VolumeTransfer` to delete.
"""
if self.api_version.matches('3.55'):
return self._delete(
"/volume-transfers/%s" % base.getid(transfer_id))
return self._delete("/os-volume-transfer/%s" % base.getid(transfer_id))