本文整理汇总了Python中quantum.plugins.nec.db.api.get_portinfo函数的典型用法代码示例。如果您正苦于以下问题：Python get_portinfo函数的具体用法？Python get_portinfo怎么用？Python get_portinfo使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_portinfo函数的11个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _test_portinfo_change
def _test_portinfo_change(self, portinfo_change_first=True):
with self.port() as port:
port_id = port['port']['id']
self.assertEqual(self.ofc.create_ofc_port.call_count, 0)
portinfo = {'id': port_id, 'port_no': 123}
self._rpcapi_update_ports(added=[portinfo])
self.assertEqual(self.ofc.create_ofc_port.call_count, 1)
self.assertEqual(self.ofc.delete_ofc_port.call_count, 0)
self.assertEqual(ndb.get_portinfo(self.context.session,
port_id).port_no, 123)
if portinfo_change_first:
portinfo = {'id': port_id, 'port_no': 456}
self._rpcapi_update_ports(added=[portinfo])
# OFC port is recreated.
self.assertEqual(self.ofc.create_ofc_port.call_count, 2)
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
self.assertEqual(ndb.get_portinfo(self.context.session,
port_id).port_no, 456)
if not portinfo_change_first:
# The port is expected to delete when exiting with-clause.
self.assertEqual(self.ofc.create_ofc_port.call_count, 1)
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
portinfo = {'id': port_id, 'port_no': 456}
self._rpcapi_update_ports(added=[portinfo])
# No OFC operations are expected.
self.assertEqual(self.ofc.create_ofc_port.call_count, 1)
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
self.assertEqual(ndb.get_portinfo(self.context.session,
port_id).port_no, 456)
开发者ID:DestinyOneSystems，项目名称:quantum，代码行数:33，代码来源:test_nec_plugin.py
示例2: update_ports
def update_ports(self, rpc_context, **kwargs):
"""Update ports' information and activate/deavtivate them.
Expected input format is:
{'topic': 'q-agent-notifier',
'agent_id': 'nec-q-agent.' + <hostname>,
'datapath_id': <datapath_id of br-int on remote host>,
'port_added': [<new PortInfo>,...],
'port_removed': [<removed Port ID>,...]}
"""
LOG.debug(_("NECPluginV2RPCCallbacks.update_ports() called, "
"kwargs=%s ."), kwargs)
topic = kwargs['topic']
datapath_id = kwargs['datapath_id']
session = rpc_context.session
for p in kwargs.get('port_added', []):
id = p['id']
port = self.plugin.get_port(rpc_context, id)
if port and ndb.get_portinfo(session, id):
ndb.del_portinfo(session, id)
self.plugin.deactivate_port(rpc_context, port)
ndb.add_portinfo(session, id, datapath_id, p['port_no'],
mac=p.get('mac', ''))
self.plugin.activate_port_if_ready(rpc_context, port)
for id in kwargs.get('port_removed', []):
port = self.plugin.get_port(rpc_context, id)
if port and ndb.get_portinfo(session, id):
ndb.del_portinfo(session, id)
self.plugin.deactivate_port(rpc_context, port)
开发者ID:AmirolAhmad，项目名称:quantum，代码行数:29，代码来源:nec_plugin.py
示例3: testf_del_portinfo
def testf_del_portinfo(self):
"""test delete portinfo"""
i, d, p, v, m, n = self.get_portinfo_random_params()
ndb.add_portinfo(self.session, i, d, p, v, m)
portinfo = ndb.get_portinfo(self.session, i)
self.assertEqual(portinfo.id, i)
ndb.del_portinfo(self.session, i)
portinfo_none = ndb.get_portinfo(self.session, i)
self.assertEqual(None, portinfo_none)
开发者ID:AmirolAhmad，项目名称:quantum，代码行数:9，代码来源:test_db.py
示例4: teste_get_portinfo
def teste_get_portinfo(self):
"""test get portinfo"""
i, d, p, v, m, n = self.get_portinfo_random_params()
ndb.add_portinfo(self.session, i, d, p, v, m)
portinfo = ndb.get_portinfo(self.session, i)
self.assertEqual(portinfo.id, i)
self.assertEqual(portinfo.datapath_id, d)
self.assertEqual(portinfo.port_no, p)
self.assertEqual(portinfo.vlan_id, v)
self.assertEqual(portinfo.mac, m)
portinfo_none = ndb.get_portinfo(self.session, n)
self.assertEqual(None, portinfo_none)
开发者ID:AmirolAhmad，项目名称:quantum，代码行数:13，代码来源:test_db.py
示例5: update_ports
def update_ports(self, rpc_context, **kwargs):
"""Update ports' information and activate/deavtivate them.
Expected input format is:
{'topic': 'q-agent-notifier',
'agent_id': 'nec-q-agent.' + <hostname>,
'datapath_id': <datapath_id of br-int on remote host>,
'port_added': [<new PortInfo>,...],
'port_removed': [<removed Port ID>,...]}
"""
LOG.debug(_("NECPluginV2RPCCallbacks.update_ports() called, " "kwargs=%s ."), kwargs)
datapath_id = kwargs["datapath_id"]
session = rpc_context.session
for p in kwargs.get("port_added", []):
id = p["id"]
portinfo = ndb.get_portinfo(session, id)
if portinfo:
ndb.del_portinfo(session, id)
ndb.add_portinfo(session, id, datapath_id, p["port_no"], mac=p.get("mac", ""))
port = self._get_port(rpc_context, id)
if port:
if portinfo:
self.plugin.deactivate_port(rpc_context, port)
self.plugin.activate_port_if_ready(rpc_context, port)
for id in kwargs.get("port_removed", []):
portinfo = ndb.get_portinfo(session, id)
if not portinfo:
LOG.debug(
_(
"update_ports(): ignore port_removed message "
"due to portinfo for port_id=%s was not "
"registered"
),
id,
)
continue
if portinfo.datapath_id != datapath_id:
LOG.debug(
_(
"update_ports(): ignore port_removed message "
"received from different host "
"(registered_datapath_id=%(registered)s, "
"received_datapath_id=%(received)s)."
),
{"registered": portinfo.datapath_id, "received": datapath_id},
)
continue
ndb.del_portinfo(session, id)
port = self._get_port(rpc_context, id)
if port:
self.plugin.deactivate_port(rpc_context, port)
开发者ID:Apsu，项目名称:quantum，代码行数:51，代码来源:nec_plugin.py
示例6: test_portinfo_added_unknown_port
def test_portinfo_added_unknown_port(self):
portinfo = {'id': 'dummy-p1', 'port_no': 123}
self._rpcapi_update_ports(added=[portinfo])
self.assertIsNotNone(ndb.get_portinfo(self.context.session,
'dummy-p1'))
self.assertEqual(self.ofc.exists_ofc_port.call_count, 0)
self.assertEqual(self.ofc.create_ofc_port.call_count, 0)
开发者ID:DestinyOneSystems，项目名称:quantum，代码行数:7，代码来源:test_nec_plugin.py
示例7: activate_port_if_ready
def activate_port_if_ready(self, context, port, network=None):
"""Activate port by creating port on OFC if ready.
Activate port and packet_filters associated with the port.
Conditions to activate port on OFC are:
* port admin_state is UP
* network admin_state is UP
* portinfo are available (to identify port on OFC)
"""
if not network:
network = super(NECPluginV2, self).get_network(context,
port['network_id'])
port_status = OperationalStatus.ACTIVE
if not port['admin_state_up']:
LOG.debug("activate_port_if_ready(): skip, "
"port.admin_state_up is False.")
port_status = OperationalStatus.DOWN
elif not network['admin_state_up']:
LOG.debug("activate_port_if_ready(): skip, "
"network.admin_state_up is False.")
port_status = OperationalStatus.DOWN
elif not ndb.get_portinfo(port['id']):
LOG.debug("activate_port_if_ready(): skip, "
"no portinfo for this port.")
port_status = OperationalStatus.DOWN
# activate packet_filters before creating port on OFC.
if self.packet_filter_enabled:
if port_status is OperationalStatus.ACTIVE:
filters = dict(in_port=[port['id']],
status=[OperationalStatus.DOWN],
admin_state_up=[True])
pfs = (super(NECPluginV2, self).
get_packet_filters(context, filters=filters))
for pf in pfs:
self._activate_packet_filter_if_ready(context, pf,
network=network,
in_port=port)
if port_status in [OperationalStatus.ACTIVE]:
if self.ofc.exists_ofc_port(port['id']):
LOG.debug("activate_port_if_ready(): skip, "
"ofc_port already exists.")
else:
try:
self.ofc.create_ofc_port(port['tenant_id'],
port['network_id'],
port['id'])
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = "create_ofc_port() failed due to %s" % exc
LOG.error(reason)
port_status = OperationalStatus.ERROR
if port_status is not port['status']:
self._update_resource_status(context, "port", port['id'],
port_status)
开发者ID:missall，项目名称:quantum，代码行数:57，代码来源:nec_plugin.py
示例8: _activate_packet_filter_if_ready
def _activate_packet_filter_if_ready(self, context, packet_filter,
network=None, in_port=None):
"""Activate packet_filter by creating filter on OFC if ready.
Conditions to create packet_filter on OFC are:
* packet_filter admin_state is UP
* network admin_state is UP
* (if 'in_port' is specified) portinfo is available
"""
net_id = packet_filter['network_id']
if not network:
network = super(NECPluginV2, self).get_network(context, net_id)
in_port_id = packet_filter.get("in_port")
if in_port_id and not in_port:
in_port = super(NECPluginV2, self).get_port(context, in_port_id)
pf_status = OperationalStatus.ACTIVE
if not packet_filter['admin_state_up']:
LOG.debug("_activate_packet_filter_if_ready(): skip, "
"packet_filter.admin_state_up is False.")
pf_status = OperationalStatus.DOWN
elif not network['admin_state_up']:
LOG.debug("_activate_packet_filter_if_ready(): skip, "
"network.admin_state_up is False.")
pf_status = OperationalStatus.DOWN
elif in_port_id and in_port_id is in_port.get('id'):
LOG.debug("_activate_packet_filter_if_ready(): skip, "
"invalid in_port_id.")
pf_status = OperationalStatus.DOWN
elif in_port_id and not ndb.get_portinfo(in_port_id):
LOG.debug("_activate_packet_filter_if_ready(): skip, "
"no portinfo for in_port.")
pf_status = OperationalStatus.DOWN
if pf_status in [OperationalStatus.ACTIVE]:
if self.ofc.exists_ofc_packet_filter(packet_filter['id']):
LOG.debug("_activate_packet_filter_if_ready(): skip, "
"ofc_packet_filter already exists.")
else:
try:
(self.ofc.
create_ofc_packet_filter(packet_filter['tenant_id'],
packet_filter['network_id'],
packet_filter['id'],
packet_filter))
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = ("create_ofc_packet_filter() failed due to "
"%s" % exc)
LOG.error(reason)
pf_status = OperationalStatus.ERROR
if pf_status is not packet_filter['status']:
self._update_resource_status(context, "packet_filter",
packet_filter['id'], pf_status)
开发者ID:missall，项目名称:quantum，代码行数:54，代码来源:nec_plugin.py
示例9: create_ofc_port
def create_ofc_port(self, context, port_id, port):
ofc_net_id = self._get_ofc_id(context, "ofc_network",
port['network_id'])
ofc_net_id = self.driver.convert_ofc_network_id(
context, ofc_net_id, port['tenant_id'])
portinfo = ndb.get_portinfo(context.session, port_id)
if not portinfo:
raise nexc.PortInfoNotFound(id=port_id)
ofc_port_id = self.driver.create_port(ofc_net_id, portinfo, port_id)
self._add_ofc_item(context, "ofc_port", port_id, ofc_port_id)
开发者ID:Apsu，项目名称:quantum，代码行数:11，代码来源:ofc_manager.py
示例10: create_ofc_packet_filter
def create_ofc_packet_filter(self, context, filter_id, filter_dict):
ofc_net_id = self._get_ofc_id(context, "ofc_network",
filter_dict['network_id'])
ofc_net_id = self.driver.convert_ofc_network_id(
context, ofc_net_id, filter_dict['tenant_id'])
in_port_id = filter_dict.get('in_port')
portinfo = None
if in_port_id:
portinfo = ndb.get_portinfo(context.session, in_port_id)
if not portinfo:
raise nexc.PortInfoNotFound(id=in_port_id)
ofc_pf_id = self.driver.create_filter(ofc_net_id,
filter_dict, portinfo, filter_id)
self._add_ofc_item(context, "ofc_packet_filter", filter_id, ofc_pf_id)
开发者ID:Apsu，项目名称:quantum，代码行数:15，代码来源:ofc_manager.py
示例11: _get_portinfo
def _get_portinfo(self, port_id):
return ndb.get_portinfo(self.context.session, port_id)
开发者ID:DestinyOneSystems，项目名称:quantum，代码行数:2，代码来源:test_nec_plugin.py
注：本文中的quantum.plugins.nec.db.api.get_portinfo函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论