Remove mox/mox3 usage from test_cli20.py

Change-Id: Ib4dc5f6c7ba57d94364f19a049c690d3817c723a
Partial-Bug: #1753504
This commit is contained in:
Hongbin Lu
2018-03-22 21:52:29 +00:00
committed by Slawek Kaplonski
parent f5f810a624
commit f6b1e85899

View File

@@ -20,7 +20,6 @@ import json
import sys import sys
import mock import mock
from mox3 import mox
from oslo_utils import encodeutils from oslo_utils import encodeutils
from oslotest import base from oslotest import base
import requests import requests
@@ -93,12 +92,12 @@ def end_url(path, query=None):
return query and _url_str + "?" + query or _url_str return query and _url_str + "?" + query or _url_str
class MyUrlComparator(mox.Comparator): class MyUrlComparator(object):
def __init__(self, lhs, client): def __init__(self, lhs, client):
self.lhs = lhs self.lhs = lhs
self.client = client self.client = client
def equals(self, rhs): def __eq__(self, rhs):
lhsp = urlparse.urlparse(self.lhs) lhsp = urlparse.urlparse(self.lhs)
rhsp = urlparse.urlparse(rhs) rhsp = urlparse.urlparse(rhs)
@@ -118,7 +117,7 @@ class MyUrlComparator(mox.Comparator):
return str(self) return str(self)
class MyComparator(mox.Comparator): class MyComparator(object):
def __init__(self, lhs, client): def __init__(self, lhs, client):
self.lhs = lhs self.lhs = lhs
self.client = client self.client = client
@@ -159,7 +158,7 @@ class MyComparator(mox.Comparator):
return self._com_list(lhs, rhs) return self._com_list(lhs, rhs)
return lhs == rhs return lhs == rhs
def equals(self, rhs): def __eq__(self, rhs):
if self.client: if self.client:
rhs = self.client.deserialize(rhs, 200) rhs = self.client.deserialize(rhs, 200)
return self._com(self.lhs, rhs) return self._com(self.lhs, rhs)
@@ -228,7 +227,6 @@ class CLITestV20Base(base.BaseTestCase):
if plurals is not None: if plurals is not None:
client.Client.EXTED_PLURALS.update(plurals) client.Client.EXTED_PLURALS.update(plurals)
self.metadata = {'plurals': client.Client.EXTED_PLURALS} self.metadata = {'plurals': client.Client.EXTED_PLURALS}
self.mox = mox.Mox()
self.endurl = ENDURL self.endurl = ENDURL
self.fake_stdout = FakeStdout() self.fake_stdout = FakeStdout()
@@ -263,9 +261,6 @@ class CLITestV20Base(base.BaseTestCase):
parent_id=None, no_api_call=False, parent_id=None, no_api_call=False,
expected_exception=None, expected_exception=None,
**kwargs): **kwargs):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource: if not cmd_resource:
cmd_resource = resource cmd_resource = resource
if (resource in self.non_admin_status_resources): if (resource in self.non_admin_status_resources):
@@ -292,34 +287,37 @@ class CLITestV20Base(base.BaseTestCase):
path = getattr(self.client, resource_plural + "_path") path = getattr(self.client, resource_plural + "_path")
if parent_id: if parent_id:
path = path % parent_id path = path % parent_id
mox_body = MyComparator(body, self.client) mock_body = MyComparator(body, self.client)
if not no_api_call:
self.client.httpclient.request(
end_url(path), 'POST',
body=mox_body,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('create_' + resource) cmd_parser = cmd.get_parser('create_' + resource)
if expected_exception: resp = (MyResp(200), resstr)
self.assertRaises(expected_exception,
shell.run_command, cmd, cmd_parser, args) with mock.patch.object(cmd, "get_client",
else: return_value=self.client) as mock_get_client, \
shell.run_command(cmd, cmd_parser, args) mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
if expected_exception:
self.assertRaises(expected_exception,
shell.run_command, cmd, cmd_parser, args)
else:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
if not no_api_call:
mock_request.assert_called_once_with(
end_url(path), 'POST',
body=mock_body,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
if not expected_exception:
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)
if name: if name:
self.assertIn(name, _str) self.assertIn(name, _str)
self.mox.VerifyAll()
self.mox.UnsetStubs()
def _test_list_columns(self, cmd, resources, def _test_list_columns(self, cmd, resources,
resources_out, args=('-f', 'json'), resources_out, args=('-f', 'json'),
cmd_resources=None, parent_id=None): cmd_resources=None, parent_id=None):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resources: if not cmd_resources:
cmd_resources = resources cmd_resources = resources
@@ -328,25 +326,27 @@ class CLITestV20Base(base.BaseTestCase):
path = getattr(self.client, cmd_resources + "_path") path = getattr(self.client, cmd_resources + "_path")
if parent_id: if parent_id:
path = path % parent_id path = path % parent_id
self.client.httpclient.request( cmd_parser = cmd.get_parser("list_" + cmd_resources)
resp = (MyResp(200), resstr)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
end_url(path), 'GET', end_url(path), 'GET',
body=None, body=None,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + cmd_resources)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
def _test_list_resources(self, resources, cmd, detail=False, tags=(), def _test_list_resources(self, resources, cmd, detail=False, tags=(),
fields_1=(), fields_2=(), page_size=None, fields_1=(), fields_2=(), page_size=None,
sort_key=(), sort_dir=(), response_contents=None, sort_key=(), sort_dir=(), response_contents=None,
base_args=None, path=None, cmd_resources=None, base_args=None, path=None, cmd_resources=None,
parent_id=None, output_format=None, query=""): parent_id=None, output_format=None, query=""):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resources: if not cmd_resources:
cmd_resources = resources cmd_resources = resources
if response_contents is None: if response_contents is None:
@@ -422,18 +422,22 @@ class CLITestV20Base(base.BaseTestCase):
if output_format: if output_format:
args.append('-f') args.append('-f')
args.append(output_format) args.append(output_format)
self.client.httpclient.request( cmd_parser = cmd.get_parser("list_" + cmd_resources)
MyUrlComparator(end_url(path, query), resp = (MyResp(200), resstr)
self.client),
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
MyUrlComparator(end_url(path, query), self.client),
'GET', 'GET',
body=None, body=None,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + cmd_resources)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
if response_contents is None: if response_contents is None:
self.assertIn('myid1', _str) self.assertIn('myid1', _str)
@@ -443,9 +447,6 @@ class CLITestV20Base(base.BaseTestCase):
base_args=None, base_args=None,
cmd_resources=None, cmd_resources=None,
parent_id=None, query=""): parent_id=None, query=""):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resources: if not cmd_resources:
cmd_resources = resources cmd_resources = resources
@@ -461,29 +462,34 @@ class CLITestV20Base(base.BaseTestCase):
{'id': 'myid4', }]} {'id': 'myid4', }]}
resstr1 = self.client.serialize(reses1) resstr1 = self.client.serialize(reses1)
resstr2 = self.client.serialize(reses2) resstr2 = self.client.serialize(reses2)
self.client.httpclient.request(
end_url(path, query), 'GET',
body=None,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr1))
self.client.httpclient.request(
MyUrlComparator(end_url(path, fake_query),
self.client), 'GET',
body=None,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr2))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("list_" + cmd_resources) cmd_parser = cmd.get_parser("list_" + cmd_resources)
args = base_args if base_args is not None else [] args = base_args if base_args is not None else []
shell.run_command(cmd, cmd_parser, args) mock_request_calls = [
self.mox.VerifyAll() mock.call(
self.mox.UnsetStubs() end_url(path, query), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN})),
mock.call(
MyUrlComparator(end_url(path, fake_query),
self.client), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))]
mock_request_resp = [(MyResp(200), resstr1), (MyResp(200), resstr2)]
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient,
"request") as mock_request:
mock_request.side_effect = mock_request_resp
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
self.assertEqual(2, mock_request.call_count)
mock_request.assert_has_calls(mock_request_calls)
def _test_update_resource(self, resource, cmd, myid, args, extrafields, def _test_update_resource(self, resource, cmd, myid, args, extrafields,
cmd_resource=None, parent_id=None): cmd_resource=None, parent_id=None):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource: if not cmd_resource:
cmd_resource = resource cmd_resource = resource
@@ -493,27 +499,29 @@ class CLITestV20Base(base.BaseTestCase):
path = path % (parent_id, myid) path = path % (parent_id, myid)
else: else:
path = path % myid path = path % myid
mox_body = MyComparator(body, self.client) mock_body = MyComparator(body, self.client)
self.client.httpclient.request( cmd_parser = cmd.get_parser("update_" + cmd_resource)
resp = (MyResp(204), None)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
MyUrlComparator(end_url(path), self.client), MyUrlComparator(end_url(path), self.client),
'PUT', 'PUT',
body=mox_body, body=mock_body,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("update_" + cmd_resource)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)
def _test_show_resource(self, resource, cmd, myid, args, fields=(), def _test_show_resource(self, resource, cmd, myid, args, fields=(),
cmd_resource=None, parent_id=None): cmd_resource=None, parent_id=None):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource: if not cmd_resource:
cmd_resource = resource cmd_resource = resource
@@ -527,53 +535,68 @@ class CLITestV20Base(base.BaseTestCase):
path = path % (parent_id, myid) path = path % (parent_id, myid)
else: else:
path = path % myid path = path % myid
self.client.httpclient.request( cmd_parser = cmd.get_parser("show_" + cmd_resource)
resp = (MyResp(200), resstr)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
end_url(path, query), 'GET', end_url(path, query), 'GET',
body=None, body=None,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("show_" + cmd_resource)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)
self.assertIn('myname', _str) self.assertIn('myname', _str)
def _test_set_path_and_delete(self, path, parent_id, myid, def _test_set_path_and_delete(self, path, parent_id, myid,
mock_request_calls, mock_request_returns,
delete_fail=False): delete_fail=False):
return_val = 404 if delete_fail else 204 return_val = 404 if delete_fail else 204
if parent_id: if parent_id:
path = path % (parent_id, myid) path = path % (parent_id, myid)
else: else:
path = path % (myid) path = path % (myid)
self.client.httpclient.request( mock_request_returns.append((MyResp(return_val), None))
mock_request_calls.append(mock.call(
end_url(path), 'DELETE', end_url(path), 'DELETE',
body=None, body=None,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN})))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(
return_val), None))
def _test_delete_resource(self, resource, cmd, myid, args, def _test_delete_resource(self, resource, cmd, myid, args,
cmd_resource=None, parent_id=None, cmd_resource=None, parent_id=None,
extra_id=None, delete_fail=False): extra_id=None, delete_fail=False):
self.mox.StubOutWithMock(cmd, "get_client") mock_request_calls = []
self.mox.StubOutWithMock(self.client.httpclient, "request") mock_request_returns = []
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource: if not cmd_resource:
cmd_resource = resource cmd_resource = resource
path = getattr(self.client, cmd_resource + "_path") path = getattr(self.client, cmd_resource + "_path")
self._test_set_path_and_delete(path, parent_id, myid) self._test_set_path_and_delete(path, parent_id, myid,
mock_request_calls,
mock_request_returns)
# extra_id is used to test for bulk_delete # extra_id is used to test for bulk_delete
if extra_id: if extra_id:
self._test_set_path_and_delete(path, parent_id, extra_id, self._test_set_path_and_delete(path, parent_id, extra_id,
mock_request_calls,
mock_request_returns,
delete_fail) delete_fail)
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("delete_" + cmd_resource) cmd_parser = cmd.get_parser("delete_" + cmd_resource)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll() with mock.patch.object(cmd, "get_client",
self.mox.UnsetStubs() return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient,
"request") as mock_request:
mock_request.side_effect = mock_request_returns
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_has_calls(mock_request_calls)
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)
if extra_id: if extra_id:
@@ -582,24 +605,25 @@ class CLITestV20Base(base.BaseTestCase):
def _test_update_resource_action(self, resource, cmd, myid, action, args, def _test_update_resource_action(self, resource, cmd, myid, action, args,
body, expected_code=200, retval=None, body, expected_code=200, retval=None,
cmd_resource=None): cmd_resource=None):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource: if not cmd_resource:
cmd_resource = resource cmd_resource = resource
path = getattr(self.client, cmd_resource + "_path") path = getattr(self.client, cmd_resource + "_path")
path_action = '%s/%s' % (myid, action) path_action = '%s/%s' % (myid, action)
self.client.httpclient.request( cmd_parser = cmd.get_parser("update_" + cmd_resource)
resp = (MyResp(expected_code), retval)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
end_url(path % path_action), 'PUT', end_url(path % path_action), 'PUT',
body=MyComparator(body, self.client), body=MyComparator(body, self.client),
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
'X-Auth-Token', TOKEN)).AndReturn((MyResp(expected_code),
retval))
self.mox.ReplayAll()
cmd_parser = cmd.get_parser("update_" + cmd_resource)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)
@@ -641,7 +665,6 @@ class ListCommandTestCase(CLITestV20Base):
query=query) query=query)
def _test_list_resources_with_arg_error(self, base_args=''): def _test_list_resources_with_arg_error(self, base_args=''):
self.addCleanup(self.mox.UnsetStubs)
resources = 'test_resources' resources = 'test_resources'
cmd = TestListCommand(MyApp(sys.stdout), None) cmd = TestListCommand(MyApp(sys.stdout), None)
# argparse parse error leads to SystemExit # argparse parse error leads to SystemExit
@@ -690,7 +713,6 @@ class ListCommandTestCase(CLITestV20Base):
class ClientV2TestJson(CLITestV20Base): class ClientV2TestJson(CLITestV20Base):
def test_do_request_unicode(self): def test_do_request_unicode(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
unicode_text = u'\u7f51\u7edc' unicode_text = u'\u7f51\u7edc'
# url with unicode # url with unicode
action = u'/test' action = u'/test'
@@ -705,46 +727,40 @@ class ClientV2TestJson(CLITestV20Base):
unicode_text) unicode_text)
expected_auth_token = encodeutils.safe_encode(unicode_text) expected_auth_token = encodeutils.safe_encode(unicode_text)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
resp = (MyResp(200, resp_headers), expect_body)
self.client.httpclient.request( with mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
result = self.client.do_request('PUT', action, body=body,
params=params)
mock_request.assert_called_once_with(
end_url(expected_action, query=expect_query), end_url(expected_action, query=expect_query),
'PUT', body=expect_body, 'PUT', body=expect_body,
headers=mox.ContainsKeyValue( headers=ContainsKeyValue({'X-Auth-Token': expected_auth_token}))
'X-Auth-Token',
expected_auth_token)).AndReturn((MyResp(200, resp_headers),
expect_body))
self.mox.ReplayAll()
result = self.client.do_request('PUT', action, body=body,
params=params)
self.mox.VerifyAll()
self.mox.UnsetStubs()
# test response with unicode # test response with unicode
self.assertEqual(body, result) self.assertEqual(body, result)
def test_do_request_error_without_response_body(self): def test_do_request_error_without_response_body(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
params = {'test': 'value'} params = {'test': 'value'}
expect_query = six.moves.urllib.parse.urlencode(params) expect_query = six.moves.urllib.parse.urlencode(params)
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
resp = (MyResp(400, headers=resp_headers, reason='An error'), '')
self.client.httpclient.request( with mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
error = self.assertRaises(exceptions.NeutronClientException,
self.client.do_request, 'PUT', '/test',
body='', params=params)
mock_request.assert_called_once_with(
MyUrlComparator(end_url('/test', query=expect_query), self.client), MyUrlComparator(end_url('/test', query=expect_query), self.client),
'PUT', body='', 'PUT', body='',
headers=mox.ContainsKeyValue('X-Auth-Token', 'token') headers=ContainsKeyValue({'X-Auth-Token': 'token'}))
).AndReturn((MyResp(400, headers=resp_headers, reason='An error'), ''))
self.mox.ReplayAll()
error = self.assertRaises(exceptions.NeutronClientException,
self.client.do_request, 'PUT', '/test',
body='', params=params)
expected_error = "An error\nNeutron server returns " \ expected_error = "An error\nNeutron server returns " \
"request_ids: %s" % [REQUEST_ID] "request_ids: %s" % [REQUEST_ID]
self.assertEqual(expected_error, str(error)) self.assertEqual(expected_error, str(error))
self.mox.VerifyAll()
self.mox.UnsetStubs()
def test_do_request_with_long_uri_exception(self): def test_do_request_with_long_uri_exception(self):
long_string = 'x' * 8200 # 8200 > MAX_URI_LEN:8192 long_string = 'x' * 8200 # 8200 > MAX_URI_LEN:8192
@@ -755,31 +771,28 @@ class ClientV2TestJson(CLITestV20Base):
self.assertNotEqual(0, exception.excess) self.assertNotEqual(0, exception.excess)
def test_do_request_request_ids(self): def test_do_request_request_ids(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
params = {'test': 'value'} params = {'test': 'value'}
expect_query = six.moves.urllib.parse.urlencode(params) expect_query = six.moves.urllib.parse.urlencode(params)
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
body = params body = params
expect_body = self.client.serialize(body) expect_body = self.client.serialize(body)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
self.client.httpclient.request( resp = (MyResp(200, resp_headers), expect_body)
with mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
result = self.client.do_request('PUT', '/test', body=body,
params=params)
mock_request.assert_called_once_with(
MyUrlComparator(end_url('/test', query=expect_query), self.client), MyUrlComparator(end_url('/test', query=expect_query), self.client),
'PUT', body=expect_body, 'PUT', body=expect_body,
headers=mox.ContainsKeyValue('X-Auth-Token', 'token') headers=ContainsKeyValue({'X-Auth-Token': 'token'}))
).AndReturn((MyResp(200, resp_headers), expect_body))
self.mox.ReplayAll()
result = self.client.do_request('PUT', '/test', body=body,
params=params)
self.mox.VerifyAll()
self.mox.UnsetStubs()
self.assertEqual(body, result) self.assertEqual(body, result)
self.assertEqual([REQUEST_ID], result.request_ids) self.assertEqual([REQUEST_ID], result.request_ids)
def test_list_request_ids_with_retrieve_all_true(self): def test_list_request_ids_with_retrieve_all_true(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
path = '/test' path = '/test'
resources = 'tests' resources = 'tests'
fake_query = "marker=myid2&limit=2" fake_query = "marker=myid2&limit=2"
@@ -792,30 +805,27 @@ class ClientV2TestJson(CLITestV20Base):
resstr1 = self.client.serialize(reses1) resstr1 = self.client.serialize(reses1)
resstr2 = self.client.serialize(reses2) resstr2 = self.client.serialize(reses2)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
self.client.httpclient.request( resp = [(MyResp(200, resp_headers), resstr1),
end_url(path, ""), 'GET', (MyResp(200, resp_headers), resstr2)]
body=None, with mock.patch.object(self.client.httpclient, "request",
headers=mox.ContainsKeyValue( side_effect=resp) as mock_request:
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200, resp_headers), result = self.client.list(resources, path)
resstr1))
self.client.httpclient.request(
MyUrlComparator(end_url(path, fake_query),
self.client), 'GET',
body=None,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200, resp_headers),
resstr2))
self.mox.ReplayAll()
result = self.client.list(resources, path)
self.mox.VerifyAll() self.assertEqual(2, mock_request.call_count)
self.mox.UnsetStubs() mock_request.assert_has_calls([
mock.call(
end_url(path, ""), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN})),
mock.call(
MyUrlComparator(end_url(path, fake_query),
self.client), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))])
self.assertEqual([REQUEST_ID, REQUEST_ID], result.request_ids) self.assertEqual([REQUEST_ID, REQUEST_ID], result.request_ids)
def test_list_request_ids_with_retrieve_all_false(self): def test_list_request_ids_with_retrieve_all_false(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
path = '/test' path = '/test'
resources = 'tests' resources = 'tests'
fake_query = "marker=myid2&limit=2" fake_query = "marker=myid2&limit=2"
@@ -828,26 +838,26 @@ class ClientV2TestJson(CLITestV20Base):
resstr1 = self.client.serialize(reses1) resstr1 = self.client.serialize(reses1)
resstr2 = self.client.serialize(reses2) resstr2 = self.client.serialize(reses2)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
self.client.httpclient.request( resp = [(MyResp(200, resp_headers), resstr1),
end_url(path, ""), 'GET', (MyResp(200, resp_headers), resstr2)]
body=None, with mock.patch.object(self.client.httpclient, "request",
headers=mox.ContainsKeyValue( side_effect=resp) as mock_request:
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200, resp_headers), result = self.client.list(resources, path, retrieve_all=False)
resstr1)) next(result)
self.client.httpclient.request( self.assertEqual([REQUEST_ID], result.request_ids)
MyUrlComparator(end_url(path, fake_query), self.client), 'GET', next(result)
body=None, self.assertEqual([REQUEST_ID, REQUEST_ID], result.request_ids)
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200, resp_headers), self.assertEqual(2, mock_request.call_count)
resstr2)) mock_request.assert_has_calls([
self.mox.ReplayAll() mock.call(
result = self.client.list(resources, path, retrieve_all=False) end_url(path, ""), 'GET',
next(result) body=None,
self.assertEqual([REQUEST_ID], result.request_ids) headers=ContainsKeyValue({'X-Auth-Token': TOKEN})),
next(result) mock.call(
self.assertEqual([REQUEST_ID, REQUEST_ID], result.request_ids) MyUrlComparator(end_url(path, fake_query), self.client), 'GET',
self.mox.VerifyAll() body=None,
self.mox.UnsetStubs() headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))])
def test_deserialize_without_data(self): def test_deserialize_without_data(self):
data = u'' data = u''
@@ -855,51 +865,48 @@ class ClientV2TestJson(CLITestV20Base):
self.assertEqual(data, result) self.assertEqual(data, result)
def test_update_resource(self): def test_update_resource(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
params = {'test': 'value'} params = {'test': 'value'}
expect_query = six.moves.urllib.parse.urlencode(params) expect_query = six.moves.urllib.parse.urlencode(params)
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
body = params body = params
expect_body = self.client.serialize(body) expect_body = self.client.serialize(body)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
self.client.httpclient.request( resp = (MyResp(200, resp_headers), expect_body)
with mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
result = self.client._update_resource('/test', body=body,
params=params)
mock_request.assert_called_once_with(
MyUrlComparator(end_url('/test', query=expect_query), self.client), MyUrlComparator(end_url('/test', query=expect_query), self.client),
'PUT', body=expect_body, 'PUT', body=expect_body,
headers=mox.And( headers=ContainsKeyValue({'X-Auth-Token': 'token'}))
mox.ContainsKeyValue('X-Auth-Token', 'token'), self.assertNotIn('If-Match', mock_request.call_args[1]['headers'])
mox.Not(mox.In('If-Match')))
).AndReturn((MyResp(200, resp_headers), expect_body))
self.mox.ReplayAll()
result = self.client._update_resource('/test', body=body,
params=params)
self.mox.VerifyAll()
self.mox.UnsetStubs()
self.assertEqual(body, result) self.assertEqual(body, result)
self.assertEqual([REQUEST_ID], result.request_ids) self.assertEqual([REQUEST_ID], result.request_ids)
def test_update_resource_with_revision_number(self): def test_update_resource_with_revision_number(self):
self.mox.StubOutWithMock(self.client.httpclient, "request")
params = {'test': 'value'} params = {'test': 'value'}
expect_query = six.moves.urllib.parse.urlencode(params) expect_query = six.moves.urllib.parse.urlencode(params)
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
body = params body = params
expect_body = self.client.serialize(body) expect_body = self.client.serialize(body)
resp_headers = {'x-openstack-request-id': REQUEST_ID} resp_headers = {'x-openstack-request-id': REQUEST_ID}
self.client.httpclient.request( resp = (MyResp(200, resp_headers), expect_body)
with mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
result = self.client._update_resource('/test', body=body,
params=params,
revision_number=1)
mock_request.assert_called_once_with(
MyUrlComparator(end_url('/test', query=expect_query), self.client), MyUrlComparator(end_url('/test', query=expect_query), self.client),
'PUT', body=expect_body, 'PUT', body=expect_body,
headers=mox.And( headers=ContainsKeyValue(
mox.ContainsKeyValue('X-Auth-Token', 'token'), {'X-Auth-Token': 'token', 'If-Match': 'revision_number=1'}))
mox.ContainsKeyValue('If-Match', 'revision_number=1'))
).AndReturn((MyResp(200, resp_headers), expect_body))
self.mox.ReplayAll()
result = self.client._update_resource('/test', body=body,
params=params, revision_number=1)
self.mox.VerifyAll()
self.mox.UnsetStubs()
self.assertEqual(body, result) self.assertEqual(body, result)
self.assertEqual([REQUEST_ID], result.request_ids) self.assertEqual([REQUEST_ID], result.request_ids)
@@ -1051,23 +1058,21 @@ class CLITestV20ExceptionHandler(CLITestV20Base):
self.assertEqual(599, e.status_code) self.assertEqual(599, e.status_code)
def test_connection_failed(self): def test_connection_failed(self):
self.mox.StubOutWithMock(self.client.httpclient, 'request')
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
excp = requests.exceptions.ConnectionError('Connection refused')
self.client.httpclient.request( with mock.patch.object(self.client.httpclient, "request",
side_effect=excp) as mock_request:
error = self.assertRaises(exceptions.ConnectionFailed,
self.client.get, '/test')
mock_request.assert_called_once_with(
end_url('/test'), 'GET', end_url('/test'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', 'token') body=None,
).AndRaise(requests.exceptions.ConnectionError('Connection refused')) headers=ContainsKeyValue({'X-Auth-Token': 'token'}))
self.mox.ReplayAll()
error = self.assertRaises(exceptions.ConnectionFailed,
self.client.get, '/test')
# NB: ConnectionFailed has no explicit status_code, so this # NB: ConnectionFailed has no explicit status_code, so this
# tests that there is a fallback defined. # tests that there is a fallback defined.
self.assertIsNotNone(error.status_code) self.assertIsNotNone(error.status_code)
self.mox.VerifyAll()
self.mox.UnsetStubs()
class DictWithMetaTest(base.BaseTestCase): class DictWithMetaTest(base.BaseTestCase):
@@ -1183,14 +1188,14 @@ class CLITestV20OutputFormatter(CLITestV20Base):
self.assertEqual('myname', data['name']) self.assertEqual('myname', data['name'])
self.assertEqual('myid', data['id']) self.assertEqual('myid', data['id'])
def _test_list_resources_with_formatter(self, fmt): @mock.patch.object(network.ListNetwork, "extend_list")
def _test_list_resources_with_formatter(self, fmt, mock_extend_list):
resources = 'networks' resources = 'networks'
cmd = network.ListNetwork(MyApp(sys.stdout), None) cmd = network.ListNetwork(MyApp(sys.stdout), None)
# ListNetwork has its own extend_list, so we need to stub out it # ListNetwork has its own extend_list, so we need to stub out it
# to avoid an extra API call. # to avoid an extra API call.
self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
self._test_list_resources(resources, cmd, output_format=fmt) self._test_list_resources(resources, cmd, output_format=fmt)
mock_extend_list.assert_called_once_with(IsA(list), mock.ANY)
def test_list_resources_table(self): def test_list_resources_table(self):
self._test_list_resources_with_formatter('table') self._test_list_resources_with_formatter('table')