removing httplib2 test dependency

Once upon a time, the python-*client libraries were primarily built to
use httplib2. They have subsequently shift to using requests and thus
urllib3. The horizon test helpers code was maintaining a reference to
httplib2 as it intercepted errant library calls that were not mocked.

httplib2 is not actively maintained and OpenStack is moving to remove it
as a dependency. See
http://lists.openstack.org/pipermail/openstack-dev/2016-March/089225.html
for more details.

This patch removed the httplib2 dependency. Upon removing the
dependency it exposed a missed update from httplib2 to urllib3. A
function that was intended to catch unmocked calls was only listening
for httplib2 connections. This patch updates that failsafe to work with
urllib3. Upon doing so, it pointed out many, many missing mocks and in
turn, many broken tests that appeared to work because of API call
failures. This patch adds the missing mocks and fixes the broken tests.

The new failsafe prints the stack trace when an outside connection is
attempted. Additionally, to fix the fact that a missed mock used to
allow tests to potentially pass, as documented by bug 1517704, a test
failure is now forced on tests where a missing mock is detected.

Closes-Bug: #1517704
Implements blueprint: remove-httplib2-dep
Change-Id: Iaabdf03966c14c82e0c58a3b1ab1a6755c05adcb
This commit is contained in:
David Lyle 2016-03-16 10:25:26 -06:00
parent 5347015c5e
commit 037d1c0927
21 changed files with 564 additions and 180 deletions

View File

@ -179,11 +179,26 @@ class AggregatesViewTests(test.BaseAdminViewTests):
@mock.patch('openstack_dashboard.api.nova.extension_supported', @mock.patch('openstack_dashboard.api.nova.extension_supported',
mock.Mock(return_value=False)) mock.Mock(return_value=False))
@test.create_stubs({api.nova: ('aggregate_details_list', @test.create_stubs({api.nova: ('aggregate_details_list',
'availability_zone_list',), 'availability_zone_list',
api.cinder: ('tenant_absolute_limits',)}) 'tenant_absolute_limits',),
api.cinder: ('tenant_absolute_limits',),
api.neutron: ('list_extensions',),
api.network: ('tenant_floating_ip_list',
'security_group_list'),
api.keystone: ('tenant_list',)})
def test_panel_not_available(self): def test_panel_not_available(self):
api.nova.tenant_absolute_limits(IsA(http.HttpRequest)). \
MultipleTimes().AndReturn(self.limits['absolute'])
api.cinder.tenant_absolute_limits(IsA(http.HttpRequest)). \ api.cinder.tenant_absolute_limits(IsA(http.HttpRequest)). \
MultipleTimes().AndReturn(self.cinder_limits['absolute']) MultipleTimes().AndReturn(self.cinder_limits['absolute'])
api.neutron.list_extensions(IsA(http.HttpRequest)). \
AndReturn(self.api_extensions.list())
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(self.floating_ips.list())
api.network.security_group_list(IsA(http.HttpRequest)) \
.AndReturn(self.security_groups.list())
api.keystone.tenant_list(IsA(http.HttpRequest)) \
.AndReturn(self.tenants.list())
self.mox.ReplayAll() self.mox.ReplayAll()
self.patchers['aggregates'].stop() self.patchers['aggregates'].stop()

View File

@ -133,12 +133,16 @@ class InstanceViewTest(test.BaseAdminViewTests):
self.assertMessageCount(res, error=1) self.assertMessageCount(res, error=1)
self.assertItemsEqual(instances, servers) self.assertItemsEqual(instances, servers)
@test.create_stubs({api.nova: ('server_list',)}) @test.create_stubs({api.nova: ('server_list',),
api.keystone: ('tenant_list',)})
def test_index_server_list_exception(self): def test_index_server_list_exception(self):
tenants = self.tenants.list()
search_opts = {'marker': None, 'paginate': True} search_opts = {'marker': None, 'paginate': True}
api.nova.server_list(IsA(http.HttpRequest), api.nova.server_list(IsA(http.HttpRequest),
all_tenants=True, search_opts=search_opts) \ all_tenants=True, search_opts=search_opts) \
.AndRaise(self.exceptions.nova) .AndRaise(self.exceptions.nova)
api.keystone.tenant_list(IsA(http.HttpRequest)).\
AndReturn([tenants, False])
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -29,23 +29,28 @@ NETWORKS_DETAIL_URL = 'horizon:admin:networks:detail'
class NetworkPortTests(test.BaseAdminViewTests): class NetworkPortTests(test.BaseAdminViewTests):
@test.create_stubs({api.neutron: ('port_get', @test.create_stubs({api.neutron: ('network_get',
'port_get',
'is_extension_supported',)}) 'is_extension_supported',)})
def test_port_detail(self): def test_port_detail(self):
self._test_port_detail() self._test_port_detail()
@test.create_stubs({api.neutron: ('port_get', @test.create_stubs({api.neutron: ('network_get',
'port_get',
'is_extension_supported',)}) 'is_extension_supported',)})
def test_port_detail_with_mac_learning(self): def test_port_detail_with_mac_learning(self):
self._test_port_detail(mac_learning=True) self._test_port_detail(mac_learning=True)
def _test_port_detail(self, mac_learning=False): def _test_port_detail(self, mac_learning=False):
port = self.ports.first() port = self.ports.first()
network_id = self.networks.first().id
api.neutron.port_get(IsA(http.HttpRequest), port.id)\ api.neutron.port_get(IsA(http.HttpRequest), port.id)\
.AndReturn(self.ports.first()) .AndReturn(self.ports.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
'mac-learning')\ 'mac-learning')\
.MultipleTimes().AndReturn(mac_learning) .MultipleTimes().AndReturn(mac_learning)
api.neutron.network_get(IsA(http.HttpRequest), network_id)\
.AndReturn(self.networks.first())
self.mox.ReplayAll() self.mox.ReplayAll()
res = self.client.get(reverse(DETAIL_URL, args=[port.id])) res = self.client.get(reverse(DETAIL_URL, args=[port.id]))

View File

