 3fc6879894
			
		
	
	3fc6879894
	
	
	
		
			
			* Make modules use getLogger(__name__) and log to the result Change-Id: Ib6d69b4be140ec89affc86ed11e65e422d551df1
		
			
				
	
	
		
			327 lines
		
	
	
		
			14 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			327 lines
		
	
	
		
			14 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
| # vim: tabstop=4 shiftwidth=4 softtabstop=4
 | |
| 
 | |
| # Copyright 2011 OpenStack LLC.
 | |
| # All Rights Reserved.
 | |
| #
 | |
| #    Licensed under the Apache License, Version 2.0 (the "License"); you may
 | |
| #    not use this file except in compliance with the License. You may obtain
 | |
| #    a copy of the License at
 | |
| #
 | |
| #         http://www.apache.org/licenses/LICENSE-2.0
 | |
| #
 | |
| #    Unless required by applicable law or agreed to in writing, software
 | |
| #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 | |
| #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 | |
| #    License for the specific language governing permissions and limitations
 | |
| #    under the License.
 | |
| 
 | |
| from nova import exception
 | |
| from nova import log as logging
 | |
| from nova.network import model
 | |
| from nova import test
 | |
| from nova.tests import fake_network_cache_model
 | |
| 
 | |
| LOG = logging.getLogger(__name__)
 | |
| 
 | |
| 
 | |
| class RouteTests(test.TestCase):
 | |
|     def test_create_route_with_attrs(self):
 | |
|         route = fake_network_cache_model.new_route()
 | |
|         ip = fake_network_cache_model.new_ip(dict(address='192.168.1.1'))
 | |
|         self.assertEqual(route['cidr'], '0.0.0.0/24')
 | |
|         self.assertEqual(route['gateway']['address'], '192.168.1.1')
 | |
|         self.assertEqual(route['interface'], 'eth0')
 | |
| 
 | |
|     def test_routes_equal(self):
 | |
|         route1 = fake_network_cache_model.new_route()
 | |
|         route2 = fake_network_cache_model.new_route()
 | |
|         self.assertEqual(route1, route2)
 | |
| 
 | |
|     def test_routes_not_equal(self):
 | |
|         route1 = fake_network_cache_model.new_route()
 | |
|         route2 = fake_network_cache_model.new_route(dict(cidr='1.1.1.1/24'))
 | |
|         self.assertNotEqual(route1, route2)
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         route = model.Route.hydrate(
 | |
|                 {'gateway': fake_network_cache_model.new_ip(
 | |
|                                 dict(address='192.168.1.1'))})
 | |
|         self.assertEqual(route['cidr'], None)
 | |
|         self.assertEqual(route['gateway']['address'], '192.168.1.1')
 | |
|         self.assertEqual(route['interface'], None)
 | |
| 
 | |
| 
 | |
| class FixedIPTests(test.TestCase):
 | |
|     def test_createnew_fixed_ip_with_attrs(self):
 | |
|         fixed_ip = model.FixedIP(address='192.168.1.100')
 | |
|         self.assertEqual(fixed_ip['address'], '192.168.1.100')
 | |
|         self.assertEqual(fixed_ip['floating_ips'], [])
 | |
|         self.assertEqual(fixed_ip['type'], 'fixed')
 | |
|         self.assertEqual(fixed_ip['version'], 4)
 | |
| 
 | |
|     def test_create_fixed_ipv6(self):
 | |
|         fixed_ip = model.FixedIP(address='::1')
 | |
|         self.assertEqual(fixed_ip['address'], '::1')
 | |
|         self.assertEqual(fixed_ip['floating_ips'], [])
 | |
|         self.assertEqual(fixed_ip['type'], 'fixed')
 | |
|         self.assertEqual(fixed_ip['version'], 6)
 | |
| 
 | |
|     def test_create_fixed_bad_ip_fails(self):
 | |
|         self.assertRaises(exception.InvalidIpAddressError,
 | |
|                           model.FixedIP,
 | |
|                           address='picklespicklespickles')
 | |
| 
 | |
|     def test_equate_two_fixed_ips(self):
 | |
