1
# Copyright 2013 PLUMgrid, Inc. All Rights Reserved.
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
4
# not use this file except in compliance with the License. You may obtain
5
# a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
# License for the specific language governing permissions and limitations
16
Test cases for Neutron PLUMgrid Plug-in
20
from oslo.utils import importutils
22
from neutron import context
23
from neutron.extensions import portbindings
24
from neutron.extensions import providernet as provider
25
from neutron import manager
26
from neutron.plugins.plumgrid.plumgrid_plugin import plumgrid_plugin
27
from neutron.tests.unit import _test_extension_portbindings as test_bindings
28
from neutron.tests.unit import test_db_plugin as test_plugin
31
PLUM_DRIVER = ('neutron.plugins.plumgrid.drivers.fake_plumlib.Plumlib')
32
FAKE_DIRECTOR = '1.1.1.1'
34
FAKE_USERNAME = 'fake_admin'
35
FAKE_PASSWORD = 'fake_password'
39
class PLUMgridPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
40
_plugin_name = ('neutron.plugins.plumgrid.plumgrid_plugin.'
41
'plumgrid_plugin.NeutronPluginPLUMgridV2')
44
def mocked_plumlib_init(self):
45
director_plumgrid = FAKE_DIRECTOR
46
director_port = FAKE_PORT
47
director_username = FAKE_USERNAME
48
director_password = FAKE_PASSWORD
49
timeout = FAKE_TIMEOUT
50
self._plumlib = importutils.import_object(PLUM_DRIVER)
51
self._plumlib.director_conn(director_plumgrid,
52
director_port, timeout,
56
with mock.patch.object(plumgrid_plugin.NeutronPluginPLUMgridV2,
57
'plumgrid_init', new=mocked_plumlib_init):
58
super(PLUMgridPluginV2TestCase, self).setUp(self._plugin_name)
61
super(PLUMgridPluginV2TestCase, self).tearDown()
64
class TestPlumgridPluginNetworksV2(test_plugin.TestNetworksV2,
65
PLUMgridPluginV2TestCase):
69
class TestPlumgridV2HTTPResponse(test_plugin.TestV2HTTPResponse,
70
PLUMgridPluginV2TestCase):
74
class TestPlumgridPluginPortsV2(test_plugin.TestPortsV2,
75
PLUMgridPluginV2TestCase):
76
def test_range_allocation(self):
77
self.skipTest("Plugin does not support Neutron allocation process")
80
class TestPlumgridPluginSubnetsV2(test_plugin.TestSubnetsV2,
81
PLUMgridPluginV2TestCase):
83
'test_create_subnet_default_gw_conflict_allocation_pool_returns_409',
84
'test_create_subnet_defaults', 'test_create_subnet_gw_values',
85
'test_create_subnet_ipv6_gw_values',
86
'test_update_subnet_gateway_in_allocation_pool_returns_409',
87
'test_update_subnet_allocation_pools',
88
'test_update_subnet_allocation_pools_invalid_pool_for_cidr')
91
if self._testMethodName in self._unsupported:
92
self.skipTest("Plugin does not support Neutron allocation process")
93
super(TestPlumgridPluginSubnetsV2, self).setUp()
95
def test_subnet_admin_delete(self):
96
plugin = manager.NeutronManager.get_plugin()
97
admin_context = context.get_admin_context()
98
tenant_context = context.Context('', 'not_admin')
100
network1 = self._fake_network('network1')
101
network1_ret = plugin.create_network(tenant_context, network1)
103
subnet1 = self._fake_subnet(network1_ret['id'])
104
plugin.create_subnet(tenant_context, subnet1)
105
net_db = plugin.get_network(admin_context, network1_ret['id'])
107
self.assertEqual(network1_ret['tenant_id'], net_db["tenant_id"])
109
def _fake_network(self, name):
110
data = {'network': {'name': name,
111
'admin_state_up': False,
113
'router:external': [],
114
'provider:network_type': None,
115
'provider:segmentation_id': None,
116
'provider:physical_network': None}}
119
def _fake_subnet(self, net_id):
120
allocation_pools = [{'start': '10.0.0.2',
121
'end': '10.0.0.254'}]
122
return {'subnet': {'name': net_id,
123
'network_id': net_id,
124
'gateway_ip': '10.0.0.1',
125
'dns_nameservers': ['10.0.0.2'],
127
'cidr': '10.0.0.0/24',
128
'allocation_pools': allocation_pools,
133
class TestPlumgridPluginPortBinding(PLUMgridPluginV2TestCase,
134
test_bindings.PortBindingsTestCase):
135
VIF_TYPE = portbindings.VIF_TYPE_IOVISOR
136
HAS_PORT_FILTER = True
139
super(TestPlumgridPluginPortBinding, self).setUp()
142
class TestPlumgridNetworkAdminState(PLUMgridPluginV2TestCase):
143
def test_network_admin_state(self):
144
name = 'network_test'
145
admin_status_up = False
146
tenant_id = 'tenant_test'
147
network = {'network': {'name': name,
148
'admin_state_up': admin_status_up,
149
'tenant_id': tenant_id}}
150
plugin = manager.NeutronManager.get_plugin()
151
self.assertEqual(plugin._network_admin_state(network), network)
154
class TestPlumgridAllocationPool(PLUMgridPluginV2TestCase):
155
def test_allocate_pools_for_subnet(self):
157
gateway_ip = '10.0.0.254'
158
subnet = {'gateway_ip': gateway_ip,
161
allocation_pool = [{"start": '10.0.0.2',
162
"end": '10.0.0.253'}]
164
plugin = manager.NeutronManager.get_plugin()
165
pool = plugin._allocate_pools_for_subnet(context, subnet)
166
self.assertEqual(allocation_pool, pool)
168
def test_conflict_dhcp_gw_ip(self):
170
gateway_ip = '10.0.0.1'
171
subnet = {'gateway_ip': gateway_ip,
174
allocation_pool = [{"start": '10.0.0.3',
175
"end": '10.0.0.254'}]
177
plugin = manager.NeutronManager.get_plugin()
178
pool = plugin._allocate_pools_for_subnet(context, subnet)
179
self.assertEqual(allocation_pool, pool)
182
class TestPlumgridProvidernet(PLUMgridPluginV2TestCase):
184
def test_create_provider_network(self):
186
data = {'network': {'name': 'net1',
187
'admin_state_up': True,
188
'tenant_id': tenant_id,
189
provider.NETWORK_TYPE: 'vlan',
190
provider.SEGMENTATION_ID: 3333,
191
provider.PHYSICAL_NETWORK: 'phy3333'}}
193
network_req = self.new_create_request('networks', data, self.fmt)
194
net = self.deserialize(self.fmt, network_req.get_response(self.api))
195
plumlib = importutils.import_object(PLUM_DRIVER)
196
plumlib.create_network(tenant_id, net, data)
197
self.assertEqual(net['network'][provider.NETWORK_TYPE], 'vlan')
198
self.assertEqual(net['network'][provider.SEGMENTATION_ID], 3333)
199
self.assertEqual(net['network'][provider.PHYSICAL_NETWORK], 'phy3333')
202
class TestDisassociateFloatingIP(PLUMgridPluginV2TestCase):
204
def test_disassociate_floating_ip(self):
206
tenant_id = "94eb42de4e331"
207
fip_net_id = "b843d18245678"
208
fip_addr = "10.0.3.44"
209
fip_id = "e623679734051"
210
fip = {"router_id": "94eb42de4e331",
211
"tenant_id": tenant_id,
212
"floating_network_id": fip_net_id,
213
"fixed_ip_address": "192.168.8.2",
214
"floating_ip_address": fip_addr,
217
plumlib = importutils.import_object(PLUM_DRIVER)
218
fip_res = plumlib.disassociate_floatingips(fip, port_id)
219
self.assertEqual(fip_res["id"], fip_id)
220
self.assertEqual(fip_res["floating_ip_address"], fip_addr)
221
self.assertEqual(fip_res["floating_network_id"], fip_net_id)