@ -63,12 +63,20 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
redir_url = NETWORKS_INDEX_URL redir_url = NETWORKS_INDEX_URL
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get',)}) @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list',)})
def test_subnet_create_get(self): def test_subnet_create_get(self):
network = self.networks.first() network = self.networks.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnets)
self.mox.ReplayAll() self.mox.ReplayAll()
url = reverse('horizon:admin:networks:addsubnet', url = reverse('horizon:admin:networks:addsubnet',
@ -78,16 +86,20 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
self.assertTemplateUsed(res, views.WorkflowView.template_name) self.assertTemplateUsed(res, views.WorkflowView.template_name)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list',
'subnet_create',)}) 'subnet_create',)})
def test_subnet_create_post(self): def test_subnet_create_post(self):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .MultipleTimes().AndReturn(self.networks.first())
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
network.id)\ 'subnet_allocation')\
.AndReturn(self.networks.first()) .MultipleTimes().AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnets)
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -131,16 +143,21 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list',
'subnet_create',)}) 'subnet_create',)})
def test_subnet_create_post_subnet_exception(self): def test_subnet_create_post_subnet_exception(self):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .MultipleTimes().AndReturn(self.networks.first())
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
network.id)\ 'subnet_allocation')\
.AndReturn(self.networks.first()) .AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnets)
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -160,13 +177,21 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
redir_url = reverse(NETWORKS_DETAIL_URL, args=[subnet.network_id]) redir_url = reverse(NETWORKS_DETAIL_URL, args=[subnet.network_id])
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get',)}) @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list',)})
def test_subnet_create_post_cidr_inconsistent(self): def test_subnet_create_post_cidr_inconsistent(self):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnets)
self.mox.ReplayAll() self.mox.ReplayAll()
# dummy IPv6 address # dummy IPv6 address
@ -180,13 +205,21 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
expected_msg = 'Network Address and IP version are inconsistent.' expected_msg = 'Network Address and IP version are inconsistent.'
self.assertContains(res, expected_msg) self.assertContains(res, expected_msg)
@test.create_stubs({api.neutron: ('network_get',)}) @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list',)})
def test_subnet_create_post_gw_inconsistent(self): def test_subnet_create_post_gw_inconsistent(self):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnets)
self.mox.ReplayAll() self.mox.ReplayAll()
# dummy IPv6 address # dummy IPv6 address
@ -200,9 +233,16 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
self.assertContains(res, 'Gateway IP and IP version are inconsistent.') self.assertContains(res, 'Gateway IP and IP version are inconsistent.')
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post(self): def test_subnet_update_post(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnetpools.list())
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
@ -224,9 +264,16 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_gw_inconsistent(self): def test_subnet_update_post_gw_inconsistent(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest))\
.AndReturn(self.subnetpools.list())
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -36,8 +36,11 @@ GROUP_ROLE_PREFIX = constants.DOMAIN_GROUP_MEMBER_SLUG + "_role_"
class DomainsViewTests(test.BaseAdminViewTests): class DomainsViewTests(test.BaseAdminViewTests):
@test.create_stubs({api.keystone: ('domain_list',)}) @test.create_stubs({api.keystone: ('domain_get',
'domain_list',)})
def test_index(self): def test_index(self):
domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
self.mox.ReplayAll() self.mox.ReplayAll()
@ -52,9 +55,12 @@ class DomainsViewTests(test.BaseAdminViewTests):
self.assertContains(res, 'Disable Domain') self.assertContains(res, 'Disable Domain')
self.assertContains(res, 'Enable Domain') self.assertContains(res, 'Enable Domain')
@test.create_stubs({api.keystone: ('domain_list', @test.create_stubs({api.keystone: ('domain_get',
'domain_list',
'keystone_can_edit_domain')}) 'keystone_can_edit_domain')})
def test_index_with_keystone_can_edit_domain_false(self): def test_index_with_keystone_can_edit_domain_false(self):
domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
api.keystone.keystone_can_edit_domain() \ api.keystone.keystone_can_edit_domain() \
.MultipleTimes().AndReturn(False) .MultipleTimes().AndReturn(False)
@ -71,11 +77,13 @@ class DomainsViewTests(test.BaseAdminViewTests):
self.assertNotContains(res, 'Disable Domain') self.assertNotContains(res, 'Disable Domain')
self.assertNotContains(res, 'Enable Domain') self.assertNotContains(res, 'Enable Domain')
@test.create_stubs({api.keystone: ('domain_list', @test.create_stubs({api.keystone: ('domain_get',
'domain_list',
'domain_delete')}) 'domain_delete')})
def test_delete_domain(self): def test_delete_domain(self):
domain = self.domains.get(id="2") domain = self.domains.get(id="2")
api.keystone.domain_get(IsA(http.HttpRequest), '2').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
api.keystone.domain_delete(IgnoreArg(), domain.id) api.keystone.domain_delete(IgnoreArg(), domain.id)
@ -86,10 +94,12 @@ class DomainsViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL) self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL)
@test.create_stubs({api.keystone: ('domain_list', )}) @test.create_stubs({api.keystone: ('domain_get',
'domain_list', )})
def test_delete_with_enabled_domain(self): def test_delete_with_enabled_domain(self):
domain = self.domains.get(id="1") domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
self.mox.ReplayAll() self.mox.ReplayAll()
@ -100,11 +110,13 @@ class DomainsViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL) self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL)
self.assertMessageCount(error=2) self.assertMessageCount(error=2)
@test.create_stubs({api.keystone: ('domain_list', @test.create_stubs({api.keystone: ('domain_get',
'domain_list',
'domain_update')}) 'domain_update')})
def test_disable(self): def test_disable(self):
domain = self.domains.get(id="1") domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
api.keystone.domain_update(IsA(http.HttpRequest), api.keystone.domain_update(IsA(http.HttpRequest),
description=domain.description, description=domain.description,
@ -120,11 +132,13 @@ class DomainsViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL) self.assertRedirectsNoFollow(res, DOMAINS_INDEX_URL)
self.assertMessageCount(error=0) self.assertMessageCount(error=0)
@test.create_stubs({api.keystone: ('domain_list', @test.create_stubs({api.keystone: ('domain_get',
'domain_list',
'domain_update')}) 'domain_update')})
def test_enable(self): def test_enable(self):
domain = self.domains.get(id="2") domain = self.domains.get(id="2")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list()) api.keystone.domain_list(IgnoreArg()).AndReturn(self.domains.list())
api.keystone.domain_update(IsA(http.HttpRequest), api.keystone.domain_update(IsA(http.HttpRequest),
description=domain.description, description=domain.description,

View File

@ -43,11 +43,14 @@ class GroupsViewTests(test.BaseAdminViewTests):
if group.domain_id == domain_id] if group.domain_id == domain_id]
return groups return groups
@test.create_stubs({api.keystone: ('group_list',)}) @test.create_stubs({api.keystone: ('domain_get',
'group_list',)})
def test_index(self): def test_index(self):
domain_id = self._get_domain_id() domain_id = self._get_domain_id()
groups = self._get_groups(domain_id) groups = self._get_groups(domain_id)
domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.group_list(IgnoreArg(), domain=domain_id) \ api.keystone.group_list(IgnoreArg(), domain=domain_id) \
.AndReturn(groups) .AndReturn(groups)
@ -93,12 +96,14 @@ class GroupsViewTests(test.BaseAdminViewTests):
self.assertContains(res, 'Edit') self.assertContains(res, 'Edit')
self.assertContains(res, 'Delete Group') self.assertContains(res, 'Delete Group')
@test.create_stubs({api.keystone: ('group_list', @test.create_stubs({api.keystone: ('domain_get',
'group_list',
'keystone_can_edit_group')}) 'keystone_can_edit_group')})
def test_index_with_keystone_can_edit_group_false(self): def test_index_with_keystone_can_edit_group_false(self):
domain_id = self._get_domain_id() domain_id = self._get_domain_id()
groups = self._get_groups(domain_id) groups = self._get_groups(domain_id)
domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.group_list(IgnoreArg(), domain=domain_id) \ api.keystone.group_list(IgnoreArg(), domain=domain_id) \
.AndReturn(groups) .AndReturn(groups)
api.keystone.keystone_can_edit_group() \ api.keystone.keystone_can_edit_group() \
@ -115,11 +120,15 @@ class GroupsViewTests(test.BaseAdminViewTests):
self.assertNotContains(res, 'Edit') self.assertNotContains(res, 'Edit')
self.assertNotContains(res, 'Delete Group') self.assertNotContains(res, 'Delete Group')
@test.create_stubs({api.keystone: ('group_create', )}) @test.create_stubs({api.keystone: ('group_create',
'domain_get')})
def test_create(self): def test_create(self):
domain_id = self._get_domain_id() domain_id = self._get_domain_id()
domain = self.domains.get(id="1")
group = self.groups.get(id="1") group = self.groups.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1') \
.AndReturn(domain)
api.keystone.group_create(IsA(http.HttpRequest), api.keystone.group_create(IsA(http.HttpRequest),
description=group.description, description=group.description,
domain_id=domain_id, domain_id=domain_id,
@ -135,11 +144,28 @@ class GroupsViewTests(test.BaseAdminViewTests):
self.assertNoFormErrors(res) self.assertNoFormErrors(res)
self.assertMessageCount(success=1) self.assertMessageCount(success=1)
@test.create_stubs({api.keystone: ('group_create',)})
def test_create_with_domain(self): def test_create_with_domain(self):
domain = self.domains.get(id="1") domain = self.domains.get(id="1")
group = self.groups.get(id="1")
self.setSessionValues(domain_context=domain.id, self.setSessionValues(domain_context=domain.id,
domain_context_name=domain.name) domain_context_name=domain.name)
self.test_create()
api.keystone.group_create(IsA(http.HttpRequest),
description=group.description,
domain_id=domain.id,
name=group.name).AndReturn(group)
self.mox.ReplayAll()
formData = {'method': 'CreateGroupForm',
'name': group.name,
'description': group.description}
res = self.client.post(GROUP_CREATE_URL, formData)
self.assertNoFormErrors(res)
self.assertMessageCount(success=1)
@test.create_stubs({api.keystone: ('group_get', @test.create_stubs({api.keystone: ('group_get',
'group_update')}) 'group_update')})
@ -164,12 +190,15 @@ class GroupsViewTests(test.BaseAdminViewTests):
self.assertNoFormErrors(res) self.assertNoFormErrors(res)
@test.create_stubs({api.keystone: ('group_list', @test.create_stubs({api.keystone: ('domain_get',
'group_list',
'group_delete')}) 'group_delete')})
def test_delete_group(self): def test_delete_group(self):
domain_id = self._get_domain_id() domain_id = self._get_domain_id()
group = self.groups.get(id="2") group = self.groups.get(id="2")
domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.group_list(IgnoreArg(), domain=domain_id) \ api.keystone.group_list(IgnoreArg(), domain=domain_id) \
.AndReturn(self.groups.list()) .AndReturn(self.groups.list())
api.keystone.group_delete(IgnoreArg(), group.id) api.keystone.group_delete(IgnoreArg(), group.id)

View File

@ -51,9 +51,12 @@ PROJECT_DETAIL_URL = reverse('horizon:identity:projects:detail', args=[1])
class TenantsViewTests(test.BaseAdminViewTests): class TenantsViewTests(test.BaseAdminViewTests):
@test.create_stubs({api.keystone: ('tenant_list', 'domain_lookup')}) @test.create_stubs({api.keystone: ('domain_get',
'tenant_list',
'domain_lookup')})
def test_index(self): def test_index(self):
domain = self.domains.get(id="1") domain = self.domains.get(id="1")
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.tenant_list(IsA(http.HttpRequest), api.keystone.tenant_list(IsA(http.HttpRequest),
domain=None, domain=None,
paginate=True, paginate=True,
@ -1374,7 +1377,8 @@ class UpdateProjectWorkflowTests(test.BaseAdminViewTests):
'get_effective_domain_id'), 'get_effective_domain_id'),
quotas: ('get_tenant_quota_data', quotas: ('get_tenant_quota_data',
'get_disabled_quotas', 'get_disabled_quotas',
'tenant_quota_usages')}) 'tenant_quota_usages'),
api.nova: ('tenant_quota_update',)})
def test_update_project_member_update_error(self): def test_update_project_member_update_error(self):
keystone_api_version = api.keystone.VERSIONS.active keystone_api_version = api.keystone.VERSIONS.active
@ -1394,7 +1398,7 @@ class UpdateProjectWorkflowTests(test.BaseAdminViewTests):
admin=True) \ admin=True) \
.AndReturn(project) .AndReturn(project)
api.keystone.domain_get(IsA(http.HttpRequest), domain_id) \ api.keystone.domain_get(IsA(http.HttpRequest), domain_id) \
.AndReturn(self.domain) .MultipleTimes().AndReturn(self.domain)
quotas.get_disabled_quotas(IsA(http.HttpRequest)) \ quotas.get_disabled_quotas(IsA(http.HttpRequest)) \
.AndReturn(self.disabled_quotas.first()) .AndReturn(self.disabled_quotas.first())
quotas.get_tenant_quota_data(IsA(http.HttpRequest), quotas.get_tenant_quota_data(IsA(http.HttpRequest),
@ -1457,6 +1461,8 @@ class UpdateProjectWorkflowTests(test.BaseAdminViewTests):
self._check_role_list(keystone_api_version, role_assignments, groups, self._check_role_list(keystone_api_version, role_assignments, groups,
proj_users, roles, workflow_data) proj_users, roles, workflow_data)
api.nova.tenant_quota_update(IsA(http.HttpRequest), project.id,
**updated_quota)
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -637,7 +637,8 @@ class UsersViewTests(test.BaseAdminViewTests):
res, "form", 'password', res, "form", 'password',
['Password must be between 8 and 18 characters.']) ['Password must be between 8 and 18 characters.'])
@test.create_stubs({api.keystone: ('user_update_enabled', @test.create_stubs({api.keystone: ('domain_get',
'user_update_enabled',
'user_list', 'user_list',
'domain_lookup')}) 'domain_lookup')})
def test_enable_user(self): def test_enable_user(self):
@ -647,6 +648,7 @@ class UsersViewTests(test.BaseAdminViewTests):
users = self._get_users(domain_id) users = self._get_users(domain_id)
user.enabled = False user.enabled = False
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.user_list(IgnoreArg(), domain=domain_id).AndReturn(users) api.keystone.user_list(IgnoreArg(), domain=domain_id).AndReturn(users)
api.keystone.user_update_enabled(IgnoreArg(), api.keystone.user_update_enabled(IgnoreArg(),
user.id, user.id,
@ -661,7 +663,8 @@ class UsersViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, USERS_INDEX_URL) self.assertRedirectsNoFollow(res, USERS_INDEX_URL)
@test.create_stubs({api.keystone: ('user_update_enabled', @test.create_stubs({api.keystone: ('domain_get',
'user_update_enabled',
'user_list', 'user_list',
'domain_lookup')}) 'domain_lookup')})
def test_disable_user(self): def test_disable_user(self):
@ -672,6 +675,7 @@ class UsersViewTests(test.BaseAdminViewTests):
self.assertTrue(user.enabled) self.assertTrue(user.enabled)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.user_list(IgnoreArg(), domain=domain_id) \ api.keystone.user_list(IgnoreArg(), domain=domain_id) \
.AndReturn(users) .AndReturn(users)
api.keystone.user_update_enabled(IgnoreArg(), api.keystone.user_update_enabled(IgnoreArg(),
@ -687,7 +691,8 @@ class UsersViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, USERS_INDEX_URL) self.assertRedirectsNoFollow(res, USERS_INDEX_URL)
@test.create_stubs({api.keystone: ('user_update_enabled', @test.create_stubs({api.keystone: ('domain_get',
'user_update_enabled',
'user_list', 'user_list',
'domain_lookup')}) 'domain_lookup')})
def test_enable_disable_user_exception(self): def test_enable_disable_user_exception(self):
@ -697,6 +702,7 @@ class UsersViewTests(test.BaseAdminViewTests):
users = self._get_users(domain_id) users = self._get_users(domain_id)
user.enabled = False user.enabled = False
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.user_list(IgnoreArg(), domain=domain_id) \ api.keystone.user_list(IgnoreArg(), domain=domain_id) \
.AndReturn(users) .AndReturn(users)
api.keystone.user_update_enabled(IgnoreArg(), user.id, True) \ api.keystone.user_update_enabled(IgnoreArg(), user.id, True) \
@ -710,11 +716,14 @@ class UsersViewTests(test.BaseAdminViewTests):
self.assertRedirectsNoFollow(res, USERS_INDEX_URL) self.assertRedirectsNoFollow(res, USERS_INDEX_URL)
@test.create_stubs({api.keystone: ('user_list', 'domain_lookup')}) @test.create_stubs({api.keystone: ('domain_get',
'user_list',
'domain_lookup')})
def test_disabling_current_user(self): def test_disabling_current_user(self):
domain = self._get_default_domain() domain = self._get_default_domain()
domain_id = domain.id domain_id = domain.id
users = self._get_users(domain_id) users = self._get_users(domain_id)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
for i in range(0, 2): for i in range(0, 2):
api.keystone.user_list(IgnoreArg(), domain=domain_id) \ api.keystone.user_list(IgnoreArg(), domain=domain_id) \
.AndReturn(users) .AndReturn(users)
@ -730,7 +739,9 @@ class UsersViewTests(test.BaseAdminViewTests):
u'You are not allowed to disable user: ' u'You are not allowed to disable user: '
u'test_user') u'test_user')
@test.create_stubs({api.keystone: ('user_list', 'domain_lookup')}) @test.create_stubs({api.keystone: ('domain_get',
'user_list',
'domain_lookup')})
def test_disabling_current_user_domain_name(self): def test_disabling_current_user_domain_name(self):
domain = self._get_default_domain() domain = self._get_default_domain()
domains = self.domains.list() domains = self.domains.list()
@ -738,6 +749,7 @@ class UsersViewTests(test.BaseAdminViewTests):
users = self._get_users(domain_id) users = self._get_users(domain_id)
domain_lookup = dict((d.id, d.name) for d in domains) domain_lookup = dict((d.id, d.name) for d in domains)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
for u in users: for u in users:
u.domain_name = domain_lookup.get(u.domain_id) u.domain_name = domain_lookup.get(u.domain_id)
@ -755,11 +767,14 @@ class UsersViewTests(test.BaseAdminViewTests):
u'You are not allowed to disable user: ' u'You are not allowed to disable user: '
u'test_user') u'test_user')
@test.create_stubs({api.keystone: ('user_list', 'domain_lookup')}) @test.create_stubs({api.keystone: ('domain_get',
'user_list',
'domain_lookup')})
def test_delete_user_with_improper_permissions(self): def test_delete_user_with_improper_permissions(self):
domain = self._get_default_domain() domain = self._get_default_domain()
domain_id = domain.id domain_id = domain.id
users = self._get_users(domain_id) users = self._get_users(domain_id)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
for i in range(0, 2): for i in range(0, 2):
api.keystone.user_list(IgnoreArg(), domain=domain_id) \ api.keystone.user_list(IgnoreArg(), domain=domain_id) \
.AndReturn(users) .AndReturn(users)
@ -775,7 +790,9 @@ class UsersViewTests(test.BaseAdminViewTests):
u'You are not allowed to delete user: %s' u'You are not allowed to delete user: %s'
% self.request.user.username) % self.request.user.username)
@test.create_stubs({api.keystone: ('user_list', 'domain_lookup')}) @test.create_stubs({api.keystone: ('domain_get',
'user_list',
'domain_lookup')})
def test_delete_user_with_improper_permissions_domain_name(self): def test_delete_user_with_improper_permissions_domain_name(self):
domain = self._get_default_domain() domain = self._get_default_domain()
domains = self.domains.list() domains = self.domains.list()
@ -783,6 +800,7 @@ class UsersViewTests(test.BaseAdminViewTests):
users = self._get_users(domain_id) users = self._get_users(domain_id)
domain_lookup = dict((d.id, d.name) for d in domains) domain_lookup = dict((d.id, d.name) for d in domains)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
for u in users: for u in users:
u.domain_name = domain_lookup.get(u.domain_id) u.domain_name = domain_lookup.get(u.domain_id)
@ -800,11 +818,15 @@ class UsersViewTests(test.BaseAdminViewTests):
u'You are not allowed to delete user: %s' u'You are not allowed to delete user: %s'
% self.request.user.username) % self.request.user.username)
@test.create_stubs({api.keystone: ('user_get', 'tenant_get')}) @test.create_stubs({api.keystone: ('domain_get',
'user_get',
'tenant_get')})
def test_detail_view(self): def test_detail_view(self):
domain = self._get_default_domain()
user = self.users.get(id="1") user = self.users.get(id="1")
tenant = self.tenants.get(id=user.project_id) tenant = self.tenants.get(id=user.project_id)
api.keystone.domain_get(IsA(http.HttpRequest), '1').AndReturn(domain)
api.keystone.user_get(IsA(http.HttpRequest), '1').AndReturn(user) api.keystone.user_get(IsA(http.HttpRequest), '1').AndReturn(user)
api.keystone.tenant_get(IsA(http.HttpRequest), user.project_id) \ api.keystone.tenant_get(IsA(http.HttpRequest), user.project_id) \
.AndReturn(tenant) .AndReturn(tenant)