|         fixed_ip = model.FixedIP(address='::1')
 | |
|         fixed_ip2 = model.FixedIP(address='::1')
 | |
|         self.assertEqual(fixed_ip, fixed_ip2)
 | |
| 
 | |
|     def test_equate_two_dissimilar_fixed_ips_fails(self):
 | |
|         fixed_ip = model.FixedIP(address='::1')
 | |
|         fixed_ip2 = model.FixedIP(address='::2')
 | |
|         self.assertNotEqual(fixed_ip, fixed_ip2)
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         fixed_ip = model.FixedIP.hydrate({})
 | |
|         self.assertEqual(fixed_ip['floating_ips'], [])
 | |
|         self.assertEqual(fixed_ip['address'], None)
 | |
|         self.assertEqual(fixed_ip['type'], 'fixed')
 | |
|         self.assertEqual(fixed_ip['version'], None)
 | |
| 
 | |
|     def test_add_floating_ip(self):
 | |
|         fixed_ip = model.FixedIP(address='192.168.1.100')
 | |
|         fixed_ip.add_floating_ip('192.168.1.101')
 | |
|         self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101'])
 | |
| 
 | |
|     def test_add_floating_ip_repeatedly_only_one_instance(self):
 | |
|         fixed_ip = model.FixedIP(address='192.168.1.100')
 | |
|         for i in xrange(10):
 | |
|             fixed_ip.add_floating_ip('192.168.1.101')
 | |
|         self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101'])
 | |
| 
 | |
| 
 | |
| class SubnetTests(test.TestCase):
 | |
