diff --git a/cinder/tests/unit/backup/drivers/test_backup_nfs.py b/cinder/tests/unit/backup/drivers/test_backup_nfs.py index 3d6bb0d53f2..240f95b6b0f 100644 --- a/cinder/tests/unit/backup/drivers/test_backup_nfs.py +++ b/cinder/tests/unit/backup/drivers/test_backup_nfs.py @@ -671,7 +671,9 @@ class BackupNFSSwiftBasedTestCase(test.TestCase): fake_data = bytearray(size) if six.PY2: # On Python 2, zlib.compressor() accepts buffer, but not bytearray - fake_data = buffer(fake_data) + # NOTE(jsbryant): Pep8 fails on py3 based installations as buffer() + # was removed. 'noqa' used here to avoid that failure. + fake_data = buffer(fake_data) # noqa return fake_data def test_prepare_output_data_effective_compression(self): diff --git a/cinder/tests/unit/utils.py b/cinder/tests/unit/utils.py index b7560717868..34d27dfff7d 100644 --- a/cinder/tests/unit/utils.py +++ b/cinder/tests/unit/utils.py @@ -427,7 +427,9 @@ def get_file_spec(): file_spec = list(set(dir(_io.TextIOWrapper)).union( set(dir(_io.BytesIO)))) else: - file_spec = file + # NOTE(jsbryant): Pep8 on py3 based systems will fail because + # 'file' has been removed. Using noqa here to avoid the failure. + file_spec = file # noqa def generate_timeout_series(timeout): diff --git a/cinder/tests/unit/volume/drivers/nexenta/test_nexenta5_jsonrpc.py b/cinder/tests/unit/volume/drivers/nexenta/test_nexenta5_jsonrpc.py index 0fa6a29145b..0946297ccca 100644 --- a/cinder/tests/unit/volume/drivers/nexenta/test_nexenta5_jsonrpc.py +++ b/cinder/tests/unit/volume/drivers/nexenta/test_nexenta5_jsonrpc.py @@ -1,251 +1,251 @@ -# Copyright 2016 Nexenta Systems, Inc. -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -""" -Unit tests for NexentaStor 5 REST API helper -""" - -import uuid - -import mock -from mock import patch -from oslo_serialization import jsonutils -import requests -from requests import adapters - -from cinder import exception -from cinder import test -from cinder.volume.drivers.nexenta.ns5 import jsonrpc - -HOST = '1.1.1.1' -USERNAME = 'user' -PASSWORD = 'pass' - - -def gen_response(code=200, json=None): - r = requests.Response() - r.headers['Content-Type'] = 'application/json' - r.encoding = 'utf8' - r.status_code = code - r.reason = 'FAKE REASON' - r.raw = mock.Mock() - r._content = '' - if json: - r._content = jsonutils.dumps(json) - return r - - -class TestNexentaJSONProxyAuth(test.TestCase): - - @patch('cinder.volume.drivers.nexenta.ns5.jsonrpc.requests.post') - def test_https_auth(self, post): - use_https = True - port = 8443 - auth_uri = 'auth/login' - rnd_url = 'some/random/url' - - class PostSideEffect(object): - def __call__(self, *args, **kwargs): - r = gen_response() - if args[0] == '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { - 'scheme': 'https', - 'host': HOST, - 'port': port, - 'uri': auth_uri}: - token = uuid.uuid4().hex - content = {'token': token} - r._content = jsonutils.dumps(content) - return r - post_side_effect = PostSideEffect() - post.side_effect = post_side_effect - - class TestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(TestAdapter, self).__init__() - self.counter = 0 - - def send(self, request, *args, **kwargs): - # an url is being requested for the second time - if self.counter == 1: - # make the fake backend respond 401 - r = gen_response(401) - r._content = '' - r.connection = mock.Mock() - r_ = gen_response(json={'data': []}) - r.connection.send = lambda prep, **kwargs_: r_ - else: - r = gen_response(json={'data': []}) - r.request = request - self.counter += 1 - return r - - nef = jsonrpc.NexentaJSONProxy(HOST, port, USERNAME, PASSWORD, - use_https) - adapter = TestAdapter() - nef.session.mount( - '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { - 'scheme': 'https', - 'host': HOST, - 'port': port, - 'uri': rnd_url}, - adapter) - - # successful authorization - self.assertEqual({'data': []}, nef.get(rnd_url)) - - # session timeout simulation. Client must authenticate newly - self.assertEqual({'data': []}, nef.get(rnd_url)) - # auth URL must be requested two times at this moment - self.assertEqual(2, post.call_count) - - # continue with the last (second) token - self.assertEqual(nef.get(rnd_url), {'data': []}) - # auth URL must be requested two times - self.assertEqual(2, post.call_count) - - -class TestNexentaJSONProxy(test.TestCase): - - def setUp(self): - super(TestNexentaJSONProxy, self).setUp() - self.nef = jsonrpc.NexentaJSONProxy(HOST, 0, USERNAME, PASSWORD, False) - - def gen_adapter(self, code, json=None): - class TestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(TestAdapter, self).__init__() - - def send(self, request, *args, **kwargs): - r = gen_response(code, json) - r.request = request - return r - - return TestAdapter() - - def _mount_adapter(self, url, adapter): - self.nef.session.mount( - '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { - 'scheme': 'http', - 'host': HOST, - 'port': 8080, - 'uri': url}, - adapter) - - def test_post(self): - random_dict = {'data': uuid.uuid4().hex} - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) - self.assertEqual(random_dict, self.nef.post(rnd_url)) - - def test_delete(self): - random_dict = {'data': uuid.uuid4().hex} - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) - self.assertEqual(random_dict, self.nef.delete(rnd_url)) - - def test_put(self): - random_dict = {'data': uuid.uuid4().hex} - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) - self.assertEqual(random_dict, self.nef.put(rnd_url)) - - def test_get_200(self): - random_dict = {'data': uuid.uuid4().hex} - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, self.gen_adapter(200, random_dict)) - self.assertEqual(random_dict, self.nef.get(rnd_url)) - - def test_get_201(self): - random_dict = {'data': uuid.uuid4().hex} - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) - self.assertEqual(random_dict, self.nef.get(rnd_url)) - - def test_get_500(self): - class TestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(TestAdapter, self).__init__() - - def send(self, request, *args, **kwargs): - json = { - 'code': 'NEF_ERROR', - 'message': 'Some error' - } - r = gen_response(500, json) - r.request = request - return r - - adapter = TestAdapter() - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, adapter) - self.assertRaises(exception.NexentaException, self.nef.get, rnd_url) - - def test_get__not_nef_error(self): - class TestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(TestAdapter, self).__init__() - - def send(self, request, *args, **kwargs): - r = gen_response(404) - r._content = 'Page Not Found' - r.request = request - return r - - adapter = TestAdapter() - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, adapter) - self.assertRaises(exception.VolumeBackendAPIException, self.nef.get, - rnd_url) - - def test_get__not_nef_error_empty_body(self): - class TestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(TestAdapter, self).__init__() - - def send(self, request, *args, **kwargs): - r = gen_response(404) - r.request = request - return r - - adapter = TestAdapter() - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, adapter) - self.assertRaises(exception.VolumeBackendAPIException, self.nef.get, - rnd_url) - - def test_202(self): - redirect_url = 'redirect/url' - - class RedirectTestAdapter(adapters.HTTPAdapter): - - def __init__(self): - super(RedirectTestAdapter, self).__init__() - - def send(self, request, *args, **kwargs): - json = { - 'links': [{'href': redirect_url}] - } - r = gen_response(202, json) - r.request = request - return r - - rnd_url = 'some/random/url' - self._mount_adapter(rnd_url, RedirectTestAdapter()) - self._mount_adapter(redirect_url, self.gen_adapter(201)) - self.assertIsNone(self.nef.get(rnd_url)) +# Copyright 2016 Nexenta Systems, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Unit tests for NexentaStor 5 REST API helper +""" + +import uuid + +import mock +from mock import patch +from oslo_serialization import jsonutils +import requests +from requests import adapters + +from cinder import exception +from cinder import test +from cinder.volume.drivers.nexenta.ns5 import jsonrpc + +HOST = '1.1.1.1' +USERNAME = 'user' +PASSWORD = 'pass' + + +def gen_response(code=200, json=None): + r = requests.Response() + r.headers['Content-Type'] = 'application/json' + r.encoding = 'utf8' + r.status_code = code + r.reason = 'FAKE REASON' + r.raw = mock.Mock() + r._content = '' + if json: + r._content = jsonutils.dumps(json) + return r + + +class TestNexentaJSONProxyAuth(test.TestCase): + + @patch('cinder.volume.drivers.nexenta.ns5.jsonrpc.requests.post') + def test_https_auth(self, post): + use_https = True + port = 8443 + auth_uri = 'auth/login' + rnd_url = 'some/random/url' + + class PostSideEffect(object): + def __call__(self, *args, **kwargs): + r = gen_response() + if args[0] == '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { + 'scheme': 'https', + 'host': HOST, + 'port': port, + 'uri': auth_uri}: + token = uuid.uuid4().hex + content = {'token': token} + r._content = jsonutils.dumps(content) + return r + post_side_effect = PostSideEffect() + post.side_effect = post_side_effect + + class TestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(TestAdapter, self).__init__() + self.counter = 0 + + def send(self, request, *args, **kwargs): + # an url is being requested for the second time + if self.counter == 1: + # make the fake backend respond 401 + r = gen_response(401) + r._content = '' + r.connection = mock.Mock() + r_ = gen_response(json={'data': []}) + r.connection.send = lambda prep, **kwargs_: r_ + else: + r = gen_response(json={'data': []}) + r.request = request + self.counter += 1 + return r + + nef = jsonrpc.NexentaJSONProxy(HOST, port, USERNAME, PASSWORD, + use_https) + adapter = TestAdapter() + nef.session.mount( + '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { + 'scheme': 'https', + 'host': HOST, + 'port': port, + 'uri': rnd_url}, + adapter) + + # successful authorization + self.assertEqual({'data': []}, nef.get(rnd_url)) + + # session timeout simulation. Client must authenticate newly + self.assertEqual({'data': []}, nef.get(rnd_url)) + # auth URL must be requested two times at this moment + self.assertEqual(2, post.call_count) + + # continue with the last (second) token + self.assertEqual(nef.get(rnd_url), {'data': []}) + # auth URL must be requested two times + self.assertEqual(2, post.call_count) + + +class TestNexentaJSONProxy(test.TestCase): + + def setUp(self): + super(TestNexentaJSONProxy, self).setUp() + self.nef = jsonrpc.NexentaJSONProxy(HOST, 0, USERNAME, PASSWORD, False) + + def gen_adapter(self, code, json=None): + class TestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(TestAdapter, self).__init__() + + def send(self, request, *args, **kwargs): + r = gen_response(code, json) + r.request = request + return r + + return TestAdapter() + + def _mount_adapter(self, url, adapter): + self.nef.session.mount( + '%(scheme)s://%(host)s:%(port)s/%(uri)s' % { + 'scheme': 'http', + 'host': HOST, + 'port': 8080, + 'uri': url}, + adapter) + + def test_post(self): + random_dict = {'data': uuid.uuid4().hex} + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) + self.assertEqual(random_dict, self.nef.post(rnd_url)) + + def test_delete(self): + random_dict = {'data': uuid.uuid4().hex} + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) + self.assertEqual(random_dict, self.nef.delete(rnd_url)) + + def test_put(self): + random_dict = {'data': uuid.uuid4().hex} + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) + self.assertEqual(random_dict, self.nef.put(rnd_url)) + + def test_get_200(self): + random_dict = {'data': uuid.uuid4().hex} + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, self.gen_adapter(200, random_dict)) + self.assertEqual(random_dict, self.nef.get(rnd_url)) + + def test_get_201(self): + random_dict = {'data': uuid.uuid4().hex} + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, self.gen_adapter(201, random_dict)) + self.assertEqual(random_dict, self.nef.get(rnd_url)) + + def test_get_500(self): + class TestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(TestAdapter, self).__init__() + + def send(self, request, *args, **kwargs): + json = { + 'code': 'NEF_ERROR', + 'message': 'Some error' + } + r = gen_response(500, json) + r.request = request + return r + + adapter = TestAdapter() + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, adapter) + self.assertRaises(exception.NexentaException, self.nef.get, rnd_url) + + def test_get__not_nef_error(self): + class TestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(TestAdapter, self).__init__() + + def send(self, request, *args, **kwargs): + r = gen_response(404) + r._content = 'Page Not Found' + r.request = request + return r + + adapter = TestAdapter() + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, adapter) + self.assertRaises(exception.VolumeBackendAPIException, self.nef.get, + rnd_url) + + def test_get__not_nef_error_empty_body(self): + class TestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(TestAdapter, self).__init__() + + def send(self, request, *args, **kwargs): + r = gen_response(404) + r.request = request + return r + + adapter = TestAdapter() + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, adapter) + self.assertRaises(exception.VolumeBackendAPIException, self.nef.get, + rnd_url) + + def test_202(self): + redirect_url = 'redirect/url' + + class RedirectTestAdapter(adapters.HTTPAdapter): + + def __init__(self): + super(RedirectTestAdapter, self).__init__() + + def send(self, request, *args, **kwargs): + json = { + 'links': [{'href': redirect_url}] + } + r = gen_response(202, json) + r.request = request + return r + + rnd_url = 'some/random/url' + self._mount_adapter(rnd_url, RedirectTestAdapter()) + self._mount_adapter(redirect_url, self.gen_adapter(201)) + self.assertIsNone(self.nef.get(rnd_url)) diff --git a/cinder/volume/drivers/coprhd/helpers/commoncoprhdapi.py b/cinder/volume/drivers/coprhd/helpers/commoncoprhdapi.py index eaa8721b221..eafdb2e3c0d 100644 --- a/cinder/volume/drivers/coprhd/helpers/commoncoprhdapi.py +++ b/cinder/volume/drivers/coprhd/helpers/commoncoprhdapi.py @@ -49,7 +49,7 @@ URI_TASKS_BY_OPID = '/vdc/tasks/{0}' def _decode_list(data): rv = [] for item in data: - if isinstance(item, unicode): + if isinstance(item, six.text_type): item = item.encode('utf-8') elif isinstance(item, list): item = _decode_list(item) @@ -62,9 +62,9 @@ def _decode_list(data): def _decode_dict(data): rv = {} for key, value in data.items(): - if isinstance(key, unicode): + if isinstance(key, six.text_type): key = key.encode('utf-8') - if isinstance(value, unicode): + if isinstance(value, six.text_type): value = value.encode('utf-8') elif isinstance(value, list): value = _decode_list(value) @@ -204,7 +204,7 @@ def service_json_request(ip_addr, port, http_method, uri, body, " service your request") else: error_msg = response.text - if isinstance(error_msg, unicode): + if isinstance(error_msg, six.text_type): error_msg = error_msg.encode('utf-8') raise CoprHdError(CoprHdError.HTTP_ERR, (_("HTTP code: %(status_code)s"