openstack
/
shade
Code Issues Proposed changes

Convert list servers tests to requests_mock 

Also, found a few others that had been missed before.

Change-Id: I9dfbabd2b5318a59b91edf4d8a62ccc0ccb9baa0
This commit is contained in:
Monty Taylor 2017-06-15 21:50:52 -05:00
parent c5245648ed
commit 83c8bf5a34
No known key found for this signature in database
GPG Key ID: 7BAE94BC7141A594
5 changed files with 164 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
shade/tests/unit/test_caching.py
View File

@ -178,18 +178,23 @@ class TestMemoryCache(base.RequestsMockTestCase):
self.cloud.list_projects()) self.cloud.list_projects())
self.assert_calls() self.assert_calls()
@mock.patch('shade.OpenStackCloud.nova_client') def test_list_servers_no_herd(self):
def test_list_servers_no_herd(self, nova_mock):
self.cloud._SERVER_AGE = 2 self.cloud._SERVER_AGE = 2
fake_server = fakes.FakeServer('1234', '', 'ACTIVE') fake_server = fakes.make_fake_server('1234', 'name')
nova_mock.servers.list.return_value = [fake_server] self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [fake_server]}),
])
with concurrent.futures.ThreadPoolExecutor(16) as pool: with concurrent.futures.ThreadPoolExecutor(16) as pool:
for i in range(16): for i in range(16):
pool.submit(lambda: self.cloud.list_servers(bare=True)) pool.submit(lambda: self.cloud.list_servers(bare=True))
# It's possible to race-condition 16 threads all in the # It's possible to race-condition 16 threads all in the
# single initial lock without a tiny sleep # single initial lock without a tiny sleep
time.sleep(0.001) time.sleep(0.001)
self.assertEqual(1, nova_mock.servers.list.call_count)
self.assert_calls()
def test_list_volumes(self): def test_list_volumes(self):
fake_volume = fakes.FakeVolume('volume1', 'available', fake_volume = fakes.FakeVolume('volume1', 'available',

17
shade/tests/unit/test_image_snapshot.py
View File

@ -25,6 +25,9 @@ class TestImageSnapshot(base.RequestsMockTestCase):
super(TestImageSnapshot, self).setUp() super(TestImageSnapshot, self).setUp()
self.server_id = str(uuid.uuid4()) self.server_id = str(uuid.uuid4())
self.image_id = str(uuid.uuid4()) self.image_id = str(uuid.uuid4())
self.server_name = self.getUniqueString('name')
self.fake_server = fakes.make_fake_server(
self.server_id, self.server_name)
def test_create_image_snapshot_wait_until_active_never_active(self): def test_create_image_snapshot_wait_until_active_never_active(self):
snapshot_name = 'test-snapshot' snapshot_name = 'test-snapshot'
@ -97,17 +100,3 @@ class TestImageSnapshot(base.RequestsMockTestCase):
self.assertEqual(image['id'], self.image_id) self.assertEqual(image['id'], self.image_id)
self.assert_calls() self.assert_calls()
def test_create_image_snapshot_bad_name_exception(self):
self.register_uris([
dict(
method='POST',
uri='{endpoint}/servers/{server_id}/action'.format(
endpoint=fakes.COMPUTE_ENDPOINT,
server_id=self.server_id),
json=dict(servers=[])),
])
self.assertRaises(
exc.OpenStackCloudException,
self.cloud.create_image_snapshot,
'test-snapshot', self.server_id)

66
shade/tests/unit/test_server_delete_metadata.py
View File

@ -17,35 +17,59 @@ test_server_delete_metadata
Tests for the `delete_server_metadata` command. Tests for the `delete_server_metadata` command.
""" """
import mock import uuid
from shade import OpenStackCloud from shade.exc import OpenStackCloudURINotFound
from shade.exc import OpenStackCloudException from shade.tests import fakes
from shade.tests.unit import base from shade.tests.unit import base
class TestServerDeleteMetadata(base.TestCase): class TestServerDeleteMetadata(base.RequestsMockTestCase):
@mock.patch.object(OpenStackCloud, 'nova_client') def setUp(self):
def test_server_delete_metadata_with_exception(self, mock_nova): super(TestServerDeleteMetadata, self).setUp()
self.server_id = str(uuid.uuid4())
self.server_name = self.getUniqueString('name')
self.fake_server = fakes.make_fake_server(
self.server_id, self.server_name)
def test_server_delete_metadata_with_exception(self):
""" """
Test that a generic exception in the novaclient delete_meta raises Test that a missing metadata throws an exception.
an exception in delete_server_metadata.
""" """
mock_nova.servers.delete_meta.side_effect = Exception("exception") self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [self.fake_server]}),
dict(method='DELETE',
uri=self.get_mock_url(
'compute', 'public',
append=['servers', self.fake_server['id'],
'metadata', 'key']),
status_code=404),
])
self.assertRaises( self.assertRaises(
OpenStackCloudException, self.cloud.delete_server_metadata, OpenStackCloudURINotFound, self.cloud.delete_server_metadata,
{'id': 'server-id'}, ['key']) self.server_name, ['key'])
@mock.patch.object(OpenStackCloud, 'nova_client') self.assert_calls()
def test_server_delete_metadata_with_exception_reraise(self, mock_nova):
"""
Test that an OpenStackCloudException exception gets re-raised
in delete_server_metadata.
"""
mock_nova.servers.delete_meta.side_effect = OpenStackCloudException("")
self.assertRaises( def test_server_delete_metadata(self):
OpenStackCloudException, self.cloud.delete_server_metadata, self.register_uris([
'server-id', ['key']) dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [self.fake_server]}),
dict(method='DELETE',
uri=self.get_mock_url(
'compute', 'public',
append=['servers', self.fake_server['id'],
'metadata', 'key']),
status_code=200),
])
self.cloud.delete_server_metadata(self.server_id, ['key'])
self.assert_calls()