|     def test_create_subnet_with_attrs(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
| 
 | |
|         route1 = fake_network_cache_model.new_route()
 | |
| 
 | |
|         self.assertEqual(subnet['cidr'], '10.10.0.0/24')
 | |
|         self.assertEqual(subnet['dns'],
 | |
|                 [fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
 | |
|                  fake_network_cache_model.new_ip(dict(address='2.3.4.5'))])
 | |
|         self.assertEqual(subnet['gateway']['address'], '10.10.0.1')
 | |
|         self.assertEqual(subnet['ips'],
 | |
|                 [fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.2')),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                             dict(address='10.10.0.3'))])
 | |
|         self.assertEqual(subnet['routes'], [route1])
 | |
|         self.assertEqual(subnet['version'], 4)
 | |
| 
 | |
|     def test_add_route(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         route1 = fake_network_cache_model.new_route()
 | |
|         route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'})
 | |
|         subnet.add_route(route2)
 | |
|         self.assertEqual(subnet['routes'], [route1, route2])
 | |
| 
 | |
|     def test_add_route_a_lot(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         route1 = fake_network_cache_model.new_route()
 | |
|         route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'})
 | |
|         for i in xrange(10):
 | |
|             subnet.add_route(route2)
 | |
|         self.assertEqual(subnet['routes'], [route1, route2])
 | |
| 
 | |
|     def test_add_dns(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         dns = fake_network_cache_model.new_ip(dict(address='9.9.9.9'))
 | |
|         subnet.add_dns(dns)
 | |
|         self.assertEqual(subnet['dns'],
 | |
|                 [fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
 | |
|                  fake_network_cache_model.new_ip(dict(address='2.3.4.5')),
 | |
|                  fake_network_cache_model.new_ip(dict(address='9.9.9.9'))])
 | |
| 
 | |
|     def test_add_dns_a_lot(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         for i in xrange(10):
 | |
|             subnet.add_dns(fake_network_cache_model.new_ip(
 | |
|                     dict(address='9.9.9.9')))
 | |
|         self.assertEqual(subnet['dns'],
 | |
|                 [fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
 | |
|                  fake_network_cache_model.new_ip(dict(address='2.3.4.5')),
 | |
|                  fake_network_cache_model.new_ip(dict(address='9.9.9.9'))])
 | |
| 
 | |
|     def test_add_ip(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         subnet.add_ip(fake_network_cache_model.new_ip(
 | |
|                 dict(address='192.168.1.102')))
 | |
|         self.assertEqual(subnet['ips'],
 | |
|                 [fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.2')),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.3')),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                         dict(address='192.168.1.102'))])
 | |
| 
 | |
|     def test_add_ip_a_lot(self):
 | |
|         subnet = fake_network_cache_model.new_subnet()
 | |
|         for i in xrange(10):
 | |
|             subnet.add_ip(fake_network_cache_model.new_ip(
 | |
|                         dict(address='192.168.1.102')))
 | |
|         self.assertEqual(subnet['ips'],
 | |
|                 [fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.2')),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.3')),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                         dict(address='192.168.1.102'))])
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         subnet_dict = {
 | |
|             'cidr': '255.255.255.0',
 | |
|             'dns': [fake_network_cache_model.new_ip(dict(address='1.1.1.1'))],
 | |
|             'ips': [fake_network_cache_model.new_ip(dict(address='2.2.2.2'))],
 | |
|             'routes': [fake_network_cache_model.new_route()],
 | |
|             'version': 4,
 | |
|             'gateway': fake_network_cache_model.new_ip(
 | |
|                             dict(address='3.3.3.3'))}
 | |
|         subnet = model.Subnet.hydrate(subnet_dict)
 | |
| 
 | |
|         self.assertEqual(subnet['cidr'], '255.255.255.0')
 | |
|         self.assertEqual(subnet['dns'], [fake_network_cache_model.new_ip(
 | |
|                                          dict(address='1.1.1.1'))])
 | |
|         self.assertEqual(subnet['gateway']['address'], '3.3.3.3')
 | |
|         self.assertEqual(subnet['ips'], [fake_network_cache_model.new_ip(
 | |
|                                          dict(address='2.2.2.2'))])
 | |
|         self.assertEqual(subnet['routes'], [
 | |
|                     fake_network_cache_model.new_route()])
 | |
|         self.assertEqual(subnet['version'], 4)
 | |
| 
 | |
| 
 | |
| class NetworkTests(test.TestCase):
 | |
|     def test_create_network(self):
 | |
|         network = fake_network_cache_model.new_network()
 | |
|         self.assertEqual(network['id'], 1)
 | |
|         self.assertEqual(network['bridge'], 'br0')
 | |
|         self.assertEqual(network['label'], 'public')
 | |
|         self.assertEqual(network['subnets'],
 | |
|                 [fake_network_cache_model.new_subnet(),
 | |
|                  fake_network_cache_model.new_subnet(
 | |
|                         dict(cidr='255.255.255.255'))])
 | |
| 
 | |
|     def test_add_subnet(self):
 | |
|         network = fake_network_cache_model.new_network()
 | |
|         network.add_subnet(fake_network_cache_model.new_subnet(
 | |
|                     dict(cidr='0.0.0.0')))
 | |
|         self.assertEqual(network['subnets'],
 | |
|                 [fake_network_cache_model.new_subnet(),
 | |
|                  fake_network_cache_model.new_subnet(
 | |
|                         dict(cidr='255.255.255.255')),
 | |
|                  fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))])
 | |
| 
 | |
|     def test_add_subnet_a_lot(self):
 | |
|         network = fake_network_cache_model.new_network()
 | |
|         for i in xrange(10):
 | |
|             network.add_subnet(fake_network_cache_model.new_subnet(
 | |
|                     dict(cidr='0.0.0.0')))
 | |
|         self.assertEqual(network['subnets'],
 | |
|                 [fake_network_cache_model.new_subnet(),
 | |
|                  fake_network_cache_model.new_subnet(
 | |
|                         dict(cidr='255.255.255.255')),
 | |
|                  fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))])
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         new_network = dict(
 | |
|             id=1,
 | |
|             bridge='br0',
 | |
|             label='public',
 | |
|             subnets=[fake_network_cache_model.new_subnet(),
 | |
|                 fake_network_cache_model.new_subnet(
 | |
|                         dict(cidr='255.255.255.255'))])
 | |
|         network = model.Network.hydrate(fake_network_cache_model.new_network())
 | |
| 
 | |
|         self.assertEqual(network['id'], 1)
 | |
|         self.assertEqual(network['bridge'], 'br0')
 | |
