1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at:
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
18 from snaps.config.router import RouterConfigError, RouterConfig
19 from snaps.openstack import create_network
20 from snaps.openstack import create_router
21 from snaps.openstack.create_network import (
22 NetworkSettings, PortSettings)
23 from snaps.openstack.create_network import OpenStackNetwork
24 from snaps.openstack.create_router import RouterSettings
25 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
26 from snaps.openstack.utils import neutron_utils, settings_utils
28 __author__ = 'mmakati'
30 cidr1 = '10.200.201.0/24'
31 cidr2 = '10.200.202.0/24'
32 static_gateway_ip1 = '10.200.201.1'
33 static_gateway_ip2 = '10.200.202.1'
36 class RouterSettingsUnitTests(unittest.TestCase):
38 Class for testing the RouterSettings class
41 def test_no_params(self):
42 with self.assertRaises(RouterConfigError):
45 def test_empty_config(self):
46 with self.assertRaises(RouterConfigError):
47 RouterSettings(**dict())
49 def test_name_only(self):
50 settings = RouterSettings(name='foo')
51 self.assertEqual('foo', settings.name)
52 self.assertIsNone(settings.project_name)
53 self.assertIsNone(settings.external_gateway)
54 self.assertTrue(settings.admin_state_up)
55 self.assertIsNone(settings.enable_snat)
56 self.assertIsNotNone(settings.internal_subnets)
57 self.assertTrue(isinstance(settings.internal_subnets, list))
58 self.assertEqual(0, len(settings.internal_subnets))
59 self.assertIsNotNone(settings.port_settings)
60 self.assertTrue(isinstance(settings.port_settings, list))
61 self.assertEqual(0, len(settings.port_settings))
63 def test_config_with_name_only(self):
64 settings = RouterSettings(**{'name': 'foo'})
65 self.assertEqual('foo', settings.name)
66 self.assertIsNone(settings.project_name)
67 self.assertIsNone(settings.external_gateway)
68 self.assertTrue(settings.admin_state_up)
69 self.assertIsNone(settings.enable_snat)
70 self.assertIsNotNone(settings.internal_subnets)
71 self.assertTrue(isinstance(settings.internal_subnets, list))
72 self.assertEqual(0, len(settings.internal_subnets))
73 self.assertIsNotNone(settings.port_settings)
74 self.assertTrue(isinstance(settings.port_settings, list))
75 self.assertEqual(0, len(settings.port_settings))
78 port_settings = PortSettings(name='foo', network_name='bar')
79 settings = RouterSettings(
80 name='foo', project_name='bar', external_gateway='foo_gateway',
81 admin_state_up=True, enable_snat=False,
82 internal_subnets=['10.0.0.1/24'], interfaces=[port_settings])
83 self.assertEqual('foo', settings.name)
84 self.assertEqual('bar', settings.project_name)
85 self.assertEqual('foo_gateway', settings.external_gateway)
86 self.assertTrue(settings.admin_state_up)
87 self.assertFalse(settings.enable_snat)
88 self.assertIsNotNone(settings.internal_subnets)
89 self.assertTrue(isinstance(settings.internal_subnets, list))
90 self.assertEqual(1, len(settings.internal_subnets))
91 self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
92 self.assertEqual([port_settings], settings.port_settings)
94 def test_config_all(self):
95 settings = RouterSettings(
96 **{'name': 'foo', 'project_name': 'bar',
97 'external_gateway': 'foo_gateway', 'admin_state_up': True,
98 'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
100 [{'port': {'name': 'foo-port',
101 'network_name': 'bar-net'}}]})
102 self.assertEqual('foo', settings.name)
103 self.assertEqual('bar', settings.project_name)
104 self.assertEqual('foo_gateway', settings.external_gateway)
105 self.assertTrue(settings.admin_state_up)
106 self.assertFalse(settings.enable_snat)
107 self.assertIsNotNone(settings.internal_subnets)
108 self.assertTrue(isinstance(settings.internal_subnets, list))
109 self.assertEqual(1, len(settings.internal_subnets))
110 self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
111 self.assertEqual([PortSettings(**{'name': 'foo-port',
112 'network_name': 'bar-net'})],
113 settings.port_settings)
116 class CreateRouterSuccessTests(OSIntegrationTestCase):
118 Class for testing routers with various positive scenarios expected to
124 Initializes objects used for router testing
126 super(self.__class__, self).__start__()
128 self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
129 self.router_creator = None
130 self.network_creator1 = None
131 self.network_creator2 = None
132 self.neutron = neutron_utils.neutron_client(self.os_creds)
136 Cleans the remote OpenStack objects used for router testing
138 if self.router_creator:
139 self.router_creator.clean()
141 if self.network_creator1:
142 self.network_creator1.clean()
144 if self.network_creator2:
145 self.network_creator2.clean()
147 super(self.__class__, self).__clean__()
149 def test_create_router_vanilla(self):
151 Test creation of a most basic router with minimal options.
153 router_settings = RouterConfig(
154 name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
156 self.router_creator = create_router.OpenStackRouter(self.os_creds,
158 self.router_creator.create()
160 router = neutron_utils.get_router(self.neutron,
161 router_settings=router_settings)
162 self.assertIsNotNone(router)
164 self.assertEqual(self.router_creator.get_router(), router)
166 self.check_router_recreation(router, router_settings)
168 def test_create_router_admin_user_to_new_project(self):
170 Test creation of a most basic router with the admin user pointing
173 router_settings = RouterConfig(
174 name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
175 project_name=self.os_creds.project_name)
177 self.router_creator = create_router.OpenStackRouter(
178 self.admin_os_creds, router_settings)
179 self.router_creator.create()
181 router = neutron_utils.get_router(self.neutron,
182 router_settings=router_settings)
183 self.assertIsNotNone(router)
185 self.assertEqual(self.router_creator.get_router(), router)
187 self.check_router_recreation(router, router_settings)
189 def test_create_router_new_user_to_admin_project(self):
191 Test creation of a most basic router with the new user pointing
192 to the admin project.
194 router_settings = RouterConfig(
195 name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
196 project_name=self.admin_os_creds.project_name)
198 self.router_creator = create_router.OpenStackRouter(
199 self.os_creds, router_settings)
200 self.router_creator.create()
202 router = neutron_utils.get_router(self.neutron,
203 router_settings=router_settings)
204 self.assertIsNotNone(router)
206 self.assertEqual(self.router_creator.get_router(), router)
208 self.check_router_recreation(router, router_settings)
210 def test_create_delete_router(self):
212 Test that clean() will not raise an exception if the router is deleted
215 self.router_settings = RouterConfig(
216 name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
218 self.router_creator = create_router.OpenStackRouter(
219 self.os_creds, self.router_settings)
220 created_router = self.router_creator.create()
221 self.assertIsNotNone(created_router)
222 retrieved_router = neutron_utils.get_router(
223 self.neutron, router_settings=self.router_settings)
224 self.assertIsNotNone(retrieved_router)
226 neutron_utils.delete_router(self.neutron, created_router)
228 retrieved_router = neutron_utils.get_router(
229 self.neutron, router_settings=self.router_settings)
230 self.assertIsNone(retrieved_router)
232 # Should not raise an exception
233 self.router_creator.clean()
235 def test_create_router_admin_state_false(self):
237 Test creation of a basic router with admin state down.
239 router_settings = RouterConfig(
240 name=self.guid + '-pub-router', admin_state_up=False)
242 self.router_creator = create_router.OpenStackRouter(self.os_creds,
244 self.router_creator.create()
246 router = neutron_utils.get_router(self.neutron,
247 router_settings=router_settings)
248 self.assertIsNotNone(router)
250 self.assertEqual(self.router_creator.get_router(), router)
252 self.check_router_recreation(router, router_settings)
254 def test_create_router_admin_state_True(self):
256 Test creation of a basic router with admin state Up.
258 router_settings = RouterConfig(
259 name=self.guid + '-pub-router', admin_state_up=True)
261 self.router_creator = create_router.OpenStackRouter(
262 self.os_creds, router_settings)
263 self.router_creator.create()
265 router = neutron_utils.get_router(
266 self.neutron, router_settings=router_settings)
267 self.assertIsNotNone(router)
269 self.assertEqual(self.router_creator.get_router(), router)
271 self.check_router_recreation(router, router_settings)
273 def test_create_router_private_network(self):
275 Test creation of a router connected with two private networks and no
278 network_settings1 = NetworkSettings(
279 name=self.guid + '-pub-net1',
281 create_network.SubnetSettings(
282 cidr=cidr1, name=self.guid + '-pub-subnet1',
283 gateway_ip=static_gateway_ip1)])
284 network_settings2 = NetworkSettings(
285 name=self.guid + '-pub-net2',
287 create_network.SubnetSettings(
288 cidr=cidr2, name=self.guid + '-pub-subnet2',
289 gateway_ip=static_gateway_ip2)])
291 self.network_creator1 = OpenStackNetwork(self.os_creds,
293 self.network_creator2 = OpenStackNetwork(self.os_creds,
296 self.network_creator1.create()
297 self.network_creator2.create()
300 create_network.PortSettings(
301 name=self.guid + '-port1',
304 network_settings1.subnet_settings[0].name,
305 'ip': static_gateway_ip1
307 network_name=network_settings1.name,
308 project_name=self.os_creds.project_name),
309 create_network.PortSettings(
310 name=self.guid + '-port2',
312 'subnet_name': network_settings2.subnet_settings[0].name,
313 'ip': static_gateway_ip2
315 network_name=network_settings2.name,
316 project_name=self.os_creds.project_name)]
318 router_settings = RouterConfig(
319 name=self.guid + '-pub-router', port_settings=port_settings)
320 self.router_creator = create_router.OpenStackRouter(self.os_creds,
322 self.router_creator.create()
324 router = neutron_utils.get_router(
325 self.neutron, router_settings=router_settings)
327 self.assertEqual(router, self.router_creator.get_router())
329 # Instantiate second identical creator to ensure a second router
330 # has not been created
331 router_creator2 = create_router.OpenStackRouter(
332 self.os_creds, router_settings)
333 router2 = router_creator2.create()
334 self.assertIsNotNone(self.router_creator.get_router(), router2)
336 self.check_router_recreation(router2, router_settings)
338 def test_create_router_external_network(self):
340 Test creation of a router connected to an external network and a
343 network_settings = NetworkSettings(
344 name=self.guid + '-pub-net1',
346 create_network.SubnetSettings(
347 cidr=cidr1, name=self.guid + '-pub-subnet1',
348 gateway_ip=static_gateway_ip1)])
349 self.network_creator1 = OpenStackNetwork(self.os_creds,
351 self.network_creator1.create()
354 create_network.PortSettings(
355 name=self.guid + '-port1',
357 'subnet_name': network_settings.subnet_settings[0].name,
358 'ip': static_gateway_ip1}],
359 network_name=network_settings.name,
360 project_name=self.os_creds.project_name)]
362 router_settings = RouterConfig(
363 name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
364 port_settings=port_settings)
365 self.router_creator = create_router.OpenStackRouter(
366 self.os_creds, router_settings)
367 self.router_creator.create()
369 router = neutron_utils.get_router(
370 self.neutron, router_settings=router_settings)
372 self.assertEquals(router, self.router_creator.get_router())
374 self.check_router_recreation(router, router_settings)
376 def check_router_recreation(self, router, orig_settings):
378 Validates the derived RouterConfig with the original
379 :param router: the Router domain object to test
380 :param orig_settings: the original RouterConfig object that was
381 responsible for creating the router
382 :return: the derived RouterConfig object
384 derived_settings = settings_utils.create_router_settings(
385 self.neutron, router)
386 self.assertIsNotNone(derived_settings)
388 orig_settings.enable_snat, derived_settings.enable_snat)
389 self.assertEqual(orig_settings.external_gateway,
390 derived_settings.external_gateway)
391 self.assertEqual(orig_settings.name, derived_settings.name)
392 self.assertEqual(orig_settings.internal_subnets,
393 derived_settings.internal_subnets)
395 if orig_settings.external_gateway:
396 self.assertEqual(len(orig_settings.port_settings),
397 len(derived_settings.port_settings))
399 self.assertEqual(len(orig_settings.port_settings),
400 len(derived_settings.port_settings))
402 if len(orig_settings.port_settings) > 0:
403 self.assertEqual(orig_settings.port_settings[0].name,
404 derived_settings.port_settings[0].name)
406 if len(orig_settings.port_settings) > 1:
407 self.assertEqual(orig_settings.port_settings[1].name,
408 derived_settings.port_settings[1].name)
410 return derived_settings
413 class CreateRouterNegativeTests(OSIntegrationTestCase):
415 Class for testing routers with various negative scenarios expected to fail.
420 Initializes objects used for router testing
422 super(self.__class__, self).__start__()
424 self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
425 self.router_creator = None
429 Cleans the remote OpenStack objects used for router testing
431 if self.router_creator:
432 self.router_creator.clean()
434 super(self.__class__, self).__clean__()
436 def test_create_router_noname(self):
438 Test creating a router without a name.
440 with self.assertRaises(RouterConfigError):
441 router_settings = RouterConfig(
442 name=None, external_gateway=self.ext_net_name)
443 self.router_creator = create_router.OpenStackRouter(
444 self.os_creds, router_settings)
445 self.router_creator.create()
447 def test_create_router_invalid_gateway_name(self):
449 Test creating a router without a valid network gateway name.
451 with self.assertRaises(RouterConfigError):
452 router_settings = RouterConfig(
453 name=self.guid + '-pub-router',
454 external_gateway="Invalid_name")
455 self.router_creator = create_router.OpenStackRouter(
456 self.os_creds, router_settings)
457 self.router_creator.create()