View File

@ -169,7 +169,8 @@ class FloatingIpViewTests(test.TestCase):
api.network: ('floating_ip_disassociate', api.network: ('floating_ip_disassociate',
'floating_ip_supported', 'floating_ip_supported',
'tenant_floating_ip_get', 'tenant_floating_ip_get',
'tenant_floating_ip_list',)}) 'tenant_floating_ip_list',),
api.neutron: ('is_extension_supported',)})
def test_disassociate_post(self): def test_disassociate_post(self):
floating_ip = self.floating_ips.first() floating_ip = self.floating_ips.first()
@ -179,6 +180,9 @@ class FloatingIpViewTests(test.TestCase):
.AndReturn(True) .AndReturn(True)
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \ api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(self.floating_ips.list()) .AndReturn(self.floating_ips.list())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.network.floating_ip_disassociate(IsA(http.HttpRequest), api.network.floating_ip_disassociate(IsA(http.HttpRequest),
floating_ip.id) floating_ip.id)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -192,7 +196,8 @@ class FloatingIpViewTests(test.TestCase):
api.network: ('floating_ip_disassociate', api.network: ('floating_ip_disassociate',
'floating_ip_supported', 'floating_ip_supported',
'tenant_floating_ip_get', 'tenant_floating_ip_get',
'tenant_floating_ip_list',)}) 'tenant_floating_ip_list',),
api.neutron: ('is_extension_supported',)})
def test_disassociate_post_with_exception(self): def test_disassociate_post_with_exception(self):
floating_ip = self.floating_ips.first() floating_ip = self.floating_ips.first()
@ -202,6 +207,9 @@ class FloatingIpViewTests(test.TestCase):
.AndReturn(True) .AndReturn(True)
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \ api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(self.floating_ips.list()) .AndReturn(self.floating_ips.list())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation')\
.AndReturn(True)
api.network.floating_ip_disassociate(IsA(http.HttpRequest), api.network.floating_ip_disassociate(IsA(http.HttpRequest),
floating_ip.id) \ floating_ip.id) \

View File

