本文整理汇总了Python中neutron.common.ipv6_utils.is_enabled函数的典型用法代码示例。如果您正苦于以下问题：Python is_enabled函数的具体用法？Python is_enabled怎么用？Python is_enabled使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了is_enabled函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, namespace=None):
self.root_helper = cfg.CONF.AGENT.root_helper
self.iptables = iptables_manager.IptablesManager(
root_helper=self.root_helper,
use_ipv6=ipv6_utils.is_enabled())
# TODO(majopela, shihanzhang): refactor out ipset to a separate
# driver composed over this one
self.ipset = ipset_manager.IpsetManager(root_helper=self.root_helper)
# list of port which has security group
self.filtered_ports = {}
self._add_fallback_chain_v4v6()
self._defer_apply = False
self._pre_defer_filtered_ports = None
# List of security group rules for ports residing on this host
self.sg_rules = {}
self.pre_sg_rules = None
# List of security group member ips for ports residing on this host
self.sg_members = {}
self.pre_sg_members = None
self.ipset_chains = {}
self.enable_ipset = cfg.CONF.SECURITYGROUP.enable_ipset
self.defer_sg_rules = {}
self.defer_sgs = {}
self.available_local_zones = set(xrange(constants.MIN_VLAN_TAG,
constants.MAX_VLAN_TAG))
self.local_zone_map = {}
self.namespace = namespace
开发者ID:HybridCloud-dew，项目名称:hws，代码行数:27，代码来源:iptables_firewall.py
示例2: __init__
def __init__(self, namespace=None):
self.iptables = iptables_manager.IptablesManager(
use_ipv6=ipv6_utils.is_enabled(),
namespace=namespace)
# TODO(majopela, shihanzhang): refactor out ipset to a separate
# driver composed over this one
self.ipset = ipset_manager.IpsetManager(namespace=namespace)
self.ipconntrack = ip_conntrack.IpConntrackManager(
self.get_device_zone, namespace=namespace)
self._populate_initial_zone_map()
# list of port which has security group
self.filtered_ports = {}
self.unfiltered_ports = {}
self._add_fallback_chain_v4v6()
self._defer_apply = False
self._pre_defer_filtered_ports = None
self._pre_defer_unfiltered_ports = None
# List of security group rules for ports residing on this host
self.sg_rules = {}
self.pre_sg_rules = None
# List of security group member ips for ports residing on this host
self.sg_members = collections.defaultdict(
lambda: collections.defaultdict(list))
self.pre_sg_members = None
self.enable_ipset = cfg.CONF.SECURITYGROUP.enable_ipset
self._enabled_netfilter_for_bridges = False
self.updated_rule_sg_ids = set()
self.updated_sg_members = set()
self.devices_with_udpated_sg_members = collections.defaultdict(list)
开发者ID:promptworks，项目名称:neutron，代码行数:29，代码来源:iptables_firewall.py
示例3: _validate_network_device_mtu
def _validate_network_device_mtu(self):
if (ipv6_utils.is_enabled() and
self.conf.network_device_mtu < n_const.IPV6_MIN_MTU):
LOG.error(_LE("IPv6 protocol requires a minimum MTU of "
"%(min_mtu)s, while the configured value is "
"%(current_mtu)s"), {'min_mtu': n_const.IPV6_MIN_MTU,
'current_mtu': self.conf.network_device_mtu})
raise SystemExit(1)
开发者ID:danielmellado，项目名称:neutron，代码行数:8，代码来源:interface.py
示例4: __init__
def __init__(self):
self.iptables = iptables_manager.IptablesManager(
root_helper=cfg.CONF.AGENT.root_helper,
use_ipv6=ipv6_utils.is_enabled())
# list of port which has security group
self.filtered_ports = {}
self._add_fallback_chain_v4v6()
self._defer_apply = False
self._pre_defer_filtered_ports = None
开发者ID:CingHu，项目名称:neutron-ustack，代码行数:9，代码来源:iptables_firewall.py
示例5: __init__
def __init__(self, conf, router):
self.conf = conf
self.id = router["id"]
self.router = router
# TODO(cbrandily): deduplicate ns_name generation in metering/l3
self.ns_name = NS_PREFIX + self.id
self.iptables_manager = iptables_manager.IptablesManager(
namespace=self.ns_name, binary_name=WRAP_NAME, state_less=True, use_ipv6=ipv6_utils.is_enabled()
)
self.metering_labels = {}
开发者ID:klmitch，项目名称:neutron，代码行数:10，代码来源:iptables_driver.py
示例6: __init__
def __init__(self, conf, router):
self.conf = conf
self.id = router['id']
self.router = router
self.ns_name = NS_PREFIX + self.id if conf.use_namespaces else None
self.iptables_manager = iptables_manager.IptablesManager(
namespace=self.ns_name,
binary_name=WRAP_NAME,
use_ipv6=ipv6_utils.is_enabled())
self.metering_labels = {}
开发者ID:Akanksha08，项目名称:neutron，代码行数:10，代码来源:iptables_driver.py
示例7: __init__
def __init__(self, conf, router):
self.conf = conf
self.id = router["id"]
self.router = router
self.root_helper = config.get_root_helper(self.conf)
self.ns_name = NS_PREFIX + self.id if conf.use_namespaces else None
self.iptables_manager = iptables_manager.IptablesManager(
root_helper=self.root_helper,
namespace=self.ns_name,
binary_name=WRAP_NAME,
use_ipv6=ipv6_utils.is_enabled(),
)
self.metering_labels = {}
开发者ID:CiscoSystems，项目名称:neutron，代码行数:13，代码来源:iptables_driver.py
示例8: __init__
def __init__(self):
self.iptables = iptables_manager.IptablesManager(
root_helper=cfg.CONF.AGENT.root_helper,
use_ipv6=ipv6_utils.is_enabled())
# list of port which has security group
self.filtered_ports = {}
self._add_fallback_chain_v4v6()
self._defer_apply = False
self._pre_defer_filtered_ports = None
# List of security group rules for ports residing on this host
self.sg_rules = {}
self.pre_sg_rules = None
# List of security group member ips for ports residing on this host
self.sg_members = {}
self.pre_sg_members = None
开发者ID:bart20073，项目名称:neutron，代码行数:15，代码来源:iptables_firewall.py
示例9: __init__
def __init__(self):
self.iptables = iptables_manager.IptablesManager(
use_ipv6=ipv6_utils.is_enabled())
# TODO(majopela, shihanzhang): refactor out ipset to a separate
# driver composed over this one
self.ipset = ipset_manager.IpsetManager()
# list of port which has security group
self.filtered_ports = {}
self._add_fallback_chain_v4v6()
self._defer_apply = False
self._pre_defer_filtered_ports = None
# List of security group rules for ports residing on this host
self.sg_rules = {}
self.pre_sg_rules = None
# List of security group member ips for ports residing on this host
self.sg_members = {}
self.pre_sg_members = None
self.enable_ipset = cfg.CONF.SECURITYGROUP.enable_ipset
开发者ID:bradleyjones，项目名称:neutron，代码行数:18，代码来源:iptables_firewall.py
示例10: __init__
def __init__(self, host, conf=None):
if conf:
self.conf = conf
else:
self.conf = cfg.CONF
self.root_helper = config.get_root_helper(self.conf)
self.router_info = {}
self._check_config_params()
try:
self.driver = importutils.import_object(
self.conf.interface_driver,
self.conf
)
except Exception:
msg = _("Error importing interface driver "
"'%s'") % self.conf.interface_driver
LOG.error(msg)
raise SystemExit(msg)
self.context = context.get_admin_context_without_session()
self.plugin_rpc = L3PluginApi(topics.L3PLUGIN, host)
self.fullsync = True
self.updated_routers = set()
self.removed_routers = set()
self.sync_progress = False
self._delete_stale_namespaces = (self.conf.use_namespaces and
self.conf.router_delete_namespaces)
self.rpc_loop = loopingcall.FixedIntervalLoopingCall(
self._rpc_loop)
self.rpc_loop.start(interval=RPC_LOOP_INTERVAL)
super(L3NATAgent, self).__init__(conf=self.conf)
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
开发者ID:codybum，项目名称:OpenStackInAction，代码行数:38，代码来源:l3_agent.py
示例11: test_enabled
def test_enabled(self):
self.mock_read.return_value = "0"
enabled = ipv6_utils.is_enabled()
self.assertTrue(enabled)
开发者ID:cisco-openstack，项目名称:neutron，代码行数:4，代码来源:test_ipv6_utils.py
示例12: __init__
def __init__(self, host, conf=None):
if conf:
self.conf = conf
else:
self.conf = cfg.CONF
self.router_info = {}
self._check_config_params()
self.process_monitor = external_process.ProcessMonitor(
config=self.conf,
resource_type='router')
self.conf.interface_driver = const.INF_DR
self.driver = common_utils.load_interface_driver(self.conf)
self.context = n_context.get_admin_context_without_session()
self.plugin_rpc = l3_agent.L3PluginApi(topics.L3PLUGIN, host)
self.ftnt_rpc = FortinetL3PluginApi(const.FTNT_AGENT, host)
self.task_manager = tasks.TaskManager()
self.task_manager.start()
self.fortigates = []
self.fortigate = fortigate.Fortigate()
self.api_client = self.fortigate.get_apiclient()
self.initialize_fortigate()
self.router = l3_fortinet.FortinetL3ServicePlugin(
fortigate=self.fortigate)
self.fullsync = True
self.test = True
# Get the list of service plugins from Neutron Server
# This is the first place where we contact neutron-server on startup
# so retry in case its not ready to respond.
retry_count = 5
while True:
retry_count = retry_count - 1
try:
self.neutron_service_plugins = (
self.plugin_rpc.get_service_plugin_list(self.context))
except oslo_messaging.RemoteError as e:
with excutils.save_and_reraise_exception() as ctx:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot check service plugins '
'enabled at the neutron server when '
'startup due to RPC error. It happens '
'when the server does not support this '
'RPC API. If the error is '
'UnsupportedVersion you can ignore this '
'warning. Detail message: %s'), e)
self.neutron_service_plugins = None
except oslo_messaging.MessagingTimeout as e:
with excutils.save_and_reraise_exception() as ctx:
if retry_count > 0:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot check service '
'plugins enabled on the neutron '
'server. Retrying. '
'Detail message: %s'), e)
continue
break
self.metadata_driver = None
if self.conf.enable_metadata_proxy:
self.metadata_driver = metadata_driver.MetadataDriver(self)
self.namespaces_manager = namespace_manager.NamespaceManager(
self.conf,
self.driver,
self.fortigate,
self.conf.use_namespaces,
self.metadata_driver)
self._queue = queue.RouterProcessingQueue()
super(FortinetAgent, self).__init__(conf=self.conf)
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
if self.conf.agent_mode == l3_constants.L3_AGENT_MODE_DVR:
self.pd = None
else:
self.pd = pd.PrefixDelegation(
self.context, self.process_monitor, self.driver,
self.plugin_rpc.process_prefix_update,
self.create_pd_router_update, self.conf)
开发者ID:samsu，项目名称:networking-fortinet，代码行数:82，代码来源:fortinet_agent.py
示例13: __init__
def __init__(self, host, conf=None):
if conf:
self.conf = conf
else:
self.conf = cfg.CONF
self.router_info = {}
self._check_config_params()
self.process_monitor = external_process.ProcessMonitor(
config=self.conf,
resource_type='router')
self.driver = common_utils.load_interface_driver(self.conf)
self._context = n_context.get_admin_context_without_session()
self.plugin_rpc = L3PluginApi(topics.L3PLUGIN, host)
self.fullsync = True
self.sync_routers_chunk_size = SYNC_ROUTERS_MAX_CHUNK_SIZE
# Get the list of service plugins from Neutron Server
# This is the first place where we contact neutron-server on startup
# so retry in case its not ready to respond.
while True:
try:
self.neutron_service_plugins = (
self.plugin_rpc.get_service_plugin_list(self.context))
except oslo_messaging.RemoteError as e:
with excutils.save_and_reraise_exception() as ctx:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot check service plugins '
'enabled at the neutron server when '
'startup due to RPC error. It happens '
'when the server does not support this '
'RPC API. If the error is '
'UnsupportedVersion you can ignore this '
'warning. Detail message: %s'), e)
self.neutron_service_plugins = None
except oslo_messaging.MessagingTimeout as e:
with excutils.save_and_reraise_exception() as ctx:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot contact neutron server '
'to retrieve service plugins enabled. '
'Check connectivity to neutron server. '
'Retrying... '
'Detailed message: %(msg)s.'), {'msg': e})
continue
break
self.init_extension_manager(self.plugin_rpc)
self.metadata_driver = None
if self.conf.enable_metadata_proxy:
self.metadata_driver = metadata_driver.MetadataDriver(self)
self.namespaces_manager = namespace_manager.NamespaceManager(
self.conf,
self.driver,
self.metadata_driver)
self._queue = queue.RouterProcessingQueue()
super(L3NATAgent, self).__init__(host=self.conf.host)
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
self.pd = pd.PrefixDelegation(self.context, self.process_monitor,
self.driver,
self.plugin_rpc.process_prefix_update,
self.create_pd_router_update,
self.conf)
开发者ID:brandonlogan，项目名称:neutron，代码行数:71，代码来源:agent.py
示例14: __init__
def __init__(self, host, conf=None):
if conf:
self.conf = conf
else:
self.conf = cfg.CONF
self.router_info = {}
self._check_config_params()
self.process_monitor = external_process.ProcessMonitor(
config=self.conf,
resource_type='router')
try:
self.driver = importutils.import_object(
self.conf.interface_driver,
self.conf
)
except Exception:
LOG.error(_LE("Error importing interface driver "
"'%s'"), self.conf.interface_driver)
raise SystemExit(1)
self.context = n_context.get_admin_context_without_session()
self.plugin_rpc = L3PluginApi(topics.L3PLUGIN, host)
self.fullsync = True
# Get the list of service plugins from Neutron Server
# This is the first place where we contact neutron-server on startup
# so retry in case its not ready to respond.
retry_count = 5
while True:
retry_count = retry_count - 1
try:
self.neutron_service_plugins = (
self.plugin_rpc.get_service_plugin_list(self.context))
except oslo_messaging.RemoteError as e:
with excutils.save_and_reraise_exception() as ctx:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot check service plugins '
'enabled at the neutron server when '
'startup due to RPC error. It happens '
'when the server does not support this '
'RPC API. If the error is '
'UnsupportedVersion you can ignore this '
'warning. Detail message: %s'), e)
self.neutron_service_plugins = None
except oslo_messaging.MessagingTimeout as e:
with excutils.save_and_reraise_exception() as ctx:
if retry_count > 0:
ctx.reraise = False
LOG.warning(_LW('l3-agent cannot check service '
'plugins enabled on the neutron '
'server. Retrying. '
'Detail message: %s'), e)
continue
break
self.metadata_driver = None
if self.conf.enable_metadata_proxy:
self.metadata_driver = metadata_driver.MetadataDriver(self)
self.namespaces_manager = namespace_manager.NamespaceManager(
self.conf,
self.driver,
self.conf.use_namespaces,
self.metadata_driver)
self._queue = queue.RouterProcessingQueue()
super(L3NATAgent, self).__init__(conf=self.conf)
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
开发者ID:kkxue，项目名称:neutron，代码行数:73，代码来源:agent.py
示例15: test_enabled
def test_enabled(self):
self.useFixture(tools.OpenFixture(self.proc_path, '0'))
enabled = ipv6_utils.is_enabled()
self.assertTrue(enabled)
开发者ID:21atlas，项目名称:neutron，代码行数:4，代码来源:test_ipv6_utils.py
示例16: test_memoize
def test_memoize(self):
self.mock_read.return_value = "0"
ipv6_utils.is_enabled()
enabled = ipv6_utils.is_enabled()
self.assertTrue(enabled)
self.mock_read.assert_called_once_with()
开发者ID:cisco-openstack，项目名称:neutron，代码行数:6，代码来源:test_ipv6_utils.py
示例17: test_disabled
def test_disabled(self):
self.useFixture(tools.OpenFixture(self.proc_path, '1'))
enabled = ipv6_utils.is_enabled()
self.assertFalse(enabled)
开发者ID:21atlas，项目名称:neutron，代码行数:4，代码来源:test_ipv6_utils.py
示例18: test_disabled_non_exists
def test_disabled_non_exists(self):
self.mock_exists.return_value = False
enabled = ipv6_utils.is_enabled()
self.assertFalse(enabled)
self.assertFalse(self.mock_read.called)
开发者ID:cisco-openstack，项目名称:neutron，代码行数:5，代码来源:test_ipv6_utils.py
示例19: test_disabled
def test_disabled(self):
self.mock_read.return_value = "1"
enabled = ipv6_utils.is_enabled()
self.assertFalse(enabled)
开发者ID:cisco-openstack，项目名称:neutron，代码行数:4，代码来源:test_ipv6_utils.py
示例20: __init__
def __init__(self, host, conf=None):
if conf:
self.conf = conf
else:
self.conf = cfg.CONF
self.router_info = {}
self._check_config_params()
self.process_monitor = external_process.ProcessMonitor(config=self.conf, resource_type="router")
self.driver = common_utils.load_interface_driver(self.conf)
self.context = n_context.get_admin_context_without_session()
self.plugin_rpc = L3PluginApi(topics.L3PLUGIN, host)
self.fullsync = True
# Get the list of service plugins from Neutron Server
# This is the first place where we contact neutron-server on startup
# so retry in case its not ready to respond.
retry_count = 5
while True:
retry_count = retry_count - 1
try:
self.neutron_service_plugins = self.plugin_rpc.get_service_plugin_list(self.context)
except oslo_messaging.RemoteError as e:
with excutils.save_and_reraise_exception() as ctx:
ctx.reraise = False
LOG.warning(
_LW(
"l3-agent cannot check service plugins "
"enabled at the neutron server when "
"startup due to RPC error. It happens "
"when the server does not support this "
"RPC API. If the error is "
"UnsupportedVersion you can ignore this "
"warning. Detail message: %s"
),
e,
)
self.neutron_service_plugins = None
except oslo_messaging.MessagingTimeout as e:
with excutils.save_and_reraise_exception() as ctx:
if retry_count > 0:
ctx.reraise = False
LOG.warning(
_LW(
"l3-agent cannot check service "
"plugins enabled on the neutron "
"server. Retrying. "
"Detail message: %s"
),
e,
)
continue
break
self.metadata_driver = None
if self.conf.enable_metadata_proxy:
self.metadata_driver = metadata_driver.MetadataDriver(self)
self.namespaces_manager = namespace_manager.NamespaceManager(
self.conf, self.driver, self.conf.use_namespaces, self.metadata_driver
)
self._queue = queue.RouterProcessingQueue()
super(L3NATAgent, self).__init__(conf=self.conf)
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
self.pd = pd.PrefixDelegation(
self.context,
self.process_monitor,
self.driver,
self.plugin_rpc.process_prefix_update,
self.create_pd_router_update,
self.conf,
)
开发者ID:akash1808，项目名称:neutron，代码行数:79，代码来源:agent.py
注：本文中的neutron.common.ipv6_utils.is_enabled函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论