|         self.assertEqual(network['label'], 'public')
 | |
|         self.assertEqual(network['subnets'],
 | |
|                 [fake_network_cache_model.new_subnet(),
 | |
|                  fake_network_cache_model.new_subnet(
 | |
|                         dict(cidr='255.255.255.255'))])
 | |
| 
 | |
| 
 | |
| class VIFTests(test.TestCase):
 | |
|     def test_create_vif(self):
 | |
|         vif = fake_network_cache_model.new_vif()
 | |
|         self.assertEqual(vif['id'], 1)
 | |
|         self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
 | |
|         self.assertEqual(vif['network'],
 | |
|                 fake_network_cache_model.new_network())
 | |
| 
 | |
|     def test_vif_get_fixed_ips(self):
 | |
|         vif = fake_network_cache_model.new_vif()
 | |
|         fixed_ips = vif.fixed_ips()
 | |
|         ips = [fake_network_cache_model.new_ip(dict(address='10.10.0.2')),
 | |
|                 fake_network_cache_model.new_ip(
 | |
|                         dict(address='10.10.0.3'))] * 2
 | |
|         self.assertEqual(fixed_ips, ips)
 | |
| 
 | |
|     def test_vif_get_floating_ips(self):
 | |
|         vif = fake_network_cache_model.new_vif()
 | |
|         vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1')
 | |
|         floating_ips = vif.floating_ips()
 | |
|         self.assertEqual(floating_ips, ['192.168.1.1'])
 | |
| 
 | |
|     def test_vif_get_labeled_ips(self):
 | |
|         vif = fake_network_cache_model.new_vif()
 | |
|         labeled_ips = vif.labeled_ips()
 | |
|         ip_dict = {
 | |
|             'network_id': 1,
 | |
|             'ips': [fake_network_cache_model.new_ip(
 | |
|                         {'address': '10.10.0.2'}),
 | |
|                     fake_network_cache_model.new_ip(
 | |
|                         {'address': '10.10.0.3'})] * 2,
 | |
|             'network_label': 'public'}
 | |
|         self.assertEqual(labeled_ips, ip_dict)
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         new_vif = dict(
 | |
|             id=1,
 | |
|             address='127.0.0.1',
 | |
|             network=fake_network_cache_model.new_network())
 | |
|         vif = model.VIF.hydrate(fake_network_cache_model.new_vif())
 | |
|         self.assertEqual(vif['id'], 1)
 | |
|         self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
 | |
|         self.assertEqual(vif['network'],
 | |
|                 fake_network_cache_model.new_network())
 | |
| 
 | |
| 
 | |
| class NetworkInfoTests(test.TestCase):
 | |
|     def test_create_model(self):
 | |
|         ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(),
 | |
|                 fake_network_cache_model.new_vif(
 | |
|                     {'address':'bb:bb:bb:bb:bb:bb'})])
 | |
|         self.assertEqual(ninfo.fixed_ips(),
 | |
|                 [fake_network_cache_model.new_ip({'address': '10.10.0.2'}),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                     {'address': '10.10.0.3'})] * 4)
 | |
| 
 | |
|     def test_get_floating_ips(self):
 | |
|         vif = fake_network_cache_model.new_vif()
 | |
|         vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1')
 | |
|         ninfo = model.NetworkInfo([vif,
 | |
|                 fake_network_cache_model.new_vif(
 | |
|                     {'address':'bb:bb:bb:bb:bb:bb'})])
 | |
|         self.assertEqual(ninfo.floating_ips(), ['192.168.1.1'])
 | |
| 
 | |
|     def test_hydrate(self):
 | |
|         ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(),
 | |
|                 fake_network_cache_model.new_vif(
 | |
|                         {'address':'bb:bb:bb:bb:bb:bb'})])
 | |
|         deserialized = model.NetworkInfo.hydrate(ninfo)
 | |
|         self.assertEqual(ninfo.fixed_ips(),
 | |
|                 [fake_network_cache_model.new_ip({'address': '10.10.0.2'}),
 | |
|                  fake_network_cache_model.new_ip(
 | |
|                         {'address': '10.10.0.3'})] * 4)
 |