69
shade/tests/unit/test_server_set_metadata.py
View File

@ -17,35 +17,62 @@ test_server_set_metadata
Tests for the `set_server_metadata` command. Tests for the `set_server_metadata` command.
""" """
import mock import uuid
from shade import OpenStackCloud from shade.exc import OpenStackCloudBadRequest
from shade.exc import OpenStackCloudException from shade.tests import fakes
from shade.tests.unit import base from shade.tests.unit import base
class TestServerSetMetadata(base.TestCase): class TestServerSetMetadata(base.RequestsMockTestCase):
@mock.patch.object(OpenStackCloud, 'nova_client') def setUp(self):
def test_server_set_metadata_with_set_meta_exception(self, mock_nova): super(TestServerSetMetadata, self).setUp()
self.server_id = str(uuid.uuid4())
self.server_name = self.getUniqueString('name')
self.fake_server = fakes.make_fake_server(
self.server_id, self.server_name)
def test_server_set_metadata_with_exception(self):
""" """
Test that a generic exception in the novaclient set_meta raises Test that a generic exception in the novaclient delete_meta raises
an exception in set_server_metadata. an exception in delete_server_metadata.
""" """
mock_nova.servers.set_meta.side_effect = Exception("exception") self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [self.fake_server]}),
dict(method='POST',
uri=self.get_mock_url(
'compute', 'public',
append=['servers', self.fake_server['id'],
'metadata']),
validate=dict(json={'metadata': {'meta': 'data'}}),
json={},
status_code=400),
])
self.assertRaises( self.assertRaises(
OpenStackCloudException, self.cloud.set_server_metadata, OpenStackCloudBadRequest, self.cloud.set_server_metadata,
{'id': 'server-id'}, {'meta': 'data'}) self.server_name, {'meta': 'data'})
@mock.patch.object(OpenStackCloud, 'nova_client') self.assert_calls()
def test_server_set_metadata_with_exception_reraise(self, mock_nova):
"""
Test that an OpenStackCloudException exception gets re-raised
in set_server_metadata.
"""
mock_nova.servers.set_meta.side_effect = OpenStackCloudException("")
self.assertRaises( def test_server_set_metadata(self):
OpenStackCloudException, self.cloud.set_server_metadata, self.register_uris([
'server-id', {'meta': 'data'}) dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [self.fake_server]}),
dict(method='POST',
uri=self.get_mock_url(
'compute', 'public',
append=['servers', self.fake_server['id'], 'metadata']),
validate=dict(json={'metadata': {'meta': 'data'}}),
status_code=200),
])
self.cloud.set_server_metadata(self.server_id, {'meta': 'data'})
self.assert_calls()

124
shade/tests/unit/test_shade.py
View File