@ -39,49 +39,48 @@ class AccessAndSecurityTests(test.TestCase):
@test.create_stubs({api.network: ('floating_ip_supported', @test.create_stubs({api.network: ('floating_ip_supported',
'tenant_floating_ip_list', 'tenant_floating_ip_list',
'floating_ip_pools_list',
'security_group_list',), 'security_group_list',),
api.nova: ('keypair_list', api.nova: ('keypair_list',
'server_list',), 'server_list',),
api.base: ('is_service_enabled',), api.base: ('is_service_enabled',),
quotas: ('tenant_quota_usages',)}) quotas: ('tenant_quota_usages',),
api.keystone: ('list_ec2_credentials',)})
def _test_index(self, ec2_enabled=True, instanceless_ips=False): def _test_index(self, ec2_enabled=True, instanceless_ips=False):
keypairs = self.keypairs.list() keypairs = self.keypairs.list()
sec_groups = self.security_groups.list() sec_groups = self.security_groups.list()
floating_ips = self.floating_ips.list() floating_ips = self.floating_ips.list()
floating_pools = self.pools.list()
if instanceless_ips: if instanceless_ips:
for fip in floating_ips: for fip in floating_ips:
fip.instance_id = None fip.instance_id = None
quota_data = self.quota_usages.first() quota_data = self.quota_usages.first()
quota_data['security_groups']['available'] = 10 quota_data['security_groups']['available'] = 10
if not instanceless_ips: api.network.floating_ip_supported(IsA(http.HttpRequest)) \
api.nova.server_list(
IsA(http.HttpRequest)) \
.AndReturn([self.servers.list(), False])
api.nova.keypair_list(
IsA(http.HttpRequest)) \
.AndReturn(keypairs)
api.network.floating_ip_supported(
IsA(http.HttpRequest)) \
.AndReturn(True) .AndReturn(True)
api.network.tenant_floating_ip_list( if not instanceless_ips:
IsA(http.HttpRequest)) \ api.nova.server_list(IsA(http.HttpRequest)) \
.AndReturn([self.servers.list(), False])
api.nova.keypair_list(IsA(http.HttpRequest)) \
.AndReturn(keypairs)
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(floating_ips) .AndReturn(floating_ips)
api.network.security_group_list( api.network.floating_ip_pools_list(IsA(http.HttpRequest)) \
IsA(http.HttpRequest)) \ .AndReturn(floating_pools)
api.network.security_group_list(IsA(http.HttpRequest)) \
.AndReturn(sec_groups) .AndReturn(sec_groups)
quotas.tenant_quota_usages( quotas.tenant_quota_usages(IsA(http.HttpRequest)).MultipleTimes() \
IsA(http.HttpRequest)).MultipleTimes() \
.AndReturn(quota_data) .AndReturn(quota_data)
api.base.is_service_enabled( api.base.is_service_enabled(IsA(http.HttpRequest), 'network') \
IsA(http.HttpRequest), .MultipleTimes().AndReturn(True)
'network').MultipleTimes() \ api.base.is_service_enabled(IsA(http.HttpRequest), 'ec2') \
.AndReturn(True) .MultipleTimes().AndReturn(ec2_enabled)
api.base.is_service_enabled( if ec2_enabled:
IsA(http.HttpRequest), api.keystone.list_ec2_credentials(IsA(http.HttpRequest),
'ec2').MultipleTimes() \ self.user.id)\
.AndReturn(ec2_enabled) .AndReturn(self.ec2.list())
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -1002,8 +1002,18 @@ class InstanceTests(helpers.TestCase):
res = self._get_instance_details(server) res = self._get_instance_details(server)
self.assertItemsEqual(res.context['instance'].fault, server.fault) self.assertItemsEqual(res.context['instance'].fault, server.fault)
@helpers.create_stubs({console: ('get_console',)})
def test_instance_details_console_tab(self): def test_instance_details_console_tab(self):
server = self.servers.first() server = self.servers.first()
CONSOLE_OUTPUT = '/vncserver'
CONSOLE_TITLE = '&title=%s(%s)' % (server.name, server.id)
CONSOLE_URL = CONSOLE_OUTPUT + CONSOLE_TITLE
console_mock = self.mox.CreateMock(api.nova.VNCConsole)
console_mock.url = CONSOLE_OUTPUT
console.get_console(IgnoreArg(), 'AUTO', server) \
.AndReturn(('VNC', CONSOLE_URL))
tg = tabs.InstanceDetailTabs(self.request, instance=server) tg = tabs.InstanceDetailTabs(self.request, instance=server)
qs = "?%s=%s" % (tg.param_name, tg.get_tab("console").get_id()) qs = "?%s=%s" % (tg.param_name, tg.get_tab("console").get_id())
@ -2791,12 +2801,14 @@ class InstanceTests(helpers.TestCase):
api.neutron: ('network_list', api.neutron: ('network_list',
'profile_list', 'profile_list',
'port_create', 'port_create',
'port_list'), 'port_list',
'is_port_profiles_supported'),
api.nova: ('extension_supported', api.nova: ('extension_supported',
'flavor_list', 'flavor_list',
'keypair_list', 'keypair_list',
'availability_zone_list', 'availability_zone_list',
'server_create',), 'server_create',
'tenant_absolute_limits'),
api.network: ('security_group_list',), api.network: ('security_group_list',),
cinder: ('volume_list', cinder: ('volume_list',
'volume_snapshot_list',), 'volume_snapshot_list',),
@ -2811,11 +2823,6 @@ class InstanceTests(helpers.TestCase):
avail_zone = self.availability_zones.first() avail_zone = self.availability_zones.first()
quota_usages = self.quota_usages.first() quota_usages = self.quota_usages.first()
api.nova.extension_supported('BlockDeviceMappingV2Boot',
IsA(http.HttpRequest)) \
.AndReturn(True)
api.nova.flavor_list(IsA(http.HttpRequest)) \
.AndReturn(self.flavors.list())
api.glance.image_list_detailed(IsA(http.HttpRequest), api.glance.image_list_detailed(IsA(http.HttpRequest),
filters={'is_public': True, filters={'is_public': True,
'status': 'active'}) \ 'status': 'active'}) \
@ -2825,6 +2832,12 @@ class InstanceTests(helpers.TestCase):
filters={'property-owner_id': self.tenant.id, filters={'property-owner_id': self.tenant.id,
'status': 'active'}) \ 'status': 'active'}) \
.AndReturn([[], False, False]) .AndReturn([[], False, False])
quotas.tenant_quota_usages(IsA(http.HttpRequest)) \
.AndReturn(quota_usages)
api.neutron.is_port_profiles_supported()\
.MultipleTimes().AndReturn(test_with_profile)
api.neutron.network_list(IsA(http.HttpRequest), api.neutron.network_list(IsA(http.HttpRequest),
tenant_id=self.tenant.id, tenant_id=self.tenant.id,
shared=False) \ shared=False) \
@ -2843,30 +2856,14 @@ class InstanceTests(helpers.TestCase):
api.neutron.port_list(IsA(http.HttpRequest), api.neutron.port_list(IsA(http.HttpRequest),
network_id=net.id) \ network_id=net.id) \
.AndReturn(self.ports.list()) .AndReturn(self.ports.list())
api.nova.flavor_list(IsA(http.HttpRequest)) \
.AndReturn(self.flavors.list())
api.nova.keypair_list(IsA(http.HttpRequest)) \
.AndReturn(self.keypairs.list())
api.network.security_group_list(IsA(http.HttpRequest)) \
.AndReturn(self.security_groups.list())
api.nova.availability_zone_list(IsA(http.HttpRequest)) \
.AndReturn(self.availability_zones.list())
api.nova.extension_supported('DiskConfig', api.nova.extension_supported('DiskConfig',
IsA(http.HttpRequest)) \ IsA(http.HttpRequest)) \
.AndReturn(True) .AndReturn(True)
api.nova.extension_supported('ConfigDrive', api.nova.extension_supported('ConfigDrive',
IsA(http.HttpRequest)).AndReturn(True) IsA(http.HttpRequest)).AndReturn(True)
cinder.volume_list(IsA(http.HttpRequest), api.nova.tenant_absolute_limits(IsA(http.HttpRequest)) \
search_opts=VOLUME_SEARCH_OPTS) \ .MultipleTimes().AndReturn(self.limits['absolute'])
.AndReturn([])
cinder.volume_snapshot_list(IsA(http.HttpRequest),
search_opts=SNAPSHOT_SEARCH_OPTS) \
.AndReturn([])
quotas.tenant_quota_usages(IsA(http.HttpRequest)) \
.AndReturn(quota_usages)
self.mox.ReplayAll() self.mox.ReplayAll()
bad_snapshot_id = 'a-bogus-id' bad_snapshot_id = 'a-bogus-id'
@ -2890,7 +2887,7 @@ class InstanceTests(helpers.TestCase):
url = reverse('horizon:project:instances:launch') url = reverse('horizon:project:instances:launch')
res = self.client.post(url, form_data) res = self.client.post(url, form_data)
self.assertFormErrors(res, 1, "You must select a snapshot.") self.assertFormErrors(res, 3, "You must select a snapshot.")
@helpers.create_stubs({api.glance: ('image_list_detailed',), @helpers.create_stubs({api.glance: ('image_list_detailed',),
api.neutron: ('network_list', api.neutron: ('network_list',

View File

@ -89,11 +89,16 @@ class LoadBalancerTests(test.TestCase):
@test.create_stubs({api.lbaas: ('pool_list', 'member_list', @test.create_stubs({api.lbaas: ('pool_list', 'member_list',
'pool_health_monitor_list'), 'pool_health_monitor_list'),
api.network: ('floating_ip_supported', api.network: ('floating_ip_supported',
'floating_ip_simple_associate_supported') 'floating_ip_simple_associate_supported',
'tenant_floating_ip_list')
}) })
def test_index_pools(self): def test_index_pools(self):
fips = self.floating_ips.list()
self.set_up_expect() self.set_up_expect()
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)).\
AndReturn(fips)
self.mox.ReplayAll() self.mox.ReplayAll()
res = self.client.get(self.INDEX_URL) res = self.client.get(self.INDEX_URL)
@ -107,11 +112,16 @@ class LoadBalancerTests(test.TestCase):
@test.create_stubs({api.lbaas: ('pool_list', 'member_list', @test.create_stubs({api.lbaas: ('pool_list', 'member_list',
'pool_health_monitor_list'), 'pool_health_monitor_list'),
api.network: ('floating_ip_supported', api.network: ('floating_ip_supported',
'floating_ip_simple_associate_supported') 'floating_ip_simple_associate_supported',
'tenant_floating_ip_list')
}) })
def test_index_members(self): def test_index_members(self):
fips = self.floating_ips.list()
self.set_up_expect() self.set_up_expect()
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)).\
AndReturn(fips)
self.mox.ReplayAll() self.mox.ReplayAll()
res = self.client.get(self.INDEX_URL + '?tab=lbtabs__members') res = self.client.get(self.INDEX_URL + '?tab=lbtabs__members')
@ -125,11 +135,16 @@ class LoadBalancerTests(test.TestCase):
@test.create_stubs({api.lbaas: ('pool_list', 'member_list', @test.create_stubs({api.lbaas: ('pool_list', 'member_list',
'pool_health_monitor_list'), 'pool_health_monitor_list'),
api.network: ('floating_ip_supported', api.network: ('floating_ip_supported',
'floating_ip_simple_associate_supported') 'floating_ip_simple_associate_supported',
'tenant_floating_ip_list')
}) })
def test_index_monitors(self): def test_index_monitors(self):
fips = self.floating_ips.list()
self.set_up_expect() self.set_up_expect()
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)).\
AndReturn(fips)
self.mox.ReplayAll() self.mox.ReplayAll()
res = self.client.get(self.INDEX_URL + '?tab=lbtabs__monitors') res = self.client.get(self.INDEX_URL + '?tab=lbtabs__monitors')
@ -942,16 +957,20 @@ class LoadBalancerTests(test.TestCase):
'<DeletePMAssociationStep: deletepmassociationaction>', ] '<DeletePMAssociationStep: deletepmassociationaction>', ]
self.assertQuerysetEqual(workflow.steps, expected_objs) self.assertQuerysetEqual(workflow.steps, expected_objs)
@test.create_stubs({api.lbaas: ('pool_list', 'pool_delete')}) @test.create_stubs({api.lbaas: ('pool_list', 'pool_delete'),
api.network: ('tenant_floating_ip_list',)})
def test_delete_pool(self): def test_delete_pool(self):
pool_list = self.pools.list() pool_list = self.pools.list()
pool = pool_list[0] pool = pool_list[0]
fips = self.floating_ips.list()
# the test pool needs to have no vip # the test pool needs to have no vip
# in order to be able to be deleted # in order to be able to be deleted
pool.vip_id = None pool.vip_id = None
api.lbaas.pool_list( api.lbaas.pool_list(
IsA(http.HttpRequest), tenant_id=self.tenant.id) \ IsA(http.HttpRequest), tenant_id=self.tenant.id) \
.AndReturn(pool_list) .AndReturn(pool_list)
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)).\
AndReturn(fips)
api.lbaas.pool_delete(IsA(http.HttpRequest), pool.id) api.lbaas.pool_delete(IsA(http.HttpRequest), pool.id)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -961,13 +980,16 @@ class LoadBalancerTests(test.TestCase):
self.assertNoFormErrors(res) self.assertNoFormErrors(res)
@test.create_stubs({api.lbaas: ('pool_list', 'pool_get', @test.create_stubs({api.lbaas: ('pool_list', 'pool_get',
'vip_delete'), 'member_list', 'vip_delete',
'pool_health_monitor_list'),
api.network: ( api.network: (
'tenant_floating_ip_list',
'floating_ip_supported', 'floating_ip_supported',
'floating_ip_simple_associate_supported')}) 'floating_ip_simple_associate_supported')})
def test_delete_vip(self): def test_delete_vip(self):
pool = self.pools.first() pool = self.pools.first()
vip = self.vips.first() vip = self.vips.first()
fips = self.floating_ips.list()
api.lbaas.pool_list( api.lbaas.pool_list(
IsA(http.HttpRequest), tenant_id=self.tenant.id) \ IsA(http.HttpRequest), tenant_id=self.tenant.id) \
.AndReturn(self.pools.list()) .AndReturn(self.pools.list())
@ -976,6 +998,8 @@ class LoadBalancerTests(test.TestCase):
.AndReturn(True) .AndReturn(True)
api.network.floating_ip_simple_associate_supported(IgnoreArg()) \ api.network.floating_ip_simple_associate_supported(IgnoreArg()) \
.MultipleTimes().AndReturn(True) .MultipleTimes().AndReturn(True)
api.network.tenant_floating_ip_list(IsA(http.HttpRequest)).\
AndReturn(fips)
api.lbaas.vip_delete(IsA(http.HttpRequest), vip.id) api.lbaas.vip_delete(IsA(http.HttpRequest), vip.id)
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -16,10 +16,12 @@ from django.core.urlresolvers import reverse
from django import http from django import http
import django.test import django.test
from mox3.mox import IgnoreArg # noqa
from mox3.mox import IsA # noqa from mox3.mox import IsA # noqa
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from openstack_dashboard import api from openstack_dashboard import api
from openstack_dashboard.dashboards.project.instances import console
from openstack_dashboard.dashboards.project.network_topology.views import \ from openstack_dashboard.dashboards.project.network_topology.views import \
TranslationHelper TranslationHelper
from openstack_dashboard.test import helpers as test from openstack_dashboard.test import helpers as test
@ -36,7 +38,8 @@ class NetworkTopologyTests(test.TestCase):
api.neutron: ('network_list_for_tenant', api.neutron: ('network_list_for_tenant',
'network_list', 'network_list',
'router_list', 'router_list',
'port_list')}) 'port_list',),
console: ('get_console',)})
def test_json_view(self): def test_json_view(self):
self._test_json_view() self._test_json_view()
@ -44,13 +47,15 @@ class NetworkTopologyTests(test.TestCase):
OPENSTACK_NEUTRON_NETWORK={'enable_router': False}) OPENSTACK_NEUTRON_NETWORK={'enable_router': False})
@test.create_stubs({api.nova: ('server_list',), @test.create_stubs({api.nova: ('server_list',),
api.neutron: ('network_list_for_tenant', api.neutron: ('network_list_for_tenant',
'port_list')}) 'port_list'),
console: ('get_console',)})
def test_json_view_router_disabled(self): def test_json_view_router_disabled(self):
self._test_json_view(router_enable=False) self._test_json_view(router_enable=False)
def _test_json_view(self, router_enable=True): def _test_json_view(self, router_enable=True):
api.nova.server_list( api.nova.server_list(
IsA(http.HttpRequest)).AndReturn([self.servers.list(), False]) IsA(http.HttpRequest)).AndReturn([self.servers.list(), False])
tenant_networks = [net for net in self.networks.list() tenant_networks = [net for net in self.networks.list()
if not net['router:external']] if not net['router:external']]
external_networks = [net for net in self.networks.list() external_networks = [net for net in self.networks.list()
@ -58,10 +63,17 @@ class NetworkTopologyTests(test.TestCase):
api.neutron.network_list_for_tenant( api.neutron.network_list_for_tenant(
IsA(http.HttpRequest), IsA(http.HttpRequest),
self.tenant.id).AndReturn(tenant_networks) self.tenant.id).AndReturn(tenant_networks)
if router_enable:
api.neutron.network_list( for server in self.servers.list():
IsA(http.HttpRequest), if server.status != u'BUILD':
**{'router:external': True}).AndReturn(external_networks) CONSOLE_OUTPUT = '/vncserver'
CONSOLE_TITLE = '&title=%s' % server.id
CONSOLE_URL = CONSOLE_OUTPUT + CONSOLE_TITLE
console_mock = self.mox.CreateMock(api.nova.VNCConsole)
console_mock.url = CONSOLE_OUTPUT
console.get_console(IsA(http.HttpRequest), 'AUTO', server) \
.AndReturn(('VNC', CONSOLE_URL))
# router1 : gateway port not in the port list # router1 : gateway port not in the port list
# router2 : no gateway port # router2 : no gateway port
@ -73,24 +85,34 @@ class NetworkTopologyTests(test.TestCase):
tenant_id=self.tenant.id).AndReturn(routers) tenant_id=self.tenant.id).AndReturn(routers)
api.neutron.port_list( api.neutron.port_list(
IsA(http.HttpRequest)).AndReturn(self.ports.list()) IsA(http.HttpRequest)).AndReturn(self.ports.list())
if router_enable:
api.neutron.network_list(
IsA(http.HttpRequest),
**{'router:external': True}).AndReturn(external_networks)
self.mox.ReplayAll() self.mox.ReplayAll()
res = self.client.get(JSON_URL) res = self.client.get(JSON_URL)
self.assertEqual('text/json', res['Content-Type']) self.assertEqual('text/json', res['Content-Type'])
data = jsonutils.loads(res.content) data = jsonutils.loads(res.content)
# servers # servers
# result_server_urls = [(server['id'], server['url']) # result_server_urls = [(server['id'], server['url'])
# for server in data['servers']] # for server in data['servers']]
expect_server_urls = [ expect_server_urls = []
{'id': server.id, for server in self.servers.list():
expect_server = {
'id': server.id,
'name': server.name, 'name': server.name,
'status': self.trans.instance[server.status], 'status': server.status.title(),
'original_status': server.status, 'original_status': server.status,
'task': None, 'task': None,
'url': '/project/instances/%s/' % server.id} 'url': '/project/instances/%s/' % server.id
for server in self.servers.list()] }
if server.status != 'BUILD':
expect_server['console'] = 'vnc'
expect_server_urls.append(expect_server)
self.assertEqual(expect_server_urls, data['servers']) self.assertEqual(expect_server_urls, data['servers'])
# routers # routers
@ -102,7 +124,7 @@ class NetworkTopologyTests(test.TestCase):
'external_gateway_info': 'external_gateway_info':
router.external_gateway_info, router.external_gateway_info,
'name': router.name, 'name': router.name,
'status': self.trans.router[router.status], 'status': router.status.title(),
'original_status': router.status, 'original_status': router.status,
'url': '/project/routers/%s/' % router.id} 'url': '/project/routers/%s/' % router.id}
for router in routers] for router in routers]
@ -117,23 +139,23 @@ class NetworkTopologyTests(test.TestCase):
'url': '/project/networks/%s/detail' % net.id, 'url': '/project/networks/%s/detail' % net.id,
'name': net.name, 'name': net.name,
'router:external': net.router__external, 'router:external': net.router__external,
'status': self.trans.network[net.status], 'status': net.status.title(),
'original_status': net.status, 'original_status': net.status,
'subnets': []} 'subnets': []}
for net in external_networks] for net in external_networks]
expect_net_urls += [{'id': net.id, expect_net_urls.extend([{
'id': net.id,
'url': '/project/networks/%s/detail' % net.id, 'url': '/project/networks/%s/detail' % net.id,
'name': net.name, 'name': net.name,
'router:external': net.router__external, 'router:external': net.router__external,
'status': self.trans.network[net.status], 'status': net.status.title(),
'original_status': net.status, 'original_status': net.status,
'subnets': [{'cidr': subnet.cidr, 'subnets': [{
'cidr': subnet.cidr,
'id': subnet.id, 'id': subnet.id,
'url': 'url': '/project/networks/subnets/%s/detail' % subnet.id}
'/project/networks/subnets/%s/detail'
% subnet.id}
for subnet in net.subnets]} for subnet in net.subnets]}
for net in tenant_networks] for net in tenant_networks])
for exp_net in expect_net_urls: for exp_net in expect_net_urls:
if exp_net['url'] is None: if exp_net['url'] is None:
del exp_net['url'] del exp_net['url']
@ -146,7 +168,7 @@ class NetworkTopologyTests(test.TestCase):
'device_owner': port.device_owner, 'device_owner': port.device_owner,
'fixed_ips': port.fixed_ips, 'fixed_ips': port.fixed_ips,
'network_id': port.network_id, 'network_id': port.network_id,
'status': self.trans.port[port.status], 'status': port.status.title(),
'original_status': port.status, 'original_status': port.status,
'url': '/project/networks/ports/%s/detail' % port.id} 'url': '/project/networks/ports/%s/detail' % port.id}
for port in self.ports.list()] for port in self.ports.list()]

