openstack/heat
Code Issues Proposed changes

Merge "Remove mox usage from test_vpc (part 2)"

Browse Source
This commit is contained in:
Zuul 2018-06-08 03:25:36 +00:00 committed by Gerrit Code Review
commit 970bc3bdf4
1 changed files with 203 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

397
heat/tests/test_vpc.py
View File

@ -11,6 +11,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import mock
import uuid import uuid
from heat.common import exception from heat.common import exception
@ -34,48 +35,24 @@ class VPCTestBase(common.HeatTestCase):
def setUp(self): def setUp(self):
super(VPCTestBase, self).setUp() super(VPCTestBase, self).setUp()
self.m_add_interface_router = self.patchobject( self.mockclient = mock.Mock(spec=neutronclient.Client)
neutronclient.Client, 'add_interface_router', self.patchobject(neutronclient, 'Client', return_value=self.mockclient)
return_value=None) self.mockclient.add_interface_router.return_value = None
self.m_add_gateway_router = self.patchobject( self.mockclient.delete_network.return_value = None
neutronclient.Client, 'add_gateway_router') self.mockclient.delete_port.return_value = None
self.m_create_network = self.patchobject(neutronclient.Client, self.mockclient.delete_router.return_value = None
'create_network') self.mockclient.delete_subnet.return_value = None
self.m.StubOutWithMock(neutronclient.Client, 'create_port') self.mockclient.remove_interface_router.return_value = None
self.m_create_router = self.patchobject(neutronclient.Client, self.mockclient.remove_gateway_router.return_value = None
'create_router') self.mockclient.delete_security_group_rule.return_value = None
self.m_create_subnet = self.patchobject(neutronclient.Client, self.mockclient.delete_security_group.return_value = None
'create_subnet') self.vpc_name = utils.PhysName('test_stack', 'the_vpc')
self.m_delete_network = self.patchobject(neutronclient.Client, self.mock_router_for_vpc()
'delete_network',
return_value=None) self.subnet_name = utils.PhysName('test_stack', 'the_subnet')
self.m.StubOutWithMock(neutronclient.Client, 'delete_port') self.mock_show_subnet()
self.m_delete_router = self.patchobject(neutronclient.Client, self.stub_SubnetConstraint_validate()
'delete_router', self.mock_create_subnet()
return_value=None)
self.m_delete_subnet = self.patchobject(neutronclient.Client,
'delete_subnet',
return_value=None)
self.m.StubOutWithMock(neutronclient.Client, 'list_networks')
self.m_list_routers = self.patchobject(neutronclient.Client,
'list_routers')
self.m.StubOutWithMock(neutronclient.Client, 'remove_gateway_router')
self.m_remove_interface_router = self.patchobject(
neutronclient.Client, 'remove_interface_router',
return_value=None)
self.m.StubOutWithMock(neutronclient.Client, 'show_subnet')
self.m_show_network = self.patchobject(neutronclient.Client,
'show_network')
self.m.StubOutWithMock(neutronclient.Client, 'show_port')
self.m.StubOutWithMock(neutronclient.Client, 'show_router')
self.m.StubOutWithMock(neutronclient.Client, 'create_security_group')
self.m.StubOutWithMock(neutronclient.Client, 'show_security_group')
self.m.StubOutWithMock(neutronclient.Client, 'list_security_groups')
self.m.StubOutWithMock(neutronclient.Client, 'delete_security_group')
self.m.StubOutWithMock(
neutronclient.Client, 'create_security_group_rule')
self.m.StubOutWithMock(
neutronclient.Client, 'delete_security_group_rule')
def create_stack(self, templ): def create_stack(self, templ):
t = template_format.parse(templ) t = template_format.parse(templ)
@ -92,14 +69,13 @@ class VPCTestBase(common.HeatTestCase):
return stack return stack
def validate_mock_create_network(self): def validate_mock_create_network(self):
self.m_show_network.assert_called_with('aaaa') self.mockclient.show_network.assert_called_with('aaaa')
self.m_create_network.assert_called_once_with({ self.mockclient.create_network.assert_called_once_with({
'network': {'name': self.vpc_name}}) 'network': {'name': self.vpc_name}})
self.m_create_router.assert_called_once() self.mockclient.create_router.assert_called_once()
def mock_create_network(self): def mock_create_network(self):
self.vpc_name = utils.PhysName('test_stack', 'the_vpc') self.mockclient.create_network.return_value = {
self.m_create_network.return_value = {
'network': { 'network': {
'status': 'BUILD', 'status': 'BUILD',
'subnets': [], 'subnets': [],
@ -130,9 +106,9 @@ class VPCTestBase(common.HeatTestCase):
"tenant_id": "c1210485b2424d48804aad5d39c61b8f", "tenant_id": "c1210485b2424d48804aad5d39c61b8f",
"id": "aaaa" "id": "aaaa"
}}) }})
self.m_show_network.side_effect = show_network_returns self.mockclient.show_network.side_effect = show_network_returns
self.m_create_router.return_value = { self.mockclient.create_router.return_value = {
'router': { 'router': {
'status': 'BUILD', 'status': 'BUILD',
'name': self.vpc_name, 'name': self.vpc_name,
@ -140,11 +116,10 @@ class VPCTestBase(common.HeatTestCase):
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'bbbb' 'id': 'bbbb'
}} }}
self.mock_router_for_vpc()
def mock_create_subnet(self): def mock_create_subnet(self):
self.subnet_name = utils.PhysName('test_stack', 'the_subnet') self.subnet_name = utils.PhysName('test_stack', 'the_subnet')
self.m_create_subnet.return_value = { self.mockclient.create_subnet.return_value = {
'subnet': { 'subnet': {
'status': 'ACTIVE', 'status': 'ACTIVE',
'name': self.subnet_name, 'name': self.subnet_name,
@ -152,10 +127,8 @@ class VPCTestBase(common.HeatTestCase):
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'cccc'}} 'id': 'cccc'}}
self.mock_router_for_vpc()
def mock_show_subnet(self): def mock_show_subnet(self):
neutronclient.Client.show_subnet('cccc').AndReturn({ self.mockclient.show_subnet.return_value = {
'subnet': { 'subnet': {
'name': self.subnet_name, 'name': self.subnet_name,
'network_id': 'aaaa', 'network_id': 'aaaa',
@ -167,16 +140,11 @@ class VPCTestBase(common.HeatTestCase):
'cidr': '10.0.0.0/24', 'cidr': '10.0.0.0/24',
'id': 'cccc', 'id': 'cccc',
'enable_dhcp': False, 'enable_dhcp': False,
}}) }}
def mock_create_security_group(self): def mock_create_security_group(self):
self.sg_name = utils.PhysName('test_stack', 'the_sg') self.sg_name = utils.PhysName('test_stack', 'the_sg')
neutronclient.Client.create_security_group({ self.mockclient.create_security_group.return_value = {
'security_group': {
'name': self.sg_name,
'description': 'SSH access'
}
}).AndReturn({
'security_group': { 'security_group': {
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'name': self.sg_name, 'name': self.sg_name,
@ -184,9 +152,8 @@ class VPCTestBase(common.HeatTestCase):
'security_group_rules': [], 'security_group_rules': [],
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3' 'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
} }
}) }
self.create_security_group_rule_expected = {
neutronclient.Client.create_security_group_rule({
'security_group_rule': { 'security_group_rule': {
'direction': 'ingress', 'direction': 'ingress',
'remote_group_id': None, 'remote_group_id': None,
@ -197,7 +164,9 @@ class VPCTestBase(common.HeatTestCase):
'protocol': 'tcp', 'protocol': 'tcp',
'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3' 'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
} }
}).AndReturn({ }
self.mockclient.create_security_group_rule.return_value = {
'security_group_rule': { 'security_group_rule': {
'direction': 'ingress', 'direction': 'ingress',
'remote_group_id': None, 'remote_group_id': None,
@ -209,13 +178,12 @@ class VPCTestBase(common.HeatTestCase):
'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3', 'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3',
'id': 'bbbb' 'id': 'bbbb'
} }
}) }
def mock_show_security_group(self, group=None): def mock_show_security_group(self):
sg_name = utils.PhysName('test_stack', 'the_sg') sg_name = utils.PhysName('test_stack', 'the_sg')
group = group or '0389f747-7785-4757-b7bb-2ab07e4b09c3' self._group = '0389f747-7785-4757-b7bb-2ab07e4b09c3'
if group == '0389f747-7785-4757-b7bb-2ab07e4b09c3': self.mockclient.show_security_group.return_value = {
neutronclient.Client.show_security_group(group).AndReturn({
'security_group': { 'security_group': {
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'name': sg_name, 'name': sg_name,
@ -233,24 +201,10 @@ class VPCTestBase(common.HeatTestCase):
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'port_range_min': 22 'port_range_min': 22
}], }],
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'}}) 'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'}}
elif group == 'INVALID-NO-REF':
neutronclient.Client.show_security_group(group).AndRaise(
neutron_exc.NeutronClientException(status_code=404))
elif group == 'RaiseException':
neutronclient.Client.show_security_group(
'0389f747-7785-4757-b7bb-2ab07e4b09c3').AndRaise(
neutron_exc.NeutronClientException(status_code=403))
def mock_delete_security_group(self):
self.mock_show_security_group()
neutronclient.Client.delete_security_group_rule(
'bbbb').AndReturn(None)
neutronclient.Client.delete_security_group(
'0389f747-7785-4757-b7bb-2ab07e4b09c3').AndReturn(None)
def mock_router_for_vpc(self): def mock_router_for_vpc(self):
self.m_list_routers.return_value = { self.mockclient.list_routers.return_value = {
"routers": [{ "routers": [{
"status": "ACTIVE", "status": "ACTIVE",
"external_gateway_info": { "external_gateway_info": {
@ -264,13 +218,9 @@ class VPCTestBase(common.HeatTestCase):
}] }]
} }
def mock_delete_subnet(self):
# TODO(ricolin) remove this func once we all move to mock
self.mock_router_for_vpc()
def mock_create_route_table(self): def mock_create_route_table(self):
self.rt_name = utils.PhysName('test_stack', 'the_route_table') self.rt_name = utils.PhysName('test_stack', 'the_route_table')
self.m_create_router.return_value = { self.mockclient.create_router.return_value = {
'router': { 'router': {
'status': 'BUILD', 'status': 'BUILD',
'name': self.rt_name, 'name': self.rt_name,
@ -279,7 +229,7 @@ class VPCTestBase(common.HeatTestCase):
'id': 'ffff' 'id': 'ffff'
} }
} }
neutronclient.Client.show_router('ffff').AndReturn({ show_router_returns = [{
'router': { 'router': {
'status': 'BUILD', 'status': 'BUILD',
'name': self.rt_name, 'name': self.rt_name,
@ -287,8 +237,9 @@ class VPCTestBase(common.HeatTestCase):
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'ffff' 'id': 'ffff'
} }
}) }]
neutronclient.Client.show_router('ffff').AndReturn({ for i in range(3):
show_router_returns.append({
'router': { 'router': {
'status': 'ACTIVE', 'status': 'ACTIVE',
'name': self.rt_name, 'name': self.rt_name,
@ -297,20 +248,7 @@ class VPCTestBase(common.HeatTestCase):
'id': 'ffff' 'id': 'ffff'
} }
}) })
self.mock_router_for_vpc() self.mockclient.show_router.side_effect = show_router_returns
def mock_create_association(self):
# TODO(ricolin) merge mock_create_association and
# mock_delete_association func once we all move to mock
self.mock_show_subnet()
self.mock_router_for_vpc()
def mock_delete_association(self):
self.mock_show_subnet()
self.mock_router_for_vpc()
def mock_delete_route_table(self):
neutronclient.Client.remove_gateway_router('ffff').AndReturn(None)
def assertResourceState(self, resource, ref_id): def assertResourceState(self, resource, ref_id):
self.assertIsNone(resource.validate()) self.assertIsNone(resource.validate())
@ -330,7 +268,8 @@ Resources:
def mock_create_network_failed(self): def mock_create_network_failed(self):
self.vpc_name = utils.PhysName('test_stack', 'the_vpc') self.vpc_name = utils.PhysName('test_stack', 'the_vpc')
self.m_create_network.side_effect = neutron_exc.NeutronClientException exc = neutron_exc.NeutronClientException
self.mockclient.create_network.side_effect = exc
def test_vpc(self): def test_vpc(self):
self.mock_create_network() self.mock_create_network()
@ -339,16 +278,16 @@ Resources:
vpc = stack['the_vpc'] vpc = stack['the_vpc']
self.assertResourceState(vpc, 'aaaa') self.assertResourceState(vpc, 'aaaa')
self.validate_mock_create_network() self.validate_mock_create_network()
self.assertEqual(3, self.m_show_network.call_count) self.assertEqual(3, self.mockclient.show_network.call_count)
scheduler.TaskRunner(vpc.delete)() scheduler.TaskRunner(vpc.delete)()
self.m_show_network.assert_called_with('aaaa') self.mockclient.show_network.assert_called_with('aaaa')
self.assertEqual(4, self.m_show_network.call_count) self.assertEqual(4, self.mockclient.show_network.call_count)
self.assertEqual(2, self.m_list_routers.call_count) self.assertEqual(2, self.mockclient.list_routers.call_count)
self.m_list_routers.assert_called_with(name=self.vpc_name) self.mockclient.list_routers.assert_called_with(name=self.vpc_name)
self.m_delete_router.assert_called_once_with('bbbb') self.mockclient.delete_router.assert_called_once_with('bbbb')
self.m_delete_network.assert_called_once_with('aaaa') self.mockclient.delete_network.assert_called_once_with('aaaa')
def test_vpc_delete_successful_if_created_failed(self): def test_vpc_delete_successful_if_created_failed(self):
self.mock_create_network_failed() self.mock_create_network_failed()
@ -357,10 +296,10 @@ Resources:
stack = self.parse_stack(t) stack = self.parse_stack(t)
scheduler.TaskRunner(stack.create)() scheduler.TaskRunner(stack.create)()
self.assertEqual((stack.CREATE, stack.FAILED), stack.state) self.assertEqual((stack.CREATE, stack.FAILED), stack.state)
self.m_create_network.assert_called_once_with( self.mockclient.create_network.assert_called_once_with(
{'network': {'name': self.vpc_name}}) {'network': {'name': self.vpc_name}})
scheduler.TaskRunner(stack.delete)() scheduler.TaskRunner(stack.delete)()
self.m_delete_network.assert_not_called() self.mockclient.delete_network.assert_not_called()
class SubnetTest(VPCTestBase): class SubnetTest(VPCTestBase):
@ -381,34 +320,28 @@ Resources:
def test_subnet(self): def test_subnet(self):
self.mock_create_network() self.mock_create_network()
self.mock_create_subnet()
self.mock_delete_subnet()
# mock delete subnet which is already deleted
self.mock_router_for_vpc()
exc = neutron_exc.NeutronClientException(status_code=404) exc = neutron_exc.NeutronClientException(status_code=404)
self.m_remove_interface_router.side_effect = exc self.mockclient.remove_interface_router.side_effect = exc
self.m_delete_subnet.side_effect = neutron_exc.NeutronClientException( self.mockclient.delete_subnet.side_effect = exc
status_code=404)
stack = self.create_stack(self.test_template) stack = self.create_stack(self.test_template)
subnet = stack['the_subnet'] subnet = stack['the_subnet']
self.assertResourceState(subnet, 'cccc') self.assertResourceState(subnet, 'cccc')
self.m_list_routers.assert_called_with(name=self.vpc_name) self.mockclient.list_routers.assert_called_with(name=self.vpc_name)
self.validate_mock_create_network() self.validate_mock_create_network()
self.m_add_interface_router.assert_called_once_with( self.mockclient.add_interface_router.assert_called_once_with(
u'bbbb', {'subnet_id': 'cccc'}) u'bbbb', {'subnet_id': 'cccc'})
self.m_create_subnet.assert_called_once_with( self.mockclient.create_subnet.assert_called_once_with(
{'subnet': { {'subnet': {
'network_id': u'aaaa', 'network_id': u'aaaa',
'cidr': u'10.0.0.0/24', 'cidr': u'10.0.0.0/24',
'ip_version': 4, 'ip_version': 4,
'name': self.subnet_name}}) 'name': self.subnet_name}})
self.assertEqual(4, self.m_show_network.call_count) self.assertEqual(4, self.mockclient.show_network.call_count)
self.assertRaises( self.assertRaises(
exception.InvalidTemplateAttribute, exception.InvalidTemplateAttribute,
subnet.FnGetAtt, subnet.FnGetAtt,
@ -420,24 +353,24 @@ Resources:
scheduler.TaskRunner(subnet.delete)() scheduler.TaskRunner(subnet.delete)()
scheduler.TaskRunner(stack['the_vpc'].delete)() scheduler.TaskRunner(stack['the_vpc'].delete)()
self.m_show_network.assert_called_with('aaaa') self.mockclient.show_network.assert_called_with('aaaa')
self.m_list_routers.assert_called_with(name=self.vpc_name) self.mockclient.list_routers.assert_called_with(name=self.vpc_name)
self.assertEqual(2, self.m_list_routers.call_count) self.assertEqual(2, self.mockclient.list_routers.call_count)
self.assertEqual(7, self.m_show_network.call_count) self.assertEqual(7, self.mockclient.show_network.call_count)
def _mock_create_subnet_failed(self, stack_name): def _mock_create_subnet_failed(self, stack_name):
self.subnet_name = utils.PhysName(stack_name, 'the_subnet') self.subnet_name = utils.PhysName(stack_name, 'the_subnet')
self.m_create_subnet.return_value = { self.mockclient.create_subnet.return_value = {
'subnet': { 'subnet': {
'status': 'ACTIVE', 'status': 'ACTIVE',
'name': self.subnet_name, 'name': self.subnet_name,
'admin_state_up': True, 'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f', 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'cccc'}} 'id': 'cccc'}}
exc = neutron_exc.NeutronClientException(
self.m_show_network.side_effect = neutron_exc.NeutronClientException(
status_code=404) status_code=404)
self.mockclient.show_network.side_effect = exc
def test_create_failed_delete_success(self): def test_create_failed_delete_success(self):
stack_name = 'test_subnet_' stack_name = 'test_subnet_'
@ -459,19 +392,19 @@ Resources:
ref_id = rsrc.FnGetRefId() ref_id = rsrc.FnGetRefId()
self.assertEqual(u'cccc', ref_id) self.assertEqual(u'cccc', ref_id)
self.m_create_subnet.assert_called_once_with( self.mockclient.create_subnet.assert_called_once_with(
{'subnet': { {'subnet': {
'network_id': u'aaaa', 'network_id': u'aaaa',
'cidr': u'10.0.0.0/24', 'cidr': u'10.0.0.0/24',
'ip_version': 4, 'ip_version': 4,
'name': self.subnet_name}}) 'name': self.subnet_name}})
self.assertEqual(1, self.m_show_network.call_count) self.assertEqual(1, self.mockclient.show_network.call_count)
self.assertIsNone(scheduler.TaskRunner(rsrc.delete)()) self.assertIsNone(scheduler.TaskRunner(rsrc.delete)())
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state) self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
self.assertEqual(2, self.m_show_network.call_count) self.assertEqual(2, self.mockclient.show_network.call_count)
self.m_delete_subnet.assert_called_once_with('cccc') self.mockclient.delete_subnet.assert_called_once_with('cccc')
class NetworkInterfaceTest(VPCTestBase): class NetworkInterfaceTest(VPCTestBase):
@ -578,13 +511,20 @@ Resources:
- INVALID-NO-REF - INVALID-NO-REF
''' '''
def setUp(self):
super(NetworkInterfaceTest, self).setUp()
self.mock_create_network()
self.mock_create_security_group()
self.mock_show_security_group()
self.mock_show_network_interface()
def mock_create_network_interface( def mock_create_network_interface(
self, security_groups=['0389f747-7785-4757-b7bb-2ab07e4b09c3']): self, security_groups=['0389f747-7785-4757-b7bb-2ab07e4b09c3']):
self.patchobject(resource.Resource, 'is_using_neutron', self.patchobject(resource.Resource, 'is_using_neutron',
return_value=True) return_value=True)
self.nic_name = utils.PhysName('test_stack', 'the_nic') self.nic_name = utils.PhysName('test_stack', 'the_nic')
port = {'network_id': 'aaaa', self._port = {'network_id': 'aaaa',
'fixed_ips': [{ 'fixed_ips': [{
'subnet_id': u'cccc', 'subnet_id': u'cccc',
'ip_address': u'10.0.0.100' 'ip_address': u'10.0.0.100'
@ -592,9 +532,9 @@ Resources:
'name': self.nic_name, 'name': self.nic_name,
'admin_state_up': True} 'admin_state_up': True}
if security_groups: if security_groups:
port['security_groups'] = security_groups self._port['security_groups'] = security_groups
neutronclient.Client.create_port({'port': port}).AndReturn({ self.mockclient.create_port.return_value = {
'port': { 'port': {
'admin_state_up': True, 'admin_state_up': True,
'device_id': '', 'device_id': '',
@ -612,11 +552,11 @@ Resources:
'status': 'ACTIVE', 'status': 'ACTIVE',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f' 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f'
} }
}) }
def mock_show_network_interface(self): def mock_show_network_interface(self):
self.nic_name = utils.PhysName('test_stack', 'the_nic') self.nic_name = utils.PhysName('test_stack', 'the_nic')
neutronclient.Client.show_port('dddd').AndReturn({ self.mockclient.show_port.return_value = {
'port': { 'port': {
'admin_state_up': True, 'admin_state_up': True,
'device_id': '', 'device_id': '',
@ -635,26 +575,23 @@ Resources:
'status': 'ACTIVE', 'status': 'ACTIVE',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f' 'tenant_id': 'c1210485b2424d48804aad5d39c61b8f'
} }
}) }
def mock_delete_network_interface(self):
neutronclient.Client.delete_port('dddd').AndReturn(None)
def test_network_interface(self): def test_network_interface(self):
self.mock_create_security_group()
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface() self.mock_create_network_interface()
self.mock_show_network_interface()
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.mock_delete_security_group()
self.m.ReplayAll()
stack = self.create_stack(self.test_template) stack = self.create_stack(self.test_template)
self.validate_mock_create_network()
self.mockclient.create_port.assert_called_once_with(
{'port': self._port})
self.mockclient.create_security_group.assert_called_once_with({
'security_group': {
'name': self.sg_name,
'description': 'SSH access'
}
})
self.mockclient.create_security_group_rule.assert_called_once_with(
self.create_security_group_rule_expected)
try: try:
self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state) self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state)
rsrc = stack['the_nic'] rsrc = stack['the_nic']
@ -662,49 +599,57 @@ Resources:
self.assertEqual('10.0.0.100', rsrc.FnGetAtt('PrivateIpAddress')) self.assertEqual('10.0.0.100', rsrc.FnGetAtt('PrivateIpAddress'))
finally: finally:
scheduler.TaskRunner(stack.delete)() scheduler.TaskRunner(stack.delete)()
self.mockclient.delete_port.assert_called_once_with('dddd')
self.m.VerifyAll() self.mockclient.show_subnet.assert_called_once_with('cccc')
self.mockclient.show_port.assert_called_once_with('dddd')
self.mockclient.show_security_group.assert_called_once_with(
self._group)
self.mockclient.delete_security_group.assert_called_once_with(
'0389f747-7785-4757-b7bb-2ab07e4b09c3')
self.mockclient.delete_security_group_rule.assert_called_once_with(
'bbbb')
def test_network_interface_existing_groupset(self): def test_network_interface_existing_groupset(self):
self.m.StubOutWithMock(parser.Stack, 'resource_by_refid') self.patchobject(parser.Stack, 'resource_by_refid')
self.mock_create_security_group()
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface() self.mock_create_network_interface()
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.mock_delete_security_group()
self.m.ReplayAll()
stack = self.create_stack(self.test_template) stack = self.create_stack(self.test_template)
self.validate_mock_create_network()
self.mockclient.create_port.assert_called_once_with(
{'port': self._port})
self.mockclient.create_security_group.assert_called_once_with({
'security_group': {
'name': self.sg_name,
'description': 'SSH access'
}
})
self.mockclient.create_security_group_rule.assert_called_once_with(
self.create_security_group_rule_expected)
try: try:
self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state) self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state)
rsrc = stack['the_nic'] rsrc = stack['the_nic']
self.assertResourceState(rsrc, 'dddd') self.assertResourceState(rsrc, 'dddd')
finally: finally:
stack.delete() stack.delete()
self.mockclient.delete_port.assert_called_once_with('dddd')
self.m.VerifyAll() self.mockclient.show_subnet.assert_called_once_with('cccc')
self.mockclient.show_security_group.assert_called_once_with(
self._group)
self.mockclient.delete_security_group.assert_called_once_with(
'0389f747-7785-4757-b7bb-2ab07e4b09c3')
self.mockclient.delete_security_group_rule.assert_called_once_with(
'bbbb')
def test_network_interface_no_groupset(self): def test_network_interface_no_groupset(self):
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface(security_groups=None) self.mock_create_network_interface(security_groups=None)
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.m.ReplayAll()
stack = self.create_stack(self.test_template_no_groupset) stack = self.create_stack(self.test_template_no_groupset)
self.mockclient.create_port.assert_called_once_with(
{'port': self._port})
stack.delete() stack.delete()
self.mockclient.delete_port.assert_called_once_with('dddd')
self.m.VerifyAll() self.mockclient.show_subnet.assert_called_once_with('cccc')
def test_network_interface_error(self): def test_network_interface_error(self):
self.assertRaises( self.assertRaises(
@ -746,9 +691,15 @@ Resources:
SubnetId: {Ref: the_subnet} SubnetId: {Ref: the_subnet}
''' '''
def setUp(self):
super(InternetGatewayTest, self).setUp()
self.mock_create_internet_gateway()
self.mock_create_network()
self.mock_create_route_table()
def mock_create_internet_gateway(self): def mock_create_internet_gateway(self):
neutronclient.Client.list_networks( self.mockclient.list_networks.return_value = {
**{'router:external': True}).AndReturn({'networks': [{ 'networks': [{
'status': 'ACTIVE', 'status': 'ACTIVE',
'subnets': [], 'subnets': [],
'name': 'nova', 'name': 'nova',
@ -757,34 +708,25 @@ Resources:
'admin_state_up': True, 'admin_state_up': True,
'shared': True, 'shared': True,
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3' 'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
}]}) }]}
def mock_delete_gateway_attachment(self):
neutronclient.Client.remove_gateway_router('ffff').AndReturn(None)
def test_internet_gateway(self): def test_internet_gateway(self):
self.mock_create_internet_gateway()
self.mock_create_network()
self.mock_create_subnet()
self.mock_create_route_table()
self.stub_SubnetConstraint_validate()
self.mock_create_association()
self.mock_delete_gateway_attachment()
self.mock_delete_association()
self.mock_delete_route_table()
self.mock_delete_subnet()
self.m.ReplayAll()
stack = self.create_stack(self.test_template) stack = self.create_stack(self.test_template)
self.m_create_router.assert_called_with( self.mockclient.show_network.assert_called_with('aaaa')
self.mockclient.create_network.assert_called_with({
'network': {'name': self.vpc_name}})
self.assertEqual(2, self.mockclient.create_router.call_count)
self.mockclient.create_router.assert_called_with(
{'router': {'name': self.rt_name}}) {'router': {'name': self.rt_name}})
self.m_add_interface_router.assert_called_once_with( self.mockclient.add_interface_router.assert_called_once_with(
u'bbbb', {'subnet_id': 'cccc'}) u'bbbb', {'subnet_id': 'cccc'})
self.mockclient.list_networks.assert_called_once_with(
**{'router:external': True})
gateway = stack['the_gateway'] gateway = stack['the_gateway']
self.assertResourceState(gateway, gateway.physical_resource_name()) self.assertResourceState(gateway, gateway.physical_resource_name())
self.m_add_gateway_router.assert_called_once_with( self.mockclient.add_gateway_router.assert_called_once_with(
'ffff', {'network_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'}) 'ffff', {'network_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'})
attachment = stack['the_attachment'] attachment = stack['the_attachment']
@ -794,11 +736,16 @@ Resources:
self.assertEqual([route_table], list(attachment._vpc_route_tables())) self.assertEqual([route_table], list(attachment._vpc_route_tables()))
stack.delete() stack.delete()
self.m_remove_interface_router.assert_called_with( self.mockclient.remove_interface_router.assert_called_with(
'ffff', 'ffff',
{'subnet_id': u'cccc'}) {'subnet_id': u'cccc'})
self.m_delete_router.assert_called_once_with('ffff') self.mockclient.remove_gateway_router.assert_called_with('ffff')
self.m.VerifyAll() self.assertEqual(2, self.mockclient.remove_gateway_router.call_count)
self.assertEqual(2, self.mockclient.show_subnet.call_count)
self.mockclient.show_subnet.assert_called_with('cccc')
self.mockclient.show_router.assert_called_with('ffff')
self.assertEqual(2, self.mockclient.show_router.call_count)
self.mockclient.delete_router.assert_called_once_with('ffff')
class RouteTableTest(VPCTestBase): class RouteTableTest(VPCTestBase):
@ -827,26 +774,24 @@ Resources:
SubnetId: {Ref: the_subnet} SubnetId: {Ref: the_subnet}
''' '''
def test_route_table(self): def setUp(self):
super(RouteTableTest, self).setUp()
self.mock_create_network() self.mock_create_network()
self.mock_create_subnet()
self.mock_create_route_table() self.mock_create_route_table()
self.stub_SubnetConstraint_validate()
self.mock_create_association()
self.mock_delete_association()
self.mock_delete_route_table()
self.mock_delete_subnet()
self.m.ReplayAll()
def test_route_table(self):
stack = self.create_stack(self.test_template) stack = self.create_stack(self.test_template)
self.m_create_router.assert_called_with( self.mockclient.create_router.assert_called_with(
{'router': {'name': self.rt_name}}) {'router': {'name': self.rt_name}})
self.m_add_interface_router.assert_called_once_with( self.mockclient.add_interface_router.assert_called_once_with(
u'bbbb', {'subnet_id': 'cccc'}) u'bbbb', {'subnet_id': 'cccc'})
route_table = stack['the_route_table'] route_table = stack['the_route_table']
self.assertResourceState(route_table, 'ffff') self.assertResourceState(route_table, 'ffff')
self.mockclient.show_network.assert_called_with('aaaa')
self.mockclient.create_network.assert_called_with({
'network': {'name': self.vpc_name}})
self.assertEqual(2, self.mockclient.create_router.call_count)
association = stack['the_association'] association = stack['the_association']
self.assertResourceState(association, 'the_association') self.assertResourceState(association, 'the_association')
@ -855,8 +800,12 @@ Resources:
scheduler.TaskRunner(route_table.delete)() scheduler.TaskRunner(route_table.delete)()
stack.delete() stack.delete()
self.m_remove_interface_router.assert_called_with( self.mockclient.remove_interface_router.assert_called_with(
'ffff', 'ffff',
{'subnet_id': u'cccc'}) {'subnet_id': u'cccc'})
self.m_delete_router.assert_called_once_with('ffff') self.mockclient.remove_gateway_router.assert_called_once_with('ffff')
self.m.VerifyAll() self.assertEqual(2, self.mockclient.show_subnet.call_count)
self.mockclient.show_subnet.assert_called_with('cccc')
self.assertEqual(2, self.mockclient.show_router.call_count)
self.mockclient.show_router.assert_called_with('ffff')
self.mockclient.delete_router.assert_called_once_with('ffff')