@ -11,7 +11,7 @@
# under the License. # under the License.
import mock import mock
import munch import uuid
import testtools import testtools
@ -68,32 +68,49 @@ class TestShade(base.RequestsMockTestCase):
r = self.cloud.get_image('doesNotExist') r = self.cloud.get_image('doesNotExist')
self.assertIsNone(r) self.assertIsNone(r)
@mock.patch.object(shade.OpenStackCloud, '_expand_server') def test_get_server(self):
@mock.patch.object(shade.OpenStackCloud, 'list_servers') server1 = fakes.make_fake_server('123', 'mickey')
def test_get_server(self, mock_list, mock_expand): server2 = fakes.make_fake_server('345', 'mouse')
server1 = dict(id='123', name='mickey')
server2 = dict(id='345', name='mouse') self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': [server1, server2]}),
])
def expand_server(server, detailed, bare):
return server
mock_expand.side_effect = expand_server
mock_list.return_value = [server1, server2]
r = self.cloud.get_server('mickey') r = self.cloud.get_server('mickey')
self.assertIsNotNone(r) self.assertIsNotNone(r)
self.assertDictEqual(server1, r) self.assertEqual(server1['name'], r['name'])
self.assert_calls()
def test_get_server_not_found(self):
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
json={'servers': []}),
])
@mock.patch.object(shade.OpenStackCloud, 'search_servers')
def test_get_server_not_found(self, mock_search):
mock_search.return_value = []
r = self.cloud.get_server('doesNotExist') r = self.cloud.get_server('doesNotExist')
self.assertIsNone(r) self.assertIsNone(r)
@mock.patch.object(shade.OpenStackCloud, 'nova_client') self.assert_calls()
def test_list_servers_exception(self, mock_client):
mock_client.servers.list.side_effect = Exception() def test_list_servers_exception(self):
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['servers', 'detail']),
status_code=400)
])
self.assertRaises(exc.OpenStackCloudException, self.assertRaises(exc.OpenStackCloudException,
self.cloud.list_servers) self.cloud.list_servers)
self.assert_calls()
def test__neutron_exceptions_resource_not_found(self): def test__neutron_exceptions_resource_not_found(self):
self.register_uris([ self.register_uris([
dict(method='GET', dict(method='GET',
@ -116,64 +133,39 @@ class TestShade(base.RequestsMockTestCase):
self.cloud.list_networks) self.cloud.list_networks)
self.assert_calls() self.assert_calls()
@mock.patch.object(shade._tasks.ServerList, 'main') def test_list_servers(self):
@mock.patch('shade.meta.add_server_interfaces') server_id = str(uuid.uuid4())
def test_list_servers(self, mock_add_srv_int, mock_serverlist): server_name = self.getUniqueString('name')
'''This test verifies that calling list_servers results in a call fake_server = fakes.make_fake_server(server_id, server_name)
to the ServerList task.''' self.register_uris([
server_obj = munch.Munch({'name': 'testserver', dict(method='GET',
'id': '1', uri=self.get_mock_url(
'flavor': {}, 'compute', 'public', append=['servers', 'detail']),
'addresses': {}, json={'servers': [fake_server]}),
'accessIPv4': '', ])
'accessIPv6': '',
'image': ''})
mock_serverlist.return_value = [server_obj]
mock_add_srv_int.side_effect = [server_obj]
r = self.cloud.list_servers() r = self.cloud.list_servers()
self.assertEqual(1, len(r)) self.assertEqual(1, len(r))
self.assertEqual(1, mock_add_srv_int.call_count) self.assertEqual(server_name, r[0]['name'])
self.assertEqual('testserver', r[0]['name'])
@mock.patch.object(shade._tasks.ServerList, 'main') self.assert_calls()
@mock.patch('shade.meta.get_hostvars_from_server')
def test_list_servers_detailed(self, def test_list_servers_all_projects(self):
mock_get_hostvars_from_server,
mock_serverlist):
'''This test verifies that when list_servers is called with '''This test verifies that when list_servers is called with
`detailed=True` that it calls `get_hostvars_from_server` for each `all_projects=True` that it passes `all_tenants=True` to nova.'''
server in the list.''' self.register_uris([
mock_serverlist.return_value = [ dict(method='GET',
fakes.FakeServer('server1', '', 'ACTIVE'), uri=self.get_mock_url(
fakes.FakeServer('server2', '', 'ACTIVE'), 'compute', 'public', append=['servers', 'detail'],
] qs_elements=['all_tenants=True']),
mock_get_hostvars_from_server.side_effect = [ complete_qs=True,
{'name': 'server1', 'id': '1'}, json={'servers': []}),
{'name': 'server2', 'id': '2'}, ])
]
r = self.cloud.list_servers(detailed=True)
self.assertEqual(2, len(r))
self.assertEqual(len(r), mock_get_hostvars_from_server.call_count)
self.assertEqual('server1', r[0]['name'])
self.assertEqual('server2', r[1]['name'])
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_servers_all_projects(self, mock_nova_client):
'''This test verifies that when list_servers is called with
`all_projects=True` that it passes `all_tenants=1` to novaclient.'''
mock_nova_client.servers.list.return_value = [
fakes.FakeServer('server1', '', 'ACTIVE'),
fakes.FakeServer('server2', '', 'ACTIVE'),
]
self.cloud.list_servers(all_projects=True) self.cloud.list_servers(all_projects=True)
mock_nova_client.servers.list.assert_called_with( self.assert_calls()
search_opts={'all_tenants': True})
def test_iterate_timeout_bad_wait(self): def test_iterate_timeout_bad_wait(self):
with testtools.ExpectedException( with testtools.ExpectedException(