View File

@ -29,23 +29,28 @@ NETWORKS_DETAIL_URL = 'horizon:project:networks:detail'
class NetworkPortTests(test.TestCase): class NetworkPortTests(test.TestCase):
@test.create_stubs({api.neutron: ('port_get', @test.create_stubs({api.neutron: ('network_get',
'port_get',
'is_extension_supported',)}) 'is_extension_supported',)})
def test_port_detail(self): def test_port_detail(self):
self._test_port_detail() self._test_port_detail()
@test.create_stubs({api.neutron: ('port_get', @test.create_stubs({api.neutron: ('network_get',
'port_get',
'is_extension_supported',)}) 'is_extension_supported',)})
def test_port_detail_with_mac_learning(self): def test_port_detail_with_mac_learning(self):
self._test_port_detail(mac_learning=True) self._test_port_detail(mac_learning=True)
def _test_port_detail(self, mac_learning=False): def _test_port_detail(self, mac_learning=False):
port = self.ports.first() port = self.ports.first()
network_id = self.networks.first().id
api.neutron.port_get(IsA(http.HttpRequest), port.id)\ api.neutron.port_get(IsA(http.HttpRequest), port.id)\
.AndReturn(self.ports.first()) .AndReturn(self.ports.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
'mac-learning')\ 'mac-learning')\
.AndReturn(mac_learning) .AndReturn(mac_learning)
api.neutron.network_get(IsA(http.HttpRequest), network_id)\
.AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
'mac-learning')\ 'mac-learning')\
.AndReturn(mac_learning) .AndReturn(mac_learning)

