Switch Neutron/Network API tests to mock
This patch changes mox to mock for the following test modules: * openstack_dashboard/test/unit/api/test_network.py * openstack_dashboard/test/unit/api/test_neutron.py * openstack_dashboard/test/unit/api/rest/test_network.py * openstack_dashboard/test/unit/api/rest/test_neutron.py Partially-Implements: blueprint mock-framework-in-unit-tests Change-Id: Ib97fb1c06a883fb9fd5ce39b8fe2362a88135692
This commit is contained in:
parent
415c1066ff
commit
c5cd0b2048
|
@ -590,6 +590,7 @@ class APITestCase(TestCase):
|
|||
return self.glanceclient
|
||||
|
||||
def stub_neutronclient(self):
|
||||
self._warn_client('neutron', 'S')
|
||||
if not hasattr(self, "neutronclient"):
|
||||
self.mox.StubOutWithMock(neutron_client, 'Client')
|
||||
self.neutronclient = self.mox.CreateMock(neutron_client.Client)
|
||||
|
@ -638,6 +639,11 @@ class APIMockTestCase(APITestCase):
|
|||
self.keystoneclient = mock.Mock()
|
||||
return self.keystoneclient
|
||||
|
||||
def stub_neutronclient(self):
|
||||
if not hasattr(self, "neutronclient"):
|
||||
self.neutronclient = mock.Mock()
|
||||
return self.neutronclient
|
||||
|
||||
def stub_swiftclient(self):
|
||||
# This method should not be called.
|
||||
raise NotImplementedError
|
||||
|
|
|
@ -20,6 +20,8 @@ from openstack_dashboard.test import helpers as test
|
|||
|
||||
class RestNetworkApiSecurityGroupTests(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
@mock.patch.object(network.api, 'neutron')
|
||||
def test_security_group_detailed(self, client):
|
||||
request = self.mock_rest_request()
|
||||
|
@ -36,6 +38,8 @@ class RestNetworkApiSecurityGroupTests(test.TestCase):
|
|||
|
||||
class RestNetworkApiFloatingIpTests(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
@mock.patch.object(network.api, 'neutron')
|
||||
def test_floating_ip_list(self, client):
|
||||
request = self.mock_rest_request()
|
||||
|
|
|
@ -27,6 +27,9 @@ TEST = TestData(neutron_data.data)
|
|||
|
||||
|
||||
class NeutronNetworksTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronNetworksTestCase, self).setUp()
|
||||
self._networks = [test.mock_factory(n)
|
||||
|
@ -72,43 +75,45 @@ class NeutronNetworksTestCase(test.TestCase):
|
|||
# Services
|
||||
#
|
||||
|
||||
@test.create_stubs({api.base: ('is_service_enabled',)})
|
||||
@test.create_stubs({api.neutron: ('is_extension_supported',)})
|
||||
@mock.patch.object(neutron.api, 'neutron')
|
||||
def test_services_get(self, client):
|
||||
@mock.patch.object(api.base, 'is_service_enabled')
|
||||
@mock.patch.object(api, 'neutron')
|
||||
def test_services_get(self, client, mock_is_service_enabled):
|
||||
params = django_request.QueryDict('network_id=the_network')
|
||||
request = self.mock_rest_request(GET=params)
|
||||
|
||||
api.base.is_service_enabled(request, 'network').AndReturn(True)
|
||||
api.neutron.is_extension_supported(request, 'agent').AndReturn(True)
|
||||
mock_is_service_enabled.return_value = True
|
||||
client.is_extension_supported.return_value = True
|
||||
|
||||
client.agent_list.return_value = [
|
||||
mock.Mock(**{'to_dict.return_value': {'id': '1'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'id': '2'}})
|
||||
]
|
||||
self.mox.ReplayAll()
|
||||
|
||||
response = neutron.Services().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
mock_is_service_enabled.assert_called_once_with(request, 'network')
|
||||
client.is_extension_supported.assert_called_once_with(request, 'agent')
|
||||
client.agent_list.assert_called_once_with(
|
||||
request, network_id='the_network')
|
||||
self.assertEqual(response.content.decode('utf-8'),
|
||||
'{"items": [{"id": "1"}, {"id": "2"}]}')
|
||||
|
||||
@test.create_stubs({api.base: ('is_service_enabled',)})
|
||||
def test_services_get_disabled(self):
|
||||
@mock.patch.object(api.base, 'is_service_enabled')
|
||||
def test_services_get_disabled(self, mock_is_service_enabled):
|
||||
request = self.mock_rest_request(
|
||||
GET={"network_id": self._networks[0].id})
|
||||
|
||||
api.base.is_service_enabled(request, 'network').AndReturn(False)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
mock_is_service_enabled.return_value = False
|
||||
|
||||
response = neutron.Services().get(request)
|
||||
self.assertStatusCode(response, 501)
|
||||
mock_is_service_enabled.assert_called_once_with(request, 'network')
|
||||
|
||||
|
||||
class NeutronSubnetsTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronSubnetsTestCase, self).setUp()
|
||||
self._networks = [test.mock_factory(n)
|
||||
|
@ -143,6 +148,9 @@ class NeutronSubnetsTestCase(test.TestCase):
|
|||
|
||||
|
||||
class NeutronPortsTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronPortsTestCase, self).setUp()
|
||||
self._networks = [test.mock_factory(n)
|
||||
|
@ -164,6 +172,8 @@ class NeutronPortsTestCase(test.TestCase):
|
|||
|
||||
class NeutronTrunkTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
@mock.patch.object(neutron.api, 'neutron')
|
||||
def test_trunk_delete(self, client):
|
||||
request = self.mock_rest_request()
|
||||
|
@ -194,6 +204,9 @@ class NeutronTrunkTestCase(test.TestCase):
|
|||
|
||||
|
||||
class NeutronTrunksTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronTrunksTestCase, self).setUp()
|
||||
self._trunks = [test.mock_factory(n)
|
||||
|
@ -222,6 +235,9 @@ class NeutronTrunksTestCase(test.TestCase):
|
|||
|
||||
|
||||
class NeutronExtensionsTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronExtensionsTestCase, self).setUp()
|
||||
|
||||
|
@ -238,49 +254,53 @@ class NeutronExtensionsTestCase(test.TestCase):
|
|||
|
||||
|
||||
class NeutronDefaultQuotasTestCase(test.TestCase):
|
||||
@test.create_stubs({base: ('is_service_enabled',)})
|
||||
@mock.patch.object(neutron.api, 'neutron')
|
||||
def test_quotas_sets_defaults_get_when_service_is_enabled(self, client):
|
||||
|
||||
use_mox = False
|
||||
|
||||
@mock.patch.object(api.base, 'is_service_enabled')
|
||||
@mock.patch.object(api, 'neutron')
|
||||
def test_quotas_sets_defaults_get_when_service_is_enabled(
|
||||
self, client, mock_is_service_enabled):
|
||||
filters = {'user': {'tenant_id': 'tenant'}}
|
||||
request = self.mock_rest_request(**{'GET': dict(filters)})
|
||||
|
||||
base.is_service_enabled(request, 'network').AndReturn(True)
|
||||
|
||||
mock_is_service_enabled.return_value = True
|
||||
client.tenant_quota_get.return_value = [
|
||||
base.Quota("network", 100),
|
||||
base.Quota("q2", 101)]
|
||||
|
||||
self.mox.ReplayAll()
|
||||
|
||||
response = neutron.DefaultQuotaSets().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertItemsCollectionEqual(response, [
|
||||
{'limit': 100, 'display_name': 'Networks', 'name': 'network'},
|
||||
{'limit': 101, 'display_name': 'Q2', 'name': 'q2'}])
|
||||
|
||||
mock_is_service_enabled.assert_called_once_with(request, 'network')
|
||||
client.tenant_quota_get.assert_called_once_with(
|
||||
request,
|
||||
request.user.tenant_id)
|
||||
|
||||
@test.create_stubs({neutron.api.base: ('is_service_enabled',)})
|
||||
@mock.patch.object(neutron.api, 'neutron')
|
||||
def test_quota_sets_defaults_get_when_service_is_disabled(self, client):
|
||||
@mock.patch.object(api.base, 'is_service_enabled')
|
||||
@mock.patch.object(api, 'neutron')
|
||||
def test_quota_sets_defaults_get_when_service_is_disabled(
|
||||
self, client, mock_is_service_enabled):
|
||||
filters = {'user': {'tenant_id': 'tenant'}}
|
||||
request = self.mock_rest_request(**{'GET': dict(filters)})
|
||||
|
||||
base.is_service_enabled(request, 'network').AndReturn(False)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
mock_is_service_enabled.return_value = False
|
||||
|
||||
response = neutron.DefaultQuotaSets().get(request)
|
||||
self.assertStatusCode(response, 501)
|
||||
self.assertEqual(response.content.decode('utf-8'),
|
||||
'"Service Neutron is disabled."')
|
||||
|
||||
mock_is_service_enabled.assert_called_once_with(request, 'network')
|
||||
client.tenant_quota_get.assert_not_called()
|
||||
|
||||
|
||||
class NeutronQuotaSetsTestCase(test.TestCase):
|
||||
|
||||
use_mox = False
|
||||
|
||||
def setUp(self):
|
||||
super(NeutronQuotaSetsTestCase, self).setUp()
|
||||
|
||||
|
|
|
@ -14,6 +14,7 @@
|
|||
|
||||
import collections
|
||||
|
||||
import mock
|
||||
import netaddr
|
||||
|
||||
from django.test.utils import override_settings
|
||||
|
@ -22,14 +23,11 @@ from openstack_dashboard import api
|
|||
from openstack_dashboard.test import helpers as test
|
||||
|
||||
|
||||
class NetworkApiNeutronTestBase(test.APITestCase):
|
||||
class NetworkApiNeutronTests(test.APIMockTestCase):
|
||||
def setUp(self):
|
||||
super(NetworkApiNeutronTestBase, self).setUp()
|
||||
super(NetworkApiNeutronTests, self).setUp()
|
||||
self.qclient = self.stub_neutronclient()
|
||||
|
||||
|
||||
class NetworkApiNeutronTests(NetworkApiNeutronTestBase):
|
||||
|
||||
def _get_expected_addresses(self, server, no_fip_expected=True):
|
||||
server_ports = self.ports.filter(device_id=server.id)
|
||||
addresses = collections.defaultdict(list)
|
||||
|
@ -80,19 +78,15 @@ class NetworkApiNeutronTests(NetworkApiNeutronTestBase):
|
|||
server_networks = [net for net in self.api_networks.list()
|
||||
if net['id'] in server_network_ids]
|
||||
|
||||
self.qclient.list_ports(device_id=server_ids) \
|
||||
.AndReturn({'ports': server_ports})
|
||||
list_ports_retvals = [{'ports': server_ports}]
|
||||
self.qclient.list_ports.side_effect = list_ports_retvals
|
||||
if router_enabled:
|
||||
self.qclient.list_floatingips(tenant_id=tenant_id,
|
||||
port_id=server_port_ids) \
|
||||
.AndReturn({'floatingips': assoc_fips})
|
||||
self.qclient.list_ports(tenant_id=tenant_id) \
|
||||
.AndReturn({'ports': self.api_ports.list()})
|
||||
self.qclient.list_networks(id=frozenset(server_network_ids)) \
|
||||
.AndReturn({'networks': server_networks})
|
||||
self.qclient.list_subnets() \
|
||||
.AndReturn({'subnets': self.api_subnets.list()})
|
||||
self.mox.ReplayAll()
|
||||
self.qclient.list_floatingips.return_value = {'floatingips':
|
||||
assoc_fips}
|
||||
list_ports_retvals.append({'ports': self.api_ports.list()})
|
||||
self.qclient.list_networks.return_value = {'networks': server_networks}
|
||||
self.qclient.list_subnets.return_value = {'subnets':
|
||||
self.api_subnets.list()}
|
||||
|
||||
api.network.servers_update_addresses(self.request, servers)
|
||||
|
||||
|
@ -128,6 +122,18 @@ class NetworkApiNeutronTests(NetworkApiNeutronTestBase):
|
|||
# so it should be an empty dict.
|
||||
self.assertFalse(servers[2].addresses)
|
||||
|
||||
expected_list_ports = [mock.call(device_id=server_ids)]
|
||||
if router_enabled:
|
||||
self.qclient.list_floatingips.assert_called_once_with(
|
||||
tenant_id=tenant_id, port_id=server_port_ids)
|
||||
expected_list_ports.append(mock.call(tenant_id=tenant_id))
|
||||
else:
|
||||
self.assertEqual(0, self.qclient.list_floatingips.call_count)
|
||||
self.qclient.list_ports.assert_has_calls(expected_list_ports)
|
||||
self.qclient.list_networks.assert_called_once_with(
|
||||
id=frozenset(server_network_ids))
|
||||
self.qclient.list_subnets.assert_called_once_with()
|
||||
|
||||
@override_settings(OPENSTACK_NEUTRON_NETWORK={'enable_router': True})
|
||||
def test_servers_update_addresses(self):
|
||||
self._test_servers_update_addresses()
|
||||
|
|
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue