408 lines
16 KiB
Python
408 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2014 Red Hat, Inc.
|
|
#
|
|
# 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.
|
|
|
|
import tempfile
|
|
|
|
from oslo_concurrency import processutils
|
|
|
|
import os_net_config
|
|
from os_net_config import impl_eni
|
|
from os_net_config import objects
|
|
from os_net_config.tests import base
|
|
from os_net_config import utils
|
|
|
|
_AUTO = "auto eth0\n"
|
|
|
|
_v4_IFACE_NO_IP = _AUTO + "iface eth0 inet manual\n"
|
|
|
|
_V4_IFACE_STATIC_IP = _AUTO + """iface eth0 inet static
|
|
address 192.168.1.2
|
|
netmask 255.255.255.0
|
|
"""
|
|
|
|
_IFACE_HOTPLUG = """allow-hotplug eth0
|
|
iface eth0 inet static
|
|
address 192.168.1.2
|
|
netmask 255.255.255.0
|
|
"""
|
|
|
|
_V4_IFACE_STATIC_IP_MULTIPLE = (_V4_IFACE_STATIC_IP + _AUTO +
|
|
"""iface eth0 inet static
|
|
address 10.0.0.2
|
|
netmask 255.0.0.0
|
|
""")
|
|
|
|
_V6_IFACE_STATIC_IP = _AUTO + """iface eth0 inet6 static
|
|
address fe80::2677:3ff:fe7d:4c
|
|
netmask 128
|
|
"""
|
|
|
|
_V6_IFACE_STATIC_IP_MULTIPLE = (_V6_IFACE_STATIC_IP + _AUTO +
|
|
"""iface eth0 inet6 static
|
|
address 2001:abcd::2
|
|
netmask 64
|
|
""")
|
|
|
|
_IFACE_DHCP = _AUTO + "iface eth0 inet dhcp\n"
|
|
|
|
_OVS_PORT_BASE = _AUTO + "allow-br0 eth0\n"
|
|
|
|
_OVS_PORT_IFACE = _OVS_PORT_BASE + """iface eth0 inet manual
|
|
ovs_bridge br0
|
|
ovs_type OVSPort
|
|
"""
|
|
|
|
_OVS_BRIDGE_DHCP = """auto br0
|
|
allow-ovs br0
|
|
iface br0 inet dhcp
|
|
ovs_type OVSBridge
|
|
ovs_ports eth0
|
|
pre-up ip addr flush dev eth0
|
|
"""
|
|
|
|
_OVS_BRIDGE_DHCP_STANDALONE = _OVS_BRIDGE_DHCP + (
|
|
" ovs_extra set bridge br0 fail_mode=standalone "
|
|
"-- del-controller br0\n")
|
|
|
|
_OVS_BRIDGE_DHCP_SECURE = _OVS_BRIDGE_DHCP + \
|
|
" ovs_extra set bridge br0 fail_mode=secure\n"
|
|
|
|
_OVS_BRIDGE_DHCP_PRIMARY_INTERFACE = _OVS_BRIDGE_DHCP + \
|
|
" ovs_extra set bridge br0 other-config:hwaddr=a1:b2:c3:d4:e5\n"
|
|
|
|
_OVS_BRIDGE_DHCP_OVS_EXTRA = _OVS_BRIDGE_DHCP + \
|
|
" ovs_extra set bridge br0 other-config:hwaddr=a1:b2:c3:d4:e5" + \
|
|
" -- br-set-external-id br0 bridge-id br0\n"
|
|
|
|
|
|
_VLAN_NO_IP = """auto vlan5
|
|
iface vlan5 inet manual
|
|
vlan-raw-device eth0
|
|
"""
|
|
|
|
_VLAN_OVS_PORT = """auto vlan5
|
|
allow-br0 vlan5
|
|
iface vlan5 inet manual
|
|
ovs_bridge br0
|
|
ovs_type OVSIntPort
|
|
ovs_options tag=5
|
|
"""
|
|
|
|
_RTS = """up route add -net 172.19.0.0 netmask 255.255.255.0 gw 192.168.1.1
|
|
down route del -net 172.19.0.0 netmask 255.255.255.0 gw 192.168.1.1
|
|
up route add -net 172.20.0.0 netmask 255.255.255.0 gw 192.168.1.5 metric 100
|
|
down route del -net 172.20.0.0 netmask 255.255.255.0 gw 192.168.1.5 metric 100
|
|
"""
|
|
|
|
|
|
class TestENINetConfig(base.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestENINetConfig, self).setUp()
|
|
|
|
def stub_is_ovs_installed():
|
|
return True
|
|
self.stub_out('os_net_config.utils.is_ovs_installed',
|
|
stub_is_ovs_installed)
|
|
|
|
self.provider = impl_eni.ENINetConfig()
|
|
self.if_name = 'eth0'
|
|
|
|
def tearDown(self):
|
|
super(TestENINetConfig, self).tearDown()
|
|
|
|
def get_interface_config(self, name="eth0"):
|
|
return self.provider.interfaces[name]
|
|
|
|
def get_route_config(self):
|
|
return self.provider.routes[self.if_name]
|
|
|
|
def _default_interface(self, addr=[], rts=[], hotplug=False):
|
|
return objects.Interface(self.if_name, addresses=addr, routes=rts,
|
|
hotplug=hotplug)
|
|
|
|
def test_interface_no_ip(self):
|
|
interface = self._default_interface()
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_v4_IFACE_NO_IP, self.get_interface_config())
|
|
|
|
def test_add_interface_with_v4(self):
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = self._default_interface([v4_addr])
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V4_IFACE_STATIC_IP, self.get_interface_config())
|
|
|
|
def test_add_interface_with_hotplug(self):
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = self._default_interface(addr=[v4_addr], hotplug=True)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_IFACE_HOTPLUG, self.get_interface_config())
|
|
|
|
def test_add_interface_with_v4_multiple(self):
|
|
v4_addresses = [objects.Address('192.168.1.2/24'),
|
|
objects.Address('10.0.0.2/8')]
|
|
interface = self._default_interface(v4_addresses)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V4_IFACE_STATIC_IP_MULTIPLE,
|
|
self.get_interface_config())
|
|
|
|
def test_add_interface_with_v6(self):
|
|
v6_addr = objects.Address('fe80::2677:3ff:fe7d:4c')
|
|
interface = self._default_interface([v6_addr])
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V6_IFACE_STATIC_IP, self.get_interface_config())
|
|
|
|
def test_add_interface_with_v6_multiple(self):
|
|
v6_addresses = [objects.Address('fe80::2677:3ff:fe7d:4c'),
|
|
objects.Address('2001:abcd::2/64')]
|
|
interface = self._default_interface(v6_addresses)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V6_IFACE_STATIC_IP_MULTIPLE,
|
|
self.get_interface_config())
|
|
|
|
def test_add_interface_dhcp(self):
|
|
interface = self._default_interface()
|
|
interface.use_dhcp = True
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_IFACE_DHCP, self.get_interface_config())
|
|
|
|
def test_add_interface_with_both_v4_and_v6(self):
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
v6_addr = objects.Address('fe80::2677:3ff:fe7d:4c')
|
|
interface = self._default_interface([v4_addr, v6_addr])
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V4_IFACE_STATIC_IP + _V6_IFACE_STATIC_IP,
|
|
self.get_interface_config())
|
|
|
|
def test_add_ovs_port_interface(self):
|
|
interface = self._default_interface()
|
|
interface.ovs_port = True
|
|
interface.bridge_name = 'br0'
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
|
|
def test_network_with_routes(self):
|
|
route1 = objects.Route('192.168.1.1', '172.19.0.0/24')
|
|
route2 = objects.Route('192.168.1.5', '172.20.0.0/24',
|
|
route_options="metric 100")
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = self._default_interface([v4_addr], [route1, route2])
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_V4_IFACE_STATIC_IP, self.get_interface_config())
|
|
self.assertEqual(_RTS, self.get_route_config())
|
|
|
|
def test_network_ovs_bridge_with_dhcp(self):
|
|
interface = self._default_interface()
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface])
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP, self.provider.bridges['br0'])
|
|
|
|
def test_network_ovs_bridge_with_standalone_fail_mode(self):
|
|
interface = self._default_interface()
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface],
|
|
fail_mode='standalone')
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP_STANDALONE,
|
|
self.provider.bridges['br0'])
|
|
|
|
def test_network_ovs_bridge_with_secure_fail_mode(self):
|
|
interface = self._default_interface()
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface],
|
|
fail_mode='secure')
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP_SECURE,
|
|
self.provider.bridges['br0'])
|
|
|
|
def test_network_ovs_bridge_with_dhcp_and_primary_interface(self):
|
|
|
|
def test_interface_mac(name):
|
|
return "a1:b2:c3:d4:e5"
|
|
self.stub_out('os_net_config.utils.interface_mac', test_interface_mac)
|
|
|
|
interface = objects.Interface(self.if_name, primary=True)
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface])
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP_PRIMARY_INTERFACE,
|
|
self.provider.bridges['br0'])
|
|
|
|
def test_network_ovs_bridge_with_dhcp_and_primary_with_ovs_extra(self):
|
|
|
|
def test_interface_mac(name):
|
|
return "a1:b2:c3:d4:e5"
|
|
self.stub_out('os_net_config.utils.interface_mac', test_interface_mac)
|
|
|
|
interface = objects.Interface(self.if_name, primary=True)
|
|
ovs_extra = "br-set-external-id br0 bridge-id br0"
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface],
|
|
ovs_extra=[ovs_extra])
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP_OVS_EXTRA,
|
|
self.provider.bridges['br0'])
|
|
|
|
def test_network_ovs_bridge_with_dhcp_and_primary_with_ovs_format(self):
|
|
|
|
def test_interface_mac(name):
|
|
return "a1:b2:c3:d4:e5"
|
|
self.stub_out('os_net_config.utils.interface_mac', test_interface_mac)
|
|
|
|
interface = objects.Interface(self.if_name, primary=True)
|
|
ovs_extra = "br-set-external-id {name} bridge-id {name}"
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface],
|
|
ovs_extra=[ovs_extra])
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_interface(interface)
|
|
self.assertEqual(_OVS_PORT_IFACE, self.get_interface_config())
|
|
self.assertEqual(_OVS_BRIDGE_DHCP_OVS_EXTRA,
|
|
self.provider.bridges['br0'])
|
|
|
|
def test_vlan(self):
|
|
vlan = objects.Vlan('eth0', 5)
|
|
self.provider.add_vlan(vlan)
|
|
self.assertEqual(_VLAN_NO_IP, self.get_interface_config('vlan5'))
|
|
|
|
def test_vlan_mtu_1500(self):
|
|
vlan = objects.Vlan('eth0', 5, mtu=1500)
|
|
self.provider.add_vlan(vlan)
|
|
expected = _VLAN_NO_IP + ' mtu 1500\n'
|
|
self.assertEqual(expected, self.get_interface_config('vlan5'))
|
|
|
|
def test_vlan_ovs_bridge_int_port(self):
|
|
vlan = objects.Vlan('eth0', 5)
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[vlan])
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.add_vlan(vlan)
|
|
self.assertEqual(_VLAN_OVS_PORT, self.get_interface_config('vlan5'))
|
|
|
|
|
|
class TestENINetConfigApply(base.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestENINetConfigApply, self).setUp()
|
|
self.temp_config_file = tempfile.NamedTemporaryFile()
|
|
self.ifup_interface_names = []
|
|
|
|
def test_config_path(prefix):
|
|
return self.temp_config_file.name
|
|
self.stub_out(
|
|
'os_net_config.impl_eni._network_config_path', test_config_path)
|
|
|
|
def stub_is_ovs_installed():
|
|
return True
|
|
self.stub_out('os_net_config.utils.is_ovs_installed',
|
|
stub_is_ovs_installed)
|
|
|
|
def test_execute(*args, **kwargs):
|
|
if args[0] == '/sbin/ifup':
|
|
self.ifup_interface_names.append(args[1])
|
|
pass
|
|
|
|
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
|
|
|
|
self.provider = impl_eni.ENINetConfig()
|
|
|
|
def tearDown(self):
|
|
self.temp_config_file.close()
|
|
super(TestENINetConfigApply, self).tearDown()
|
|
|
|
def test_network_apply(self):
|
|
route1 = objects.Route('192.168.1.1', '172.19.0.0/24')
|
|
route2 = objects.Route('192.168.1.5', '172.20.0.0/24',
|
|
route_options="metric 100")
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = objects.Interface('eth0', addresses=[v4_addr],
|
|
routes=[route1, route2])
|
|
self.provider.add_interface(interface)
|
|
|
|
self.provider.apply()
|
|
iface_data = utils.get_file_data(self.temp_config_file.name)
|
|
self.assertEqual((_V4_IFACE_STATIC_IP + _RTS), iface_data)
|
|
self.assertIn('eth0', self.ifup_interface_names)
|
|
|
|
def test_apply_noactivate(self):
|
|
route1 = objects.Route('192.168.1.1', '172.19.0.0/24')
|
|
route2 = objects.Route('192.168.1.5', '172.20.0.0/24',
|
|
route_options="metric 100")
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = objects.Interface('eth0', addresses=[v4_addr],
|
|
routes=[route1, route2])
|
|
self.provider.add_interface(interface)
|
|
|
|
self.provider.apply(activate=False)
|
|
iface_data = utils.get_file_data(self.temp_config_file.name)
|
|
self.assertEqual((_V4_IFACE_STATIC_IP + _RTS), iface_data)
|
|
self.assertEqual([], self.ifup_interface_names)
|
|
|
|
def test_dhcp_ovs_bridge_network_apply(self):
|
|
interface = objects.Interface('eth0')
|
|
bridge = objects.OvsBridge('br0', use_dhcp=True,
|
|
members=[interface])
|
|
self.provider.add_interface(interface)
|
|
self.provider.add_bridge(bridge)
|
|
self.provider.apply()
|
|
iface_data = utils.get_file_data(self.temp_config_file.name)
|
|
self.assertEqual((_OVS_BRIDGE_DHCP + _OVS_PORT_IFACE), iface_data)
|
|
self.assertIn('eth0', self.ifup_interface_names)
|
|
self.assertIn('br0', self.ifup_interface_names)
|
|
|
|
def _failed_execute(*args, **kwargs):
|
|
if kwargs.get('check_exit_code', True):
|
|
raise processutils.ProcessExecutionError('Test stderr',
|
|
'Test stdout',
|
|
str(kwargs))
|
|
|
|
def test_interface_failure(self):
|
|
self.stub_out('oslo_concurrency.processutils.execute',
|
|
self._failed_execute)
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = objects.Interface('em1', addresses=[v4_addr])
|
|
self.provider.add_interface(interface)
|
|
|
|
self.assertRaises(os_net_config.ConfigurationError,
|
|
self.provider.apply)
|
|
self.assertEqual(1, len(self.provider.errors))
|
|
|
|
def test_interface_failure_multiple(self):
|
|
self.stub_out('oslo_concurrency.processutils.execute',
|
|
self._failed_execute)
|
|
v4_addr = objects.Address('192.168.1.2/24')
|
|
interface = objects.Interface('em1', addresses=[v4_addr])
|
|
v4_addr2 = objects.Address('192.168.2.2/24')
|
|
interface2 = objects.Interface('em2', addresses=[v4_addr2])
|
|
self.provider.add_interface(interface)
|
|
self.provider.add_interface(interface2)
|
|
|
|
self.assertRaises(os_net_config.ConfigurationError,
|
|
self.provider.apply)
|
|
# Even though the first one failed, we should have attempted both
|
|
self.assertEqual(2, len(self.provider.errors))
|