Start using oslo_utils.netutils.is_ipv6_enabled()
Seems that is_enabled_and_bind_by_default() from neutron.common.ipv6_utils was copied directly into oslo_utils.netutils, so start using it instead. Trivialfix Change-Id: I00fa441e7a20fcd1115485bb8ab75750e6a8cf07
This commit is contained in:
parent
0546a4482a
commit
555238da69
|
@ -35,6 +35,7 @@ from oslo_serialization import jsonutils
|
||||||
from oslo_service import loopingcall
|
from oslo_service import loopingcall
|
||||||
from oslo_service import periodic_task
|
from oslo_service import periodic_task
|
||||||
from oslo_utils import excutils
|
from oslo_utils import excutils
|
||||||
|
from oslo_utils import netutils
|
||||||
from oslo_utils import timeutils
|
from oslo_utils import timeutils
|
||||||
from osprofiler import profiler
|
from osprofiler import profiler
|
||||||
|
|
||||||
|
@ -55,7 +56,6 @@ from neutron.agent.linux import pd
|
||||||
from neutron.agent.linux import utils as linux_utils
|
from neutron.agent.linux import utils as linux_utils
|
||||||
from neutron.agent.metadata import driver as metadata_driver
|
from neutron.agent.metadata import driver as metadata_driver
|
||||||
from neutron.agent import rpc as agent_rpc
|
from neutron.agent import rpc as agent_rpc
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.common import utils
|
from neutron.common import utils
|
||||||
from neutron import manager
|
from neutron import manager
|
||||||
|
|
||||||
|
@ -330,7 +330,7 @@ class L3NATAgent(ha.AgentMixin,
|
||||||
super(L3NATAgent, self).__init__(host=self.conf.host)
|
super(L3NATAgent, self).__init__(host=self.conf.host)
|
||||||
|
|
||||||
self.target_ex_net_id = None
|
self.target_ex_net_id = None
|
||||||
self.use_ipv6 = ipv6_utils.is_enabled_and_bind_by_default()
|
self.use_ipv6 = netutils.is_ipv6_enabled()
|
||||||
|
|
||||||
self.pd = pd.PrefixDelegation(self.context, self.process_monitor,
|
self.pd = pd.PrefixDelegation(self.context, self.process_monitor,
|
||||||
self.driver,
|
self.driver,
|
||||||
|
|
|
@ -28,6 +28,7 @@ from neutron_lib.utils import file as file_utils
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
from oslo_utils import excutils
|
from oslo_utils import excutils
|
||||||
from oslo_utils import fileutils
|
from oslo_utils import fileutils
|
||||||
|
from oslo_utils import netutils
|
||||||
from oslo_utils import uuidutils
|
from oslo_utils import uuidutils
|
||||||
import six
|
import six
|
||||||
|
|
||||||
|
@ -37,7 +38,6 @@ from neutron.agent.linux import external_process
|
||||||
from neutron.agent.linux import ip_lib
|
from neutron.agent.linux import ip_lib
|
||||||
from neutron.agent.linux import iptables_manager
|
from neutron.agent.linux import iptables_manager
|
||||||
from neutron.cmd import runtime_checks as checks
|
from neutron.cmd import runtime_checks as checks
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.common import utils as common_utils
|
from neutron.common import utils as common_utils
|
||||||
from neutron.ipam import utils as ipam_utils
|
from neutron.ipam import utils as ipam_utils
|
||||||
|
|
||||||
|
@ -1523,7 +1523,7 @@ class DeviceManager(object):
|
||||||
ip_lib.IPWrapper().ensure_namespace(network.namespace)
|
ip_lib.IPWrapper().ensure_namespace(network.namespace)
|
||||||
ip_lib.set_ip_nonlocal_bind_for_namespace(network.namespace, 1,
|
ip_lib.set_ip_nonlocal_bind_for_namespace(network.namespace, 1,
|
||||||
root_namespace=True)
|
root_namespace=True)
|
||||||
if ipv6_utils.is_enabled_and_bind_by_default():
|
if netutils.is_ipv6_enabled():
|
||||||
self.driver.configure_ipv6_ra(network.namespace, 'default',
|
self.driver.configure_ipv6_ra(network.namespace, 'default',
|
||||||
constants.ACCEPT_RA_DISABLED)
|
constants.ACCEPT_RA_DISABLED)
|
||||||
|
|
||||||
|
|
|
@ -23,6 +23,7 @@ from neutron_lib import constants
|
||||||
from neutron_lib import exceptions
|
from neutron_lib import exceptions
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
|
from oslo_utils import netutils
|
||||||
from pyroute2.netlink import exceptions as netlink_exceptions
|
from pyroute2.netlink import exceptions as netlink_exceptions
|
||||||
from pyroute2.netlink import rtnl
|
from pyroute2.netlink import rtnl
|
||||||
from pyroute2.netlink.rtnl import ifaddrmsg
|
from pyroute2.netlink.rtnl import ifaddrmsg
|
||||||
|
@ -32,7 +33,6 @@ from pyroute2 import netns
|
||||||
|
|
||||||
from neutron._i18n import _
|
from neutron._i18n import _
|
||||||
from neutron.agent.common import utils
|
from neutron.agent.common import utils
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.common import utils as common_utils
|
from neutron.common import utils as common_utils
|
||||||
from neutron.privileged.agent.linux import ip_lib as privileged
|
from neutron.privileged.agent.linux import ip_lib as privileged
|
||||||
|
|
||||||
|
@ -379,7 +379,7 @@ class IPDevice(SubProcessBase):
|
||||||
'port': dport})
|
'port': dport})
|
||||||
|
|
||||||
def disable_ipv6(self):
|
def disable_ipv6(self):
|
||||||
if not ipv6_utils.is_enabled_and_bind_by_default():
|
if not netutils.is_ipv6_enabled():
|
||||||
return
|
return
|
||||||
sysctl_name = re.sub(r'\.', '/', self.name)
|
sysctl_name = re.sub(r'\.', '/', self.name)
|
||||||
cmd = ['net.ipv6.conf.%s.disable_ipv6=1' % sysctl_name]
|
cmd = ['net.ipv6.conf.%s.disable_ipv6=1' % sysctl_name]
|
||||||
|
|
|
@ -31,7 +31,6 @@ from neutron.agent.linux import ipset_manager
|
||||||
from neutron.agent.linux import iptables_comments as ic
|
from neutron.agent.linux import iptables_comments as ic
|
||||||
from neutron.agent.linux import iptables_manager
|
from neutron.agent.linux import iptables_manager
|
||||||
from neutron.common import _constants as const
|
from neutron.common import _constants as const
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.common import utils as c_utils
|
from neutron.common import utils as c_utils
|
||||||
|
|
||||||
|
|
||||||
|
@ -65,7 +64,7 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
|
||||||
def __init__(self, namespace=None):
|
def __init__(self, namespace=None):
|
||||||
self.iptables = iptables_manager.IptablesManager(
|
self.iptables = iptables_manager.IptablesManager(
|
||||||
state_less=True,
|
state_less=True,
|
||||||
use_ipv6=ipv6_utils.is_enabled_and_bind_by_default(),
|
use_ipv6=netutils.is_ipv6_enabled(),
|
||||||
namespace=namespace)
|
namespace=namespace)
|
||||||
# TODO(majopela, shihanzhang): refactor out ipset to a separate
|
# TODO(majopela, shihanzhang): refactor out ipset to a separate
|
||||||
# driver composed over this one
|
# driver composed over this one
|
||||||
|
|
|
@ -16,37 +16,12 @@
|
||||||
"""
|
"""
|
||||||
IPv6-related utilities and helper functions.
|
IPv6-related utilities and helper functions.
|
||||||
"""
|
"""
|
||||||
import os
|
|
||||||
|
|
||||||
import netaddr
|
import netaddr
|
||||||
from neutron_lib import constants as const
|
from neutron_lib import constants as const
|
||||||
from oslo_log import log
|
from oslo_log import log
|
||||||
|
|
||||||
|
|
||||||
LOG = log.getLogger(__name__)
|
LOG = log.getLogger(__name__)
|
||||||
_IS_IPV6_ENABLED = None
|
|
||||||
|
|
||||||
|
|
||||||
def is_enabled_and_bind_by_default():
|
|
||||||
"""Check if host has the IPv6 support and is configured to bind IPv6
|
|
||||||
address to new interfaces by default.
|
|
||||||
"""
|
|
||||||
global _IS_IPV6_ENABLED
|
|
||||||
|
|
||||||
if _IS_IPV6_ENABLED is None:
|
|
||||||
disabled_ipv6_path = "/proc/sys/net/ipv6/conf/default/disable_ipv6"
|
|
||||||
if os.path.exists(disabled_ipv6_path):
|
|
||||||
with open(disabled_ipv6_path, 'r') as f:
|
|
||||||
disabled = f.read().strip()
|
|
||||||
_IS_IPV6_ENABLED = disabled == "0"
|
|
||||||
else:
|
|
||||||
_IS_IPV6_ENABLED = False
|
|
||||||
if not _IS_IPV6_ENABLED:
|
|
||||||
LOG.info("IPv6 not present or configured not to bind to new "
|
|
||||||
"interfaces on this system. Please ensure IPv6 is "
|
|
||||||
"enabled and /proc/sys/net/ipv6/conf/default/"
|
|
||||||
"disable_ipv6 is set to 0 to enable IPv6.")
|
|
||||||
return _IS_IPV6_ENABLED
|
|
||||||
|
|
||||||
|
|
||||||
def is_auto_address_subnet(subnet):
|
def is_auto_address_subnet(subnet):
|
||||||
|
|
|
@ -16,11 +16,11 @@ from neutron_lib import constants as const
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_log import helpers as log_helpers
|
from oslo_log import helpers as log_helpers
|
||||||
from oslo_log import log
|
from oslo_log import log
|
||||||
|
from oslo_utils import netutils
|
||||||
|
|
||||||
from neutron.agent.l2.extensions import qos_linux as qos
|
from neutron.agent.l2.extensions import qos_linux as qos
|
||||||
from neutron.agent.linux import iptables_manager
|
from neutron.agent.linux import iptables_manager
|
||||||
from neutron.agent.linux import tc_lib
|
from neutron.agent.linux import tc_lib
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.services.qos.drivers.linuxbridge import driver
|
from neutron.services.qos.drivers.linuxbridge import driver
|
||||||
|
|
||||||
LOG = log.getLogger(__name__)
|
LOG = log.getLogger(__name__)
|
||||||
|
@ -58,7 +58,7 @@ class QosLinuxbridgeAgentDriver(qos.QosLinuxAgentDriver):
|
||||||
# created here for extension needs
|
# created here for extension needs
|
||||||
self.iptables_manager = iptables_manager.IptablesManager(
|
self.iptables_manager = iptables_manager.IptablesManager(
|
||||||
state_less=True,
|
state_less=True,
|
||||||
use_ipv6=ipv6_utils.is_enabled_and_bind_by_default())
|
use_ipv6=netutils.is_ipv6_enabled())
|
||||||
self.iptables_manager.initialize_mangle_table()
|
self.iptables_manager.initialize_mangle_table()
|
||||||
|
|
||||||
def _dscp_chain_name(self, direction, device):
|
def _dscp_chain_name(self, direction, device):
|
||||||
|
|
|
@ -16,13 +16,13 @@ from neutron_lib import constants
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_log import helpers as log_helpers
|
from oslo_log import helpers as log_helpers
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
|
from oslo_utils import netutils
|
||||||
|
|
||||||
from neutron.agent.common import utils as common_utils
|
from neutron.agent.common import utils as common_utils
|
||||||
from neutron.agent.l3 import dvr_snat_ns
|
from neutron.agent.l3 import dvr_snat_ns
|
||||||
from neutron.agent.l3 import namespaces
|
from neutron.agent.l3 import namespaces
|
||||||
from neutron.agent.linux import ip_lib
|
from neutron.agent.linux import ip_lib
|
||||||
from neutron.agent.linux import iptables_manager
|
from neutron.agent.linux import iptables_manager
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.conf.agent import common as config
|
from neutron.conf.agent import common as config
|
||||||
from neutron.services.metering.drivers import abstract_driver
|
from neutron.services.metering.drivers import abstract_driver
|
||||||
|
|
||||||
|
@ -97,7 +97,7 @@ class RouterWithMetering(object):
|
||||||
namespace=snat_ns_name,
|
namespace=snat_ns_name,
|
||||||
binary_name=WRAP_NAME,
|
binary_name=WRAP_NAME,
|
||||||
state_less=True,
|
state_less=True,
|
||||||
use_ipv6=ipv6_utils.is_enabled_and_bind_by_default())
|
use_ipv6=netutils.is_ipv6_enabled())
|
||||||
|
|
||||||
created = True
|
created = True
|
||||||
|
|
||||||
|
@ -112,7 +112,7 @@ class RouterWithMetering(object):
|
||||||
namespace=self.ns_name,
|
namespace=self.ns_name,
|
||||||
binary_name=WRAP_NAME,
|
binary_name=WRAP_NAME,
|
||||||
state_less=True,
|
state_less=True,
|
||||||
use_ipv6=ipv6_utils.is_enabled_and_bind_by_default())
|
use_ipv6=netutils.is_ipv6_enabled())
|
||||||
|
|
||||||
created = True
|
created = True
|
||||||
|
|
||||||
|
|
|
@ -17,12 +17,12 @@ import copy
|
||||||
|
|
||||||
import mock
|
import mock
|
||||||
from neutron_lib import constants
|
from neutron_lib import constants
|
||||||
|
from oslo_utils import netutils
|
||||||
import testtools
|
import testtools
|
||||||
|
|
||||||
from neutron.agent.common import ovs_lib
|
from neutron.agent.common import ovs_lib
|
||||||
from neutron.agent.l3 import agent as neutron_l3_agent
|
from neutron.agent.l3 import agent as neutron_l3_agent
|
||||||
from neutron.agent.linux import ip_lib
|
from neutron.agent.linux import ip_lib
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.common import utils as common_utils
|
from neutron.common import utils as common_utils
|
||||||
from neutron.tests.common import l3_test_common
|
from neutron.tests.common import l3_test_common
|
||||||
from neutron.tests.common import net_helpers
|
from neutron.tests.common import net_helpers
|
||||||
|
@ -127,8 +127,7 @@ class L3HATestCase(framework.L3AgentTestFramework):
|
||||||
self._assert_ipv6_forwarding(router, expected_forwarding,
|
self._assert_ipv6_forwarding(router, expected_forwarding,
|
||||||
True)
|
True)
|
||||||
|
|
||||||
@testtools.skipUnless(ipv6_utils.is_enabled_and_bind_by_default(),
|
@testtools.skipUnless(netutils.is_ipv6_enabled(), "IPv6 is not enabled")
|
||||||
"IPv6 is not enabled")
|
|
||||||
def test_ipv6_router_advts_and_fwd_after_router_state_change_master(self):
|
def test_ipv6_router_advts_and_fwd_after_router_state_change_master(self):
|
||||||
# Check that RA and forwarding are enabled when there's no IPv6
|
# Check that RA and forwarding are enabled when there's no IPv6
|
||||||
# gateway.
|
# gateway.
|
||||||
|
@ -143,8 +142,7 @@ class L3HATestCase(framework.L3AgentTestFramework):
|
||||||
expected_ra=False,
|
expected_ra=False,
|
||||||
expected_forwarding=True)
|
expected_forwarding=True)
|
||||||
|
|
||||||
@testtools.skipUnless(ipv6_utils.is_enabled_and_bind_by_default(),
|
@testtools.skipUnless(netutils.is_ipv6_enabled(), "IPv6 is not enabled")
|
||||||
"IPv6 is not enabled")
|
|
||||||
def test_ipv6_router_advts_and_fwd_after_router_state_change_backup(self):
|
def test_ipv6_router_advts_and_fwd_after_router_state_change_backup(self):
|
||||||
# Check that both RA and forwarding are disabled on backup instances
|
# Check that both RA and forwarding are disabled on backup instances
|
||||||
self._test_ipv6_router_advts_and_fwd_helper('backup',
|
self._test_ipv6_router_advts_and_fwd_helper('backup',
|
||||||
|
@ -347,8 +345,7 @@ class L3HATestCase(framework.L3AgentTestFramework):
|
||||||
raise
|
raise
|
||||||
self.assertEqual(0, ip_nonlocal_bind_value)
|
self.assertEqual(0, ip_nonlocal_bind_value)
|
||||||
|
|
||||||
@testtools.skipUnless(ipv6_utils.is_enabled_and_bind_by_default(),
|
@testtools.skipUnless(netutils.is_ipv6_enabled(), "IPv6 is not enabled")
|
||||||
"IPv6 is not enabled")
|
|
||||||
def test_ha_router_namespace_has_ipv6_forwarding_disabled(self):
|
def test_ha_router_namespace_has_ipv6_forwarding_disabled(self):
|
||||||
router_info = self.generate_router_info(enable_ha=True)
|
router_info = self.generate_router_info(enable_ha=True)
|
||||||
router_info[constants.HA_INTERFACE_KEY]['status'] = (
|
router_info[constants.HA_INTERFACE_KEY]['status'] = (
|
||||||
|
@ -369,8 +366,7 @@ class L3HATestCase(framework.L3AgentTestFramework):
|
||||||
self._wait_until_ipv6_forwarding_has_state(
|
self._wait_until_ipv6_forwarding_has_state(
|
||||||
router.ns_name, external_device_name, 1)
|
router.ns_name, external_device_name, 1)
|
||||||
|
|
||||||
@testtools.skipUnless(ipv6_utils.is_enabled_and_bind_by_default(),
|
@testtools.skipUnless(netutils.is_ipv6_enabled(), "IPv6 is not enabled")
|
||||||
"IPv6 is not enabled")
|
|
||||||
def test_ha_router_without_gw_ipv6_forwarding_state(self):
|
def test_ha_router_without_gw_ipv6_forwarding_state(self):
|
||||||
router_info = self.generate_router_info(
|
router_info = self.generate_router_info(
|
||||||
enable_ha=True, enable_gw=False)
|
enable_ha=True, enable_gw=False)
|
||||||
|
|
|
@ -25,6 +25,7 @@ from neutron_lib import constants as const
|
||||||
from neutron_lib import exceptions
|
from neutron_lib import exceptions
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
import oslo_messaging
|
import oslo_messaging
|
||||||
|
from oslo_utils import netutils
|
||||||
from oslo_utils import timeutils
|
from oslo_utils import timeutils
|
||||||
import testtools
|
import testtools
|
||||||
|
|
||||||
|
@ -1743,8 +1744,8 @@ class TestDeviceManager(base.BaseTestCase):
|
||||||
"IPWrapper")
|
"IPWrapper")
|
||||||
self.mock_ip_wrapper = self.mock_ip_wrapper_p.start()
|
self.mock_ip_wrapper = self.mock_ip_wrapper_p.start()
|
||||||
|
|
||||||
self.mock_ipv6_enabled_p = mock.patch('neutron.common.ipv6_utils.'
|
self.mock_ipv6_enabled_p = mock.patch.object(netutils,
|
||||||
'is_enabled_and_bind_by_default')
|
'is_ipv6_enabled')
|
||||||
self.mock_ipv6_enabled = self.mock_ipv6_enabled_p.start()
|
self.mock_ipv6_enabled = self.mock_ipv6_enabled_p.start()
|
||||||
self.mock_ipv6_enabled.return_value = True
|
self.mock_ipv6_enabled.return_value = True
|
||||||
|
|
||||||
|
|
|
@ -28,6 +28,7 @@ from neutron_lib import constants as lib_constants
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
from oslo_log import log
|
from oslo_log import log
|
||||||
import oslo_messaging
|
import oslo_messaging
|
||||||
|
from oslo_utils import netutils
|
||||||
from oslo_utils import timeutils
|
from oslo_utils import timeutils
|
||||||
from oslo_utils import uuidutils
|
from oslo_utils import uuidutils
|
||||||
from testtools import matchers
|
from testtools import matchers
|
||||||
|
@ -329,9 +330,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
|
||||||
destroy_metadata_proxy.assert_not_called()
|
destroy_metadata_proxy.assert_not_called()
|
||||||
|
|
||||||
def _test__configure_ipv6_params_helper(self, state, gw_port_id):
|
def _test__configure_ipv6_params_helper(self, state, gw_port_id):
|
||||||
with mock.patch(
|
with mock.patch.object(netutils, 'is_ipv6_enabled', return_value=True):
|
||||||
'neutron.common.ipv6_utils.is_enabled_and_bind_by_default',
|
|
||||||
return_value=True):
|
|
||||||
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
|
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
|
||||||
|
|
||||||
router_info = l3router.RouterInfo(agent, _uuid(), {}, **self.ri_kwargs)
|
router_info = l3router.RouterInfo(agent, _uuid(), {}, **self.ri_kwargs)
|
||||||
|
|
|
@ -16,6 +16,7 @@
|
||||||
# under the License.
|
# under the License.
|
||||||
|
|
||||||
import mock
|
import mock
|
||||||
|
from oslo_utils import netutils
|
||||||
|
|
||||||
from neutron.agent.linux import bridge_lib
|
from neutron.agent.linux import bridge_lib
|
||||||
from neutron.privileged.agent.linux import ip_lib as priv_lib
|
from neutron.privileged.agent.linux import ip_lib as priv_lib
|
||||||
|
@ -30,9 +31,8 @@ class BridgeLibTest(base.BaseTestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super(BridgeLibTest, self).setUp()
|
super(BridgeLibTest, self).setUp()
|
||||||
mock.patch(
|
mock.patch.object(netutils, 'is_ipv6_enabled',
|
||||||
'neutron.common.ipv6_utils.is_enabled_and_bind_by_default',
|
return_value=True).start()
|
||||||
return_value=True).start()
|
|
||||||
ip_wrapper = mock.patch('neutron.agent.linux.ip_lib.IPWrapper').start()
|
ip_wrapper = mock.patch('neutron.agent.linux.ip_lib.IPWrapper').start()
|
||||||
self.execute = ip_wrapper.return_value.netns.execute
|
self.execute = ip_wrapper.return_value.netns.execute
|
||||||
self.create_p = mock.patch.object(priv_lib, 'create_interface')
|
self.create_p = mock.patch.object(priv_lib, 'create_interface')
|
||||||
|
|
|
@ -21,6 +21,7 @@ import mock
|
||||||
import netaddr
|
import netaddr
|
||||||
from neutron_lib import constants
|
from neutron_lib import constants
|
||||||
from neutron_lib import exceptions
|
from neutron_lib import exceptions
|
||||||
|
from oslo_utils import netutils
|
||||||
from oslo_utils import uuidutils
|
from oslo_utils import uuidutils
|
||||||
import pyroute2
|
import pyroute2
|
||||||
from pyroute2.netlink.rtnl import ifaddrmsg
|
from pyroute2.netlink.rtnl import ifaddrmsg
|
||||||
|
@ -1447,9 +1448,7 @@ class TestSysctl(base.BaseTestCase):
|
||||||
|
|
||||||
def test_disable_ipv6_when_ipv6_globally_enabled(self):
|
def test_disable_ipv6_when_ipv6_globally_enabled(self):
|
||||||
dev = ip_lib.IPDevice('tap0', 'ns1')
|
dev = ip_lib.IPDevice('tap0', 'ns1')
|
||||||
with mock.patch.object(ip_lib.ipv6_utils,
|
with mock.patch.object(netutils, 'is_ipv6_enabled', return_value=True):
|
||||||
'is_enabled_and_bind_by_default',
|
|
||||||
return_value=True):
|
|
||||||
dev.disable_ipv6()
|
dev.disable_ipv6()
|
||||||
self.execute.assert_called_once_with(
|
self.execute.assert_called_once_with(
|
||||||
['sysctl', '-w', 'net.ipv6.conf.tap0.disable_ipv6=1'],
|
['sysctl', '-w', 'net.ipv6.conf.tap0.disable_ipv6=1'],
|
||||||
|
@ -1457,8 +1456,7 @@ class TestSysctl(base.BaseTestCase):
|
||||||
|
|
||||||
def test_disable_ipv6_when_ipv6_globally_disabled(self):
|
def test_disable_ipv6_when_ipv6_globally_disabled(self):
|
||||||
dev = ip_lib.IPDevice('tap0', 'ns1')
|
dev = ip_lib.IPDevice('tap0', 'ns1')
|
||||||
with mock.patch.object(ip_lib.ipv6_utils,
|
with mock.patch.object(netutils, 'is_ipv6_enabled',
|
||||||
'is_enabled_and_bind_by_default',
|
|
||||||
return_value=False):
|
return_value=False):
|
||||||
dev.disable_ipv6()
|
dev.disable_ipv6()
|
||||||
self.assertFalse(self.execute.called)
|
self.assertFalse(self.execute.called)
|
||||||
|
|
|
@ -15,54 +15,12 @@
|
||||||
|
|
||||||
import collections
|
import collections
|
||||||
|
|
||||||
import mock
|
|
||||||
from neutron_lib import constants
|
from neutron_lib import constants
|
||||||
from neutron_lib import fixture as lib_fixtures
|
|
||||||
|
|
||||||
from neutron.common import ipv6_utils
|
from neutron.common import ipv6_utils
|
||||||
from neutron.tests import base
|
from neutron.tests import base
|
||||||
|
|
||||||
|
|
||||||
class TestIsEnabledAndBindByDefault(base.BaseTestCase):
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
super(TestIsEnabledAndBindByDefault, self).setUp()
|
|
||||||
|
|
||||||
def reset_detection_flag():
|
|
||||||
ipv6_utils._IS_IPV6_ENABLED = None
|
|
||||||
reset_detection_flag()
|
|
||||||
self.addCleanup(reset_detection_flag)
|
|
||||||
self.mock_exists = mock.patch("os.path.exists",
|
|
||||||
return_value=True).start()
|
|
||||||
self.proc_path = '/proc/sys/net/ipv6/conf/default/disable_ipv6'
|
|
||||||
|
|
||||||
def test_enabled(self):
|
|
||||||
self.useFixture(lib_fixtures.OpenFixture(self.proc_path, '0'))
|
|
||||||
enabled = ipv6_utils.is_enabled_and_bind_by_default()
|
|
||||||
self.assertTrue(enabled)
|
|
||||||
|
|
||||||
def test_disabled(self):
|
|
||||||
self.useFixture(lib_fixtures.OpenFixture(self.proc_path, '1'))
|
|
||||||
enabled = ipv6_utils.is_enabled_and_bind_by_default()
|
|
||||||
self.assertFalse(enabled)
|
|
||||||
|
|
||||||
def test_disabled_non_exists(self):
|
|
||||||
mo = self.useFixture(
|
|
||||||
lib_fixtures.OpenFixture(self.proc_path, '1')).mock_open
|
|
||||||
self.mock_exists.return_value = False
|
|
||||||
enabled = ipv6_utils.is_enabled_and_bind_by_default()
|
|
||||||
self.assertFalse(enabled)
|
|
||||||
self.assertFalse(mo.called)
|
|
||||||
|
|
||||||
def test_memoize(self):
|
|
||||||
mo = self.useFixture(
|
|
||||||
lib_fixtures.OpenFixture(self.proc_path, '0')).mock_open
|
|
||||||
ipv6_utils.is_enabled_and_bind_by_default()
|
|
||||||
enabled = ipv6_utils.is_enabled_and_bind_by_default()
|
|
||||||
self.assertTrue(enabled)
|
|
||||||
mo.assert_called_once_with(self.proc_path, 'r')
|
|
||||||
|
|
||||||
|
|
||||||
class TestIsAutoAddressSubnet(base.BaseTestCase):
|
class TestIsAutoAddressSubnet(base.BaseTestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
|
|
@ -21,12 +21,12 @@ import mock
|
||||||
from neutron_lib.db import api as db_api
|
from neutron_lib.db import api as db_api
|
||||||
from neutron_lib import exceptions as exception
|
from neutron_lib import exceptions as exception
|
||||||
from oslo_config import cfg
|
from oslo_config import cfg
|
||||||
|
from oslo_utils import netutils
|
||||||
from six.moves import urllib
|
from six.moves import urllib
|
||||||
import testtools
|
import testtools
|
||||||
import webob
|
import webob
|
||||||
import webob.exc
|
import webob.exc
|
||||||
|
|
||||||
from neutron.common import ipv6_utils
|
|
||||||
from neutron.tests import base
|
from neutron.tests import base
|
||||||
from neutron.tests.common import helpers
|
from neutron.tests.common import helpers
|
||||||
from neutron import wsgi
|
from neutron import wsgi
|
||||||
|
@ -111,7 +111,7 @@ class TestWSGIServer(base.BaseTestCase):
|
||||||
launcher.wait.assert_called_once_with()
|
launcher.wait.assert_called_once_with()
|
||||||
|
|
||||||
@testtools.skipIf(
|
@testtools.skipIf(
|
||||||
not ipv6_utils.is_enabled_and_bind_by_default(),
|
not netutils.is_ipv6_enabled(),
|
||||||
'IPv6 support disabled on host')
|
'IPv6 support disabled on host')
|
||||||
def test_start_random_port_with_ipv6(self):
|
def test_start_random_port_with_ipv6(self):
|
||||||
server = wsgi.Server("test_random_port")
|
server = wsgi.Server("test_random_port")
|
||||||
|
|
Loading…
Reference in New Issue