View File

@ -65,12 +65,19 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, NETWORKS_INDEX_URL) self.assertRedirectsNoFollow(res, NETWORKS_INDEX_URL)
@test.create_stubs({api.neutron: ('network_get',)}) @test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_create_get(self): def test_subnet_create_get(self):
network = self.networks.first() network = self.networks.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
url = reverse('horizon:project:networks:addsubnet', url = reverse('horizon:project:networks:addsubnet',
@ -80,13 +87,20 @@ class NetworkSubnetTests(test.TestCase):
self.assertTemplateUsed(res, views.WorkflowView.template_name) self.assertTemplateUsed(res, views.WorkflowView.template_name)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'subnet_create',)}) 'subnet_create',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_create_post(self, test_with_subnetpool=False): def test_subnet_create_post(self, test_with_subnetpool=False):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -108,13 +122,20 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'subnet_create',)}) 'subnet_create',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_create_post_with_additional_attributes(self): def test_subnet_create_post_with_additional_attributes(self):
network = self.networks.list()[1] network = self.networks.list()[1]
subnet = self.subnets.list()[1] subnet = self.subnets.list()[1]
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -138,13 +159,20 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'subnet_create',)}) 'subnet_create',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_create_post_with_additional_attributes_no_gateway(self): def test_subnet_create_post_with_additional_attributes_no_gateway(self):
network = self.networks.first() network = self.networks.first()
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -194,7 +222,9 @@ class NetworkSubnetTests(test.TestCase):
test_with_subnetpool=True) test_with_subnetpool=True)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'subnet_create',)}) 'subnet_create',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_create_post_subnet_exception(self, def test_subnet_create_post_subnet_exception(self,
test_with_subnetpool=False): test_with_subnetpool=False):
network = self.networks.first() network = self.networks.first()
@ -202,6 +232,11 @@ class NetworkSubnetTests(test.TestCase):
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(self.networks.first()) .AndReturn(self.networks.first())
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -347,10 +382,10 @@ class NetworkSubnetTests(test.TestCase):
network.id).AndReturn(network) network.id).AndReturn(network)
api.neutron.is_extension_supported(IsA(http.HttpRequest), api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\ 'subnet_allocation') \
AndReturn(True) .AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\ api.neutron.subnetpool_list(IsA(http.HttpRequest)) \
AndReturn(self.subnetpools.list()) .AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
@ -731,13 +766,20 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('network_get', @test.create_stubs({api.neutron: ('network_get',
'subnet_create',)}) 'subnet_create',
'is_extension_supported',
'subnetpool_list')})
def test_v6subnet_create_post_with_slaac_attributes(self): def test_v6subnet_create_post_with_slaac_attributes(self):
network = self.networks.get(name="v6_net2") network = self.networks.get(name="v6_net2")
subnet = self.subnets.get(name="v6_subnet2") subnet = self.subnets.get(name="v6_subnet2")
api.neutron.network_get(IsA(http.HttpRequest), api.neutron.network_get(IsA(http.HttpRequest),
network.id)\ network.id)\
.AndReturn(network) .AndReturn(network)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
network_id=network.id, network_id=network.id,
name=subnet.name, name=subnet.name,
@ -761,9 +803,16 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post(self): def test_subnet_update_post(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
@ -786,14 +835,19 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_with_gateway_ip(self): def test_subnet_update_post_with_gateway_ip(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .MultipleTimes().AndReturn(subnet)
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet)
gateway_ip = '10.0.0.100' gateway_ip = '10.0.0.100'
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)) \
.AndReturn(self.subnetpools.list())
api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id, api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id,
name=subnet.name, name=subnet.name,
gateway_ip=gateway_ip, gateway_ip=gateway_ip,
@ -814,13 +868,18 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_no_gateway(self): def test_subnet_update_post_no_gateway(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .MultipleTimes().AndReturn(subnet)
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.is_extension_supported(IsA(http.HttpRequest),
.AndReturn(subnet) 'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id, api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id,
name=subnet.name, name=subnet.name,
gateway_ip=None, gateway_ip=None,
@ -841,13 +900,20 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_with_additional_attributes(self): def test_subnet_update_post_with_additional_attributes(self):
subnet = self.subnets.list()[1] subnet = self.subnets.list()[1]
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
start = subnet.allocation_pools[0]['start'] start = subnet.allocation_pools[0]['start']
end = subnet.allocation_pools[0]['end'] end = subnet.allocation_pools[0]['end']
api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id, api.neutron.subnet_update(IsA(http.HttpRequest), subnet.id,
@ -870,11 +936,18 @@ class NetworkSubnetTests(test.TestCase):
self.assertRedirectsNoFollow(res, redir_url) self.assertRedirectsNoFollow(res, redir_url)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_gw_inconsistent(self): def test_subnet_update_post_gw_inconsistent(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# dummy IPv6 address # dummy IPv6 address
@ -888,11 +961,18 @@ class NetworkSubnetTests(test.TestCase):
self.assertContains(res, 'Gateway IP and IP version are inconsistent.') self.assertContains(res, 'Gateway IP and IP version are inconsistent.')
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_invalid_nameservers(self): def test_subnet_update_post_invalid_nameservers(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# invalid DNS server address # invalid DNS server address
@ -908,11 +988,18 @@ class NetworkSubnetTests(test.TestCase):
'(value=%s)' % dns_nameservers[1]) '(value=%s)' % dns_nameservers[1])
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_invalid_routes_destination_only(self): def test_subnet_update_post_invalid_routes_destination_only(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# Start only host_route # Start only host_route
@ -930,11 +1017,18 @@ class NetworkSubnetTests(test.TestCase):
'(value=%s)' % host_routes) '(value=%s)' % host_routes)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_invalid_routes_three_entries(self): def test_subnet_update_post_invalid_routes_three_entries(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# host_route with three entries # host_route with three entries
@ -952,11 +1046,18 @@ class NetworkSubnetTests(test.TestCase):
'(value=%s)' % host_routes) '(value=%s)' % host_routes)
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_invalid_routes_invalid_destination(self): def test_subnet_update_post_invalid_routes_invalid_destination(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# invalid destination network # invalid destination network
@ -973,11 +1074,18 @@ class NetworkSubnetTests(test.TestCase):
'(value=%s)' % host_routes.split(',')[0]) '(value=%s)' % host_routes.split(',')[0])
@test.create_stubs({api.neutron: ('subnet_update', @test.create_stubs({api.neutron: ('subnet_update',
'subnet_get',)}) 'subnet_get',
'is_extension_supported',
'subnetpool_list')})
def test_subnet_update_post_invalid_routes_nexthop_ip_network(self): def test_subnet_update_post_invalid_routes_nexthop_ip_network(self):
subnet = self.subnets.first() subnet = self.subnets.first()
api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\ api.neutron.subnet_get(IsA(http.HttpRequest), subnet.id)\
.AndReturn(subnet) .AndReturn(subnet)
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation') \
.AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
self.mox.ReplayAll() self.mox.ReplayAll()
# nexthop is not an IP address # nexthop is not an IP address

View File

