本文整理汇总了Python中tricircle.db.api.get_bottom_mappings_by_top_id函数的典型用法代码示例。如果您正苦于以下问题：Python get_bottom_mappings_by_top_id函数的具体用法？Python get_bottom_mappings_by_top_id怎么用？Python get_bottom_mappings_by_top_id使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_bottom_mappings_by_top_id函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_delete_port_chain
def test_delete_port_chain(self):
project_id = TEST_TENANT_ID
q_ctx = FakeNeutronContext()
t_ctx = context.get_db_context()
self._basic_pod_setup()
fake_plugin = FakeSfcPlugin()
ids = {'t_ppg_id': [uuidutils.generate_uuid()],
'b_ppg_id': [uuidutils.generate_uuid()],
't_fc_id': [uuidutils.generate_uuid()],
'b_fc_id': [uuidutils.generate_uuid()]}
t_pc_id1, _ = self._prepare_port_chain_test(
project_id, t_ctx, 'pod_1', 0, True, ids)
fake_plugin.delete_port_chain(q_ctx, t_pc_id1)
pc_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_pc_id1, constants.RT_PORT_CHAIN)
self.assertEqual(len(TOP_PORTCHAINS), 0)
self.assertEqual(len(BOTTOM1_PORTCHAINS), 0)
self.assertEqual(len(pc_mappings), 0)
t_pc_id2, _ = self._prepare_port_chain_test(
project_id, t_ctx, 'pod_1', 0, True, ids)
BOTTOM1_PORTCHAINS.pop()
fake_plugin.delete_port_chain(q_ctx, t_pc_id2)
pc_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_pc_id2, constants.RT_PORT_CHAIN)
self.assertEqual(len(TOP_PORTCHAINS), 0)
self.assertEqual(len(pc_mappings), 0)
开发者ID:openstack，项目名称:tricircle，代码行数:28，代码来源:test_central_sfc_plugin.py
示例2: test_delete_flow_classifier
def test_delete_flow_classifier(self):
project_id = TEST_TENANT_ID
q_ctx = FakeNeutronContext()
t_ctx = context.get_db_context()
self._basic_pod_setup()
fake_plugin = FakeFcPlugin()
src_port_id = uuidutils.generate_uuid()
t_fc_id1, _ = self._prepare_flow_classifier_test(
project_id, t_ctx, 'pod_1', 0, src_port_id, True)
fake_plugin.delete_flow_classifier(q_ctx, t_fc_id1)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_fc_id1, constants.RT_FLOW_CLASSIFIER)
self.assertEqual(len(TOP_FLOWCLASSIFIERS), 0)
self.assertEqual(len(BOTTOM1_FLOWCLASSIFIERS), 0)
self.assertEqual(len(ppg_mappings), 0)
t_fc_id2, _ = self._prepare_flow_classifier_test(
project_id, t_ctx, 'pod_1', 0, src_port_id, True)
BOTTOM1_FLOWCLASSIFIERS.pop()
fake_plugin.delete_flow_classifier(q_ctx, t_fc_id2)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_fc_id2, constants.RT_FLOW_CLASSIFIER)
self.assertEqual(len(TOP_FLOWCLASSIFIERS), 0)
self.assertEqual(len(ppg_mappings), 0)
开发者ID:openstack，项目名称:tricircle，代码行数:26，代码来源:test_central_sfc_plugin.py
示例3: test_delete_port_pair
def test_delete_port_pair(self):
project_id = TEST_TENANT_ID
q_ctx = FakeNeutronContext()
t_ctx = context.get_db_context()
self._basic_pod_setup()
fake_plugin = FakeSfcPlugin()
ingress = uuidutils.generate_uuid()
egress = uuidutils.generate_uuid()
t_pp1_id, _ = self._prepare_port_pair_test(
project_id, t_ctx, 'pod_1', 0, ingress, egress, True)
fake_plugin.delete_port_pair(q_ctx, t_pp1_id)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_pp1_id, constants.RT_PORT_PAIR_GROUP)
self.assertEqual(len(TOP_PORTPAIRS), 0)
self.assertEqual(len(BOTTOM1_PORTPAIRS), 0)
self.assertEqual(len(ppg_mappings), 0)
t_pp2_id, _ = self._prepare_port_pair_test(
project_id, t_ctx, 'pod_1', 0, ingress, egress, True)
BOTTOM1_PORTPAIRS.pop()
fake_plugin.delete_port_pair(q_ctx, t_pp2_id)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_pp2_id, constants.RT_PORT_PAIR_GROUP)
self.assertEqual(len(TOP_PORTPAIRS), 0)
self.assertEqual(len(ppg_mappings), 0)
开发者ID:openstack，项目名称:tricircle，代码行数:26，代码来源:test_central_sfc_plugin.py
示例4: test_delete_port_pair_group
def test_delete_port_pair_group(self):
project_id = TEST_TENANT_ID
q_ctx = FakeNeutronContext()
t_ctx = context.get_db_context()
self._basic_pod_setup()
fake_plugin = FakeSfcPlugin()
t_pp_id = uuidutils.generate_uuid()
b_pp_id = uuidutils.generate_uuid()
t_ppg_id1, _ = self._prepare_port_pair_group_test(
project_id, t_ctx, 'pod_1', 0, [t_pp_id], True, [b_pp_id])
fake_plugin.delete_port_pair_group(q_ctx, t_ppg_id1)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_ppg_id1, constants.RT_PORT_PAIR_GROUP)
self.assertEqual(len(TOP_PORTPAIRGROUPS), 0)
self.assertEqual(len(BOTTOM1_PORTPAIRGROUPS), 0)
self.assertEqual(len(ppg_mappings), 0)
t_ppg_id2, _ = self._prepare_port_pair_group_test(
project_id, t_ctx, 'pod_1', 0, [t_pp_id], True, [b_pp_id])
BOTTOM1_PORTPAIRGROUPS.pop()
fake_plugin.delete_port_pair_group(q_ctx, t_ppg_id2)
ppg_mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, t_ppg_id2, constants.RT_PORT_PAIR_GROUP)
self.assertEqual(len(TOP_PORTPAIRGROUPS), 0)
self.assertEqual(len(ppg_mappings), 0)
开发者ID:openstack，项目名称:tricircle，代码行数:27，代码来源:test_central_sfc_plugin.py
示例5: post
def post(self, **kw):
context = t_context.extract_context_from_environ()
if 'volumeAttachment' not in kw:
pecan.abort(400, 'Request body not found')
return
body = kw['volumeAttachment']
if 'volumeId' not in body:
pecan.abort(400, 'Volume not set')
return
server_mappings = db_api.get_bottom_mappings_by_top_id(
context, self.server_id, constants.RT_SERVER)
if not server_mappings:
pecan.abort(404, 'Server not found')
return
volume_mappings = db_api.get_bottom_mappings_by_top_id(
context, body['volumeId'], constants.RT_VOLUME)
if not volume_mappings:
pecan.abort(404, 'Volume not found')
return
server_pod_name = server_mappings[0][0]['pod_name']
volume_pod_name = volume_mappings[0][0]['pod_name']
if server_pod_name != volume_pod_name:
LOG.error(_LE('Server %(server)s is in pod %(server_pod)s and '
'volume %(volume)s is in pod %(volume_pod)s, which '
'are not the same.'),
{'server': self.server_id,
'server_pod': server_pod_name,
'volume': body['volumeId'],
'volume_pod': volume_pod_name})
pecan.abort(400, 'Server and volume not in the same pod')
return
device = None
if 'device' in body:
device = body['device']
# this regular expression is copied from nova/block_device.py
match = re.match('(^/dev/x{0,1}[a-z]{0,1}d{0,1})([a-z]+)[0-9]*$',
device)
if not match:
pecan.abort(400, 'Invalid device path')
return
client = self._get_client(server_pod_name)
volume = client.action_server_volumes(
context, 'create_server_volume',
server_mappings[0][1], volume_mappings[0][1], device)
return {'volumeAttachment': volume.to_dict()}
开发者ID:Ronghui，项目名称:tricircle，代码行数:50，代码来源:volume.py
示例6: get_real_shadow_resource_iterator
def get_real_shadow_resource_iterator(t_ctx, res_type, res_id):
shadow_res_type = t_constants.REAL_SHADOW_TYPE_MAP[res_type]
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, res_id, res_type)
mappings.extend(db_api.get_bottom_mappings_by_top_id(
t_ctx, res_id, shadow_res_type))
processed_pod_set = set()
for pod, bottom_res_id in mappings:
region_name = pod['region_name']
if region_name in processed_pod_set:
continue
processed_pod_set.add(region_name)
yield pod, bottom_res_id
开发者ID:LongXQ，项目名称:tricircle，代码行数:14，代码来源:helper.py
示例7: update_port_pair_group_precommit
def update_port_pair_group_precommit(self, context):
plugin_context = context._plugin_context
t_ctx = t_context.get_context_from_neutron_context(
context._plugin_context)
port_pair_group = context.current
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, port_pair_group['id'], t_constants.RT_PORT_PAIR_GROUP)
if mappings:
portchain_id = self._get_chain_id_by_group_id(
plugin_context, context._plugin, port_pair_group['id'])
if port_pair_group['port_pairs']:
net_id = self._get_net_id_by_portpairgroups(
plugin_context, context._plugin, [port_pair_group['id']])
elif context.original['port_pairs']:
portpair_id = context.original['port_pairs'][0]
port_pair = context._plugin._get_port_pair(
plugin_context, portpair_id)
net_id = self._get_net_id_by_port_id(
plugin_context, port_pair['ingress'])
else:
net_id = ''
if not portchain_id and not net_id:
return
self.xjob_handler.sync_service_function_chain(
t_ctx, port_pair_group['project_id'], portchain_id, net_id,
t_constants.POD_NOT_SPECIFIED)
开发者ID:openstack，项目名称:tricircle，代码行数:26，代码来源:central_sfc_driver.py
示例8: test_get_bottom_mappings_by_top_id
def test_get_bottom_mappings_by_top_id(self):
for i in xrange(3):
pod = {'pod_id': 'test_pod_uuid_%d' % i,
'pod_name': 'test_pod_%d' % i,
'az_name': 'test_az_uuid_%d' % i}
api.create_pod(self.context, pod)
route1 = {
'top_id': 'top_uuid',
'pod_id': 'test_pod_uuid_0',
'resource_type': 'port'}
route2 = {
'top_id': 'top_uuid',
'pod_id': 'test_pod_uuid_1',
'bottom_id': 'bottom_uuid_1',
'resource_type': 'port'}
route3 = {
'top_id': 'top_uuid',
'pod_id': 'test_pod_uuid_2',
'bottom_id': 'bottom_uuid_2',
'resource_type': 'neutron'}
routes = [route1, route2, route3]
with self.context.session.begin():
for route in routes:
core.create_resource(
self.context, models.ResourceRouting, route)
mappings = api.get_bottom_mappings_by_top_id(self.context,
'top_uuid', 'port')
self.assertEqual('test_pod_uuid_1', mappings[0][0]['pod_id'])
self.assertEqual('bottom_uuid_1', mappings[0][1])
开发者ID:Ronghui，项目名称:tricircle，代码行数:29，代码来源:test_api.py
示例9: create_security_group_rule
def create_security_group_rule(self, q_context, security_group_rule):
rule = security_group_rule['security_group_rule']
if rule['remote_group_id']:
raise n_exceptions.RemoteGroupNotSupported()
sg_id = rule['security_group_id']
sg = self.get_security_group(q_context, sg_id)
if sg['name'] == 'default':
raise n_exceptions.DefaultGroupUpdateNotSupported()
new_rule = super(TricircleSecurityGroupMixin,
self).create_security_group_rule(q_context,
security_group_rule)
t_context = context.get_context_from_neutron_context(q_context)
mappings = db_api.get_bottom_mappings_by_top_id(
t_context, sg_id, constants.RT_SG)
try:
for pod, b_sg_id in mappings:
client = self._get_client(pod['pod_name'])
rule['security_group_id'] = b_sg_id
self._safe_create_security_group_rule(
t_context, client, {'security_group_rule': rule})
except Exception:
super(TricircleSecurityGroupMixin,
self).delete_security_group_rule(q_context, new_rule['id'])
raise n_exceptions.BottomPodOperationFailure(
resource='security group rule', pod_name=pod['pod_name'])
return new_rule
开发者ID:Ronghui，项目名称:tricircle，代码行数:29，代码来源:security_groups.py
示例10: delete_security_group_rule
def delete_security_group_rule(self, q_context, _id):
rule = self.get_security_group_rule(q_context, _id)
if rule['remote_group_id']:
raise n_exceptions.RemoteGroupNotSupported()
sg_id = rule['security_group_id']
sg = self.get_security_group(q_context, sg_id)
if sg['name'] == 'default':
raise n_exceptions.DefaultGroupUpdateNotSupported()
t_context = context.get_context_from_neutron_context(q_context)
mappings = db_api.get_bottom_mappings_by_top_id(
t_context, sg_id, constants.RT_SG)
try:
for pod, b_sg_id in mappings:
client = self._get_client(pod['pod_name'])
rule['security_group_id'] = b_sg_id
b_sg = client.get_security_groups(t_context, b_sg_id)
for b_rule in b_sg['security_group_rules']:
if not self._compare_rule(b_rule, rule):
continue
self._safe_delete_security_group_rule(t_context, client,
b_rule['id'])
break
except Exception:
raise n_exceptions.BottomPodOperationFailure(
resource='security group rule', pod_name=pod['pod_name'])
super(TricircleSecurityGroupMixin,
self).delete_security_group_rule(q_context, _id)
开发者ID:Ronghui，项目名称:tricircle，代码行数:30，代码来源:security_groups.py
示例11: test_prepare_neutron_element
def test_prepare_neutron_element(self):
t_pod, b_pod = self._prepare_pod()
port = {'id': 'top_port_id'}
body = {'port': {'name': 'top_port_id'}}
_, bottom_port_id = self.controller._prepare_neutron_element(
self.context, b_pod, port, 'port', body)
mappings = api.get_bottom_mappings_by_top_id(self.context,
'top_port_id', 'port')
self.assertEqual(bottom_port_id, mappings[0][1])
开发者ID:Ronghui，项目名称:tricircle，代码行数:9，代码来源:test_server.py
示例12: get_trunks
def get_trunks(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None, page_reverse=False):
ret = []
bottom_top_map = {}
top_bottom_map = {}
t_ctx = t_context.get_context_from_neutron_context(context)
route_filters = [{'key': 'resource_type',
'comparator': 'eq',
'value': t_constants.RT_TRUNK}]
routes = db_api.list_resource_routings(t_ctx, route_filters)
for route in routes:
bottom_top_map[route['bottom_id']] = route['top_id']
top_bottom_map[route['top_id']] = route['bottom_id']
if limit:
if marker:
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, marker, t_constants.RT_TRUNK)
# if mapping exists, we retrieve trunk information
# from bottom, otherwise from top
if mappings:
pod_id = mappings[0][0]['pod_id']
current_pod = db_api.get_pod(t_ctx, pod_id)
ret = self._get_trunks_from_pod_with_limit(
context, current_pod, bottom_top_map, top_bottom_map,
filters, limit, marker)
else:
ret = self._get_trunks_from_top_with_limit(
context, top_bottom_map, filters, limit, marker)
else:
current_pod = db_api.get_next_bottom_pod(t_ctx)
# if current_pod exists, we retrieve trunk information
# from bottom, otherwise from top
if current_pod:
ret = self._get_trunks_from_pod_with_limit(
context, current_pod, bottom_top_map, top_bottom_map,
filters, limit, None)
else:
ret = self._get_trunks_from_top_with_limit(
context, top_bottom_map, filters, limit, None)
else:
pods = db_api.list_pods(t_ctx)
_filters = self._transform_trunk_filters(filters, top_bottom_map)
for pod in pods:
if not pod['az_name']:
continue
client = self._get_client(pod['region_name'])
pod_trunks = client.list_trunks(t_ctx, filters=_filters)
ret.extend(pod_trunks)
ret = self._map_trunks_from_bottom_to_top(ret, bottom_top_map)
top_trunks = self._get_trunks_from_top(context,
top_bottom_map, filters)
ret.extend(top_trunks)
return [super(TricircleTrunkPlugin, self)._fields(trunk, fields)
for trunk in ret]
开发者ID:southeast02，项目名称:tricircle，代码行数:57，代码来源:central_trunk_plugin.py
示例13: test_get_bottom_mappings_by_top_id
def test_get_bottom_mappings_by_top_id(self):
self._create_pod(0, 'test_az_uuid_0')
self._create_pod(1, 'test_az_uuid_1')
self._create_pod(2, 'test_az_uuid_2')
self._create_resource_mappings()
mappings = api.get_bottom_mappings_by_top_id(self.context,
'top_uuid', 'network')
self.assertEqual('test_pod_uuid_1', mappings[0][0]['pod_id'])
self.assertEqual('top_uuid', mappings[0][1])
开发者ID:LongXQ，项目名称:tricircle，代码行数:9，代码来源:test_api.py
示例14: test_delete_mappings_by_top_id
def test_delete_mappings_by_top_id(self):
self._create_pod(0, 'test_az_uuid_0')
self._create_pod(1, 'test_az_uuid_1')
self._create_pod(2, 'test_az_uuid_2')
self._create_resource_mappings()
top_id = 'top_uuid'
api.delete_mappings_by_top_id(self.context, top_id,
pod_id='test_pod_uuid_0')
mappings = api.get_bottom_mappings_by_top_id(
self.context, top_id, 'network')
# entry in pod_uuid_0 is deleted, entry in pod_uuid_1 is left
self.assertEqual(1, len(mappings))
self.assertEqual('test_pod_uuid_1', mappings[0][0]['pod_id'])
api.delete_mappings_by_top_id(self.context, top_id)
mappings = api.get_bottom_mappings_by_top_id(
self.context, top_id, 'network')
self.assertEqual(0, len(mappings))
开发者ID:LongXQ，项目名称:tricircle，代码行数:19，代码来源:test_api.py
示例15: _get_pod_by_top_id
def _get_pod_by_top_id(self, context, _id):
mappings = db_api.get_bottom_mappings_by_top_id(
context, _id,
cons.RT_VOLUME)
if not mappings or len(mappings) != 1:
return None
return mappings[0][0]
开发者ID:Ronghui，项目名称:tricircle，代码行数:10，代码来源:volume.py
示例16: test_prepare_neutron_element_create_res_exception
def test_prepare_neutron_element_create_res_exception(self, mock_method):
mock_method.side_effect = FakeException()
t_pod, b_pod = self._prepare_pod()
port = {'id': 'top_port_id'}
body = {'port': {'name': 'top_port_id'}}
self.assertRaises(FakeException,
self.controller._prepare_neutron_element,
self.context, b_pod, port, 'port', body)
mappings = api.get_bottom_mappings_by_top_id(self.context,
'top_port_id', 'port')
self.assertEqual(0, len(mappings))
开发者ID:Ronghui，项目名称:tricircle，代码行数:11，代码来源:test_server.py
示例17: delete_trunk
def delete_trunk(self, context, trunk_id):
t_ctx = t_context.get_context_from_neutron_context(context)
res = super(TricircleTrunkPlugin, self).get_trunk(context, trunk_id)
with context.session.begin():
super(TricircleTrunkPlugin, self).delete_trunk(context, trunk_id)
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, trunk_id, t_constants.RT_TRUNK)
if mappings:
b_pod = mappings[0][0]
self.xjob_handler.sync_trunk(t_ctx, res['project_id'],
trunk_id, b_pod['pod_id'])
开发者ID:southeast02，项目名称:tricircle，代码行数:11，代码来源:central_trunk_plugin.py
示例18: update_trunk
def update_trunk(self, context, trunk_id, trunk):
# update trunk
t_ctx = t_context.get_context_from_neutron_context(context)
with context.session.begin():
res = super(TricircleTrunkDriver, self).update_trunk(
context, trunk_id, trunk)
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, trunk_id, t_constants.RT_TRUNK)
if mappings:
b_pod = mappings[0][0]
self.xjob_handler.sync_trunk(t_ctx, res['project_id'],
trunk_id, b_pod['pod_id'])
return res
开发者ID:openstack，项目名称:tricircle，代码行数:13，代码来源:central_trunk_driver.py
示例19: remove_subports
def remove_subports(self, context, trunk_id, subports):
t_ctx = t_context.get_context_from_neutron_context(context)
with context.session.begin():
self.update_subports_device_id(context, subports, '', '')
res = super(TricircleTrunkDriver, self).remove_subports(
context, trunk_id, subports)
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, trunk_id, t_constants.RT_TRUNK)
if mappings:
b_pod = mappings[0][0]
self.xjob_handler.sync_trunk(
t_ctx, res['project_id'], trunk_id, b_pod['pod_id'])
return res
开发者ID:openstack，项目名称:tricircle，代码行数:14，代码来源:central_trunk_driver.py
示例20: add_subports
def add_subports(self, context, trunk_id, subports):
t_ctx = t_context.get_context_from_neutron_context(context)
with context.session.begin():
res = super(TricircleTrunkPlugin, self).add_subports(
context, trunk_id, subports)
self.update_subports_device_id(context, subports, trunk_id,
t_constants.DEVICE_OWNER_SUBPORT)
mappings = db_api.get_bottom_mappings_by_top_id(
t_ctx, trunk_id, t_constants.RT_TRUNK)
if mappings:
b_pod = mappings[0][0]
self.xjob_handler.sync_trunk(
t_ctx, res['project_id'], trunk_id, b_pod['pod_id'])
return res
开发者ID:southeast02，项目名称:tricircle，代码行数:15，代码来源:central_trunk_plugin.py
注：本文中的tricircle.db.api.get_bottom_mappings_by_top_id函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论