@ -323,9 +323,16 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.assertItemsEqual(subnets, [self.subnets.first()]) self.assertItemsEqual(subnets, [self.subnets.first()])
self.assertEqual(len(ports), 0) self.assertEqual(len(ports), 0)
@test.create_stubs({api.neutron: ('profile_list',)}) @test.create_stubs({api.neutron: ('profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_get(self, def test_network_create_get(self,
test_with_profile=False): test_with_profile=False):
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
if test_with_profile: if test_with_profile:
net_profiles = self.net_profiles.list() net_profiles = self.net_profiles.list()
api.neutron.profile_list(IsA(http.HttpRequest), api.neutron.profile_list(IsA(http.HttpRequest),
@ -349,7 +356,9 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.test_network_create_get(test_with_profile=True) self.test_network_create_get(test_with_profile=True)
@test.create_stubs({api.neutron: ('network_create', @test.create_stubs({api.neutron: ('network_create',
'profile_list',)}) 'profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_post(self, def test_network_create_post(self,
test_with_profile=False): test_with_profile=False):
network = self.networks.first() network = self.networks.first()
@ -362,6 +371,11 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
api.neutron.profile_list(IsA(http.HttpRequest), api.neutron.profile_list(IsA(http.HttpRequest),
'network').AndReturn(net_profiles) 'network').AndReturn(net_profiles)
params['net_profile_id'] = net_profile_id params['net_profile_id'] = net_profile_id
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.network_create(IsA(http.HttpRequest), api.neutron.network_create(IsA(http.HttpRequest),
**params).AndReturn(network) **params).AndReturn(network)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -381,7 +395,9 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.assertRedirectsNoFollow(res, INDEX_URL) self.assertRedirectsNoFollow(res, INDEX_URL)
@test.create_stubs({api.neutron: ('network_create', @test.create_stubs({api.neutron: ('network_create',
'profile_list',)}) 'profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_post_with_shared(self, test_with_profile=False): def test_network_create_post_with_shared(self, test_with_profile=False):
network = self.networks.first() network = self.networks.first()
params = {'name': network.name, params = {'name': network.name,
@ -393,6 +409,11 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
api.neutron.profile_list(IsA(http.HttpRequest), api.neutron.profile_list(IsA(http.HttpRequest),
'network').AndReturn(net_profiles) 'network').AndReturn(net_profiles)
params['net_profile_id'] = net_profile_id params['net_profile_id'] = net_profile_id
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.network_create(IsA(http.HttpRequest), api.neutron.network_create(IsA(http.HttpRequest),
**params).AndReturn(network) **params).AndReturn(network)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -418,7 +439,9 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
@test.create_stubs({api.neutron: ('network_create', @test.create_stubs({api.neutron: ('network_create',
'subnet_create', 'subnet_create',
'profile_list',)}) 'profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_post_with_subnet(self, def test_network_create_post_with_subnet(self,
test_with_profile=False, test_with_profile=False,
test_with_ipv6=True): test_with_ipv6=True):
@ -442,6 +465,11 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
if not test_with_ipv6: if not test_with_ipv6:
subnet.ip_version = 4 subnet.ip_version = 4
subnet_params['ip_version'] = subnet.ip_version subnet_params['ip_version'] = subnet.ip_version
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.network_create(IsA(http.HttpRequest), api.neutron.network_create(IsA(http.HttpRequest),
**params).AndReturn(network) **params).AndReturn(network)
api.neutron.subnet_create(IsA(http.HttpRequest), api.neutron.subnet_create(IsA(http.HttpRequest),
@ -471,7 +499,9 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.test_network_create_post_with_subnet(test_with_ipv6=False) self.test_network_create_post_with_subnet(test_with_ipv6=False)
@test.create_stubs({api.neutron: ('network_create', @test.create_stubs({api.neutron: ('network_create',
'profile_list',)}) 'profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_post_network_exception(self, def test_network_create_post_network_exception(self,
test_with_profile=False): test_with_profile=False):
network = self.networks.first() network = self.networks.first()
@ -484,6 +514,11 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
api.neutron.profile_list(IsA(http.HttpRequest), api.neutron.profile_list(IsA(http.HttpRequest),
'network').AndReturn(net_profiles) 'network').AndReturn(net_profiles)
params['net_profile_id'] = net_profile_id params['net_profile_id'] = net_profile_id
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.network_create(IsA(http.HttpRequest), api.neutron.network_create(IsA(http.HttpRequest),
**params).AndRaise(self.exceptions.neutron) **params).AndRaise(self.exceptions.neutron)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -509,7 +544,9 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
test_with_profile=True) test_with_profile=True)
@test.create_stubs({api.neutron: ('network_create', @test.create_stubs({api.neutron: ('network_create',
'profile_list')}) 'profile_list',
'is_extension_supported',
'subnetpool_list')})
def test_network_create_post_with_subnet_network_exception( def test_network_create_post_with_subnet_network_exception(
self, self,
test_with_profile=False, test_with_profile=False,
@ -526,6 +563,11 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
api.neutron.profile_list(IsA(http.HttpRequest), api.neutron.profile_list(IsA(http.HttpRequest),
'network').AndReturn(net_profiles) 'network').AndReturn(net_profiles)
params['net_profile_id'] = net_profile_id params['net_profile_id'] = net_profile_id
api.neutron.is_extension_supported(IsA(http.HttpRequest),
'subnet_allocation').\
AndReturn(True)
api.neutron.subnetpool_list(IsA(http.HttpRequest)).\
AndReturn(self.subnetpools.list())
api.neutron.network_create(IsA(http.HttpRequest), api.neutron.network_create(IsA(http.HttpRequest),
**params).AndRaise(self.exceptions.neutron) **params).AndRaise(self.exceptions.neutron)
self.mox.ReplayAll() self.mox.ReplayAll()

View File

@ -253,7 +253,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
DETAIL_PATH = 'horizon:%s:routers:detail' % DASHBOARD DETAIL_PATH = 'horizon:%s:routers:detail' % DASHBOARD
@test.create_stubs({api.neutron: ('router_create', @test.create_stubs({api.neutron: ('router_create',
'get_feature_permission',)}) 'get_feature_permission',
'network_list')})
def test_router_create_post(self): def test_router_create_post(self):
router = self.routers.first() router = self.routers.first()
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
@ -262,6 +263,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
"l3-ha", "create")\ "l3-ha", "create")\
.AndReturn(False) .AndReturn(False)
api.neutron.network_list(IsA(http.HttpRequest))\
.AndReturn(self.networks.list())
params = {'name': router.name, params = {'name': router.name,
'admin_state_up': str(router.admin_state_up)} 'admin_state_up': str(router.admin_state_up)}
api.neutron.router_create(IsA(http.HttpRequest), **params)\ api.neutron.router_create(IsA(http.HttpRequest), **params)\
@ -277,7 +280,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
self.assertRedirectsNoFollow(res, self.INDEX_URL) self.assertRedirectsNoFollow(res, self.INDEX_URL)
@test.create_stubs({api.neutron: ('router_create', @test.create_stubs({api.neutron: ('router_create',
'get_feature_permission',)}) 'get_feature_permission',
'network_list')})
def test_router_create_post_mode_server_default(self): def test_router_create_post_mode_server_default(self):
router = self.routers.first() router = self.routers.first()
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
@ -286,6 +290,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
"l3-ha", "create")\ "l3-ha", "create")\
.AndReturn(True) .AndReturn(True)
api.neutron.network_list(IsA(http.HttpRequest))\
.AndReturn(self.networks.list())
params = {'name': router.name, params = {'name': router.name,
'admin_state_up': str(router.admin_state_up)} 'admin_state_up': str(router.admin_state_up)}
api.neutron.router_create(IsA(http.HttpRequest), **params)\ api.neutron.router_create(IsA(http.HttpRequest), **params)\
@ -303,7 +309,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
self.assertRedirectsNoFollow(res, self.INDEX_URL) self.assertRedirectsNoFollow(res, self.INDEX_URL)
@test.create_stubs({api.neutron: ('router_create', @test.create_stubs({api.neutron: ('router_create',
'get_feature_permission',)}) 'get_feature_permission',
'network_list')})
def test_dvr_ha_router_create_post(self): def test_dvr_ha_router_create_post(self):
router = self.routers.first() router = self.routers.first()
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
@ -312,6 +319,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
"l3-ha", "create")\ "l3-ha", "create")\
.MultipleTimes().AndReturn(True) .MultipleTimes().AndReturn(True)
api.neutron.network_list(IsA(http.HttpRequest))\
.AndReturn(self.networks.list())
param = {'name': router.name, param = {'name': router.name,
'distributed': True, 'distributed': True,
'ha': True, 'ha': True,
@ -331,7 +340,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
self.assertRedirectsNoFollow(res, self.INDEX_URL) self.assertRedirectsNoFollow(res, self.INDEX_URL)
@test.create_stubs({api.neutron: ('router_create', @test.create_stubs({api.neutron: ('router_create',
'get_feature_permission',)}) 'get_feature_permission',
'network_list')})
def test_router_create_post_exception_error_case_409(self): def test_router_create_post_exception_error_case_409(self):
router = self.routers.first() router = self.routers.first()
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
@ -341,6 +351,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
"l3-ha", "create")\ "l3-ha", "create")\
.AndReturn(False) .AndReturn(False)
self.exceptions.neutron.status_code = 409 self.exceptions.neutron.status_code = 409
api.neutron.network_list(IsA(http.HttpRequest))\
.MultipleTimes().AndReturn(self.networks.list())
params = {'name': router.name, params = {'name': router.name,
'admin_state_up': str(router.admin_state_up)} 'admin_state_up': str(router.admin_state_up)}
api.neutron.router_create(IsA(http.HttpRequest), **params)\ api.neutron.router_create(IsA(http.HttpRequest), **params)\
@ -356,7 +368,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
self.assertRedirectsNoFollow(res, self.INDEX_URL) self.assertRedirectsNoFollow(res, self.INDEX_URL)
@test.create_stubs({api.neutron: ('router_create', @test.create_stubs({api.neutron: ('router_create',
'get_feature_permission',)}) 'get_feature_permission',
'network_list')})
def test_router_create_post_exception_error_case_non_409(self): def test_router_create_post_exception_error_case_non_409(self):
router = self.routers.first() router = self.routers.first()
api.neutron.get_feature_permission(IsA(http.HttpRequest), api.neutron.get_feature_permission(IsA(http.HttpRequest),
@ -366,6 +379,8 @@ class RouterActionTests(RouterMixin, test.TestCase):
"l3-ha", "create")\ "l3-ha", "create")\
.MultipleTimes().AndReturn(False) .MultipleTimes().AndReturn(False)
self.exceptions.neutron.status_code = 999 self.exceptions.neutron.status_code = 999
api.neutron.network_list(IsA(http.HttpRequest))\
.MultipleTimes().AndReturn(self.networks.list())
params = {'name': router.name, params = {'name': router.name,
'admin_state_up': str(router.admin_state_up)} 'admin_state_up': str(router.admin_state_up)}
api.neutron.router_create(IsA(http.HttpRequest), **params)\ api.neutron.router_create(IsA(http.HttpRequest), **params)\

View File

@ -27,6 +27,7 @@ from heatclient.common import template_format as hc_format
from openstack_dashboard import api from openstack_dashboard import api
from openstack_dashboard.test import helpers as test from openstack_dashboard.test import helpers as test
from openstack_dashboard.dashboards.project.stacks import api as project_api
from openstack_dashboard.dashboards.project.stacks import forms from openstack_dashboard.dashboards.project.stacks import forms
from openstack_dashboard.dashboards.project.stacks import mappings from openstack_dashboard.dashboards.project.stacks import mappings
from openstack_dashboard.dashboards.project.stacks import tables from openstack_dashboard.dashboards.project.stacks import tables
@ -803,7 +804,8 @@ class StackTests(test.TestCase):
self.assertEqual(res.context['stack_preview']['stack_name'], self.assertEqual(res.context['stack_preview']['stack_name'],
stack.stack_name) stack.stack_name)
@test.create_stubs({api.heat: ('stack_get', 'template_get')}) @test.create_stubs({api.heat: ('stack_get', 'template_get',
'resources_list')})
def test_detail_stack_topology(self): def test_detail_stack_topology(self):
stack = self.stacks.first() stack = self.stacks.first()
template = self.stack_templates.first() template = self.stack_templates.first()
@ -811,6 +813,8 @@ class StackTests(test.TestCase):
.MultipleTimes().AndReturn(stack) .MultipleTimes().AndReturn(stack)
api.heat.template_get(IsA(http.HttpRequest), stack.id) \ api.heat.template_get(IsA(http.HttpRequest), stack.id) \
.AndReturn(json.loads(template.validate)) .AndReturn(json.loads(template.validate))
api.heat.resources_list(IsA(http.HttpRequest), stack.stack_name) \
.AndReturn([])
self.mox.ReplayAll() self.mox.ReplayAll()
url = '?'.join([reverse(DETAIL_URL, args=[stack.id]), url = '?'.join([reverse(DETAIL_URL, args=[stack.id]),
@ -825,7 +829,8 @@ class StackTests(test.TestCase):
self.assertIn('stack-green.svg', d3_data) self.assertIn('stack-green.svg', d3_data)
self.assertIn('Create Complete', d3_data) self.assertIn('Create Complete', d3_data)
@test.create_stubs({api.heat: ('stack_get', 'template_get')}) @test.create_stubs({api.heat: ('stack_get', 'template_get'),
project_api: ('d3_data',)})
def test_detail_stack_overview(self): def test_detail_stack_overview(self):
stack = self.stacks.first() stack = self.stacks.first()
template = self.stack_templates.first() template = self.stack_templates.first()
@ -833,6 +838,8 @@ class StackTests(test.TestCase):
.MultipleTimes().AndReturn(stack) .MultipleTimes().AndReturn(stack)
api.heat.template_get(IsA(http.HttpRequest), stack.id) \ api.heat.template_get(IsA(http.HttpRequest), stack.id) \
.AndReturn(json.loads(template.validate)) .AndReturn(json.loads(template.validate))
project_api.d3_data(IsA(http.HttpRequest), stack_id=stack.id) \
.AndReturn(json.dumps({"nodes": [], "stack": {}}))
self.mox.ReplayAll() self.mox.ReplayAll()
url = '?'.join([reverse(DETAIL_URL, args=[stack.id]), url = '?'.join([reverse(DETAIL_URL, args=[stack.id]),
@ -844,7 +851,8 @@ class StackTests(test.TestCase):
'project/stacks/_detail_overview.html') 'project/stacks/_detail_overview.html')
self.assertEqual(stack.stack_name, overview_data.stack_name) self.assertEqual(stack.stack_name, overview_data.stack_name)
@test.create_stubs({api.heat: ('stack_get', 'template_get')}) @test.create_stubs({api.heat: ('stack_get', 'template_get'),
project_api: ('d3_data',)})
def test_detail_stack_resources(self): def test_detail_stack_resources(self):
stack = self.stacks.first() stack = self.stacks.first()
template = self.stack_templates.first() template = self.stack_templates.first()
@ -852,6 +860,8 @@ class StackTests(test.TestCase):
.MultipleTimes().AndReturn(stack) .MultipleTimes().AndReturn(stack)
api.heat.template_get(IsA(http.HttpRequest), stack.id) \ api.heat.template_get(IsA(http.HttpRequest), stack.id) \
.AndReturn(json.loads(template.validate)) .AndReturn(json.loads(template.validate))
project_api.d3_data(IsA(http.HttpRequest), stack_id=stack.id) \
.AndReturn(json.dumps({"nodes": [], "stack": {}}))
self.mox.ReplayAll() self.mox.ReplayAll()
url = '?'.join([reverse(DETAIL_URL, args=[stack.id]), url = '?'.join([reverse(DETAIL_URL, args=[stack.id]),

View File

@ -20,6 +20,7 @@ import collections
import copy import copy
from functools import wraps # noqa from functools import wraps # noqa
import os import os
import traceback
import unittest import unittest
import django import django
@ -35,7 +36,6 @@ from ceilometerclient.v2 import client as ceilometer_client
from cinderclient import client as cinder_client from cinderclient import client as cinder_client
import glanceclient import glanceclient
from heatclient import client as heat_client from heatclient import client as heat_client
import httplib2
from importlib import import_module from importlib import import_module
from keystoneclient.v2_0 import client as keystone_client from keystoneclient.v2_0 import client as keystone_client
import mock import mock
@ -44,6 +44,7 @@ from neutronclient.v2_0 import client as neutron_client
from novaclient.v2 import client as nova_client from novaclient.v2 import client as nova_client
from openstack_auth import user from openstack_auth import user
from openstack_auth import utils from openstack_auth import utils
from requests.packages.urllib3.connection import HTTPConnection
import six import six
from six import moves from six import moves
from swiftclient import client as swift_client from swiftclient import client as swift_client
@ -151,14 +152,21 @@ class TestCase(horizon_helpers.TestCase):
* The ability to override specific time data controls for easier testing. * The ability to override specific time data controls for easier testing.
* Several handy additional assertion methods. * Several handy additional assertion methods.
""" """
def setUp(self):
def fake_conn_request(*args, **kwargs):
raise Exception("An external URI request tried to escape through "
"an httplib2 client. Args: %s, kwargs: %s"
% (args, kwargs))
self._real_conn_request = httplib2.Http._conn_request # To force test failures when unmocked API calls are attempted, provide
httplib2.Http._conn_request = fake_conn_request # boolean variable to store failures
missing_mocks = False
def fake_conn_request(self):
# print a stacktrace to illustrate where the unmocked API call
# is being made from
traceback.print_stack()
# forcing a test failure for missing mock
self.missing_mocks = True
def setUp(self):
self._real_conn_request = HTTPConnection.connect
HTTPConnection.connect = self.fake_conn_request
self._real_context_processor = context_processors.openstack self._real_context_processor = context_processors.openstack
context_processors.openstack = lambda request: self.context context_processors.openstack = lambda request: self.context
@ -202,12 +210,16 @@ class TestCase(horizon_helpers.TestCase):
self.patchers['aggregates'].start() self.patchers['aggregates'].start()
def tearDown(self): def tearDown(self):
httplib2.Http._conn_request = self._real_conn_request HTTPConnection.connect = self._real_conn_request
context_processors.openstack = self._real_context_processor context_processors.openstack = self._real_context_processor
utils.get_user = self._real_get_user utils.get_user = self._real_get_user
mock.patch.stopall() mock.patch.stopall()
super(TestCase, self).tearDown() super(TestCase, self).tearDown()
# cause a test failure if an unmocked API call was attempted
if self.missing_mocks:
raise AssertionError("An unmocked API call was made.")
def setActiveUser(self, id=None, token=None, username=None, tenant_id=None, def setActiveUser(self, id=None, token=None, username=None, tenant_id=None,
service_catalog=None, tenant_name=None, roles=None, service_catalog=None, tenant_name=None, roles=None,
authorized_tenants=None, enabled=True, domain_id=None, authorized_tenants=None, enabled=True, domain_id=None,

View File

@ -16,7 +16,6 @@ django-babel>=0.4.0 # BSD
django-compressor>=2.0 # MIT django-compressor>=2.0 # MIT
django-openstack-auth>=2.2.0 # Apache-2.0 django-openstack-auth>=2.2.0 # Apache-2.0
django-pyscss>=2.0.2 # BSD License (2 clause) django-pyscss>=2.0.2 # BSD License (2 clause)
httplib2>=0.7.5 # MIT
iso8601>=0.1.11 # MIT iso8601>=0.1.11 # MIT
netaddr!=0.7.16,>=0.7.12 # BSD netaddr!=0.7.16,>=0.7.12 # BSD
oslo.concurrency>=3.5.0 # Apache-2.0 oslo.concurrency>=3.5.0 # Apache-2.0

View File

@ -21,6 +21,7 @@ nosexcover # BSD
openstack.nose-plugin>=0.7 # Apache-2.0 openstack.nose-plugin>=0.7 # Apache-2.0
oslosphinx!=3.4.0,>=2.5.0 # Apache-2.0 oslosphinx!=3.4.0,>=2.5.0 # Apache-2.0
reno>=1.6.2 # Apache2 reno>=1.6.2 # Apache2
requests>=2.8.1,!=2.9.0 # Apache-2.0
selenium>=2.50.1 # Apache-2.0 selenium>=2.50.1 # Apache-2.0
sphinx!=1.2.0,!=1.3b1,<1.3,>=1.1.2 # BSD sphinx!=1.2.0,!=1.3b1,<1.3,>=1.1.2 # BSD
testtools>=1.4.0 # MIT testtools>=1.4.0 # MIT