本文整理汇总了Python中quantum.openstack.common.uuidutils.generate_uuid函数的典型用法代码示例。如果您正苦于以下问题:Python generate_uuid函数的具体用法?Python generate_uuid怎么用?Python generate_uuid使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了generate_uuid函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_create_for_sg_rule
def test_create_for_sg_rule(self):
sg_id = uuidutils.generate_uuid()
sg_name = 'test-sg'
in_chain_id = uuidutils.generate_uuid()
out_chain_id = uuidutils.generate_uuid()
self.mock_api_cfg.chains_in = [
_create_test_sg_in_chain(sg_id, sg_name, in_chain_id,
self._tenant_id),
_create_test_sg_out_chain(sg_id, sg_name, out_chain_id,
self._tenant_id)]
sg_rule_id = uuidutils.generate_uuid()
sg_rule = _create_test_sg_rule(self._tenant_id, sg_id, sg_rule_id)
props = {"os_sg_rule_id": sg_rule_id}
calls = [mock.call.add_rule().port_group(None).type(
'accept').nw_proto(6).nw_src_address(
'192.168.1.0').nw_src_length(24).tp_src_start(
None).tp_src_end(None).tp_dst_start(1).tp_dst_end(
65535).properties(props).create()]
self.client.create_for_sg_rule(sg_rule)
# Egress chain rule added
self.mock_api_cfg.chains_out[0].assert_has_calls(calls)
开发者ID:Apsu,项目名称:quantum,代码行数:25,代码来源:test_midonet_lib.py
示例2: create_security_group
def create_security_group(self, context, security_group, default_sg=False):
"""Create security group.
If default_sg is true that means we are a default security group for
a given tenant if it does not exist.
"""
s = security_group['security_group']
tenant_id = self._get_tenant_id_for_create(context, s)
if not default_sg:
self._ensure_default_security_group(context, tenant_id)
with context.session.begin(subtransactions=True):
security_group_db = SecurityGroup(id=s.get('id') or (
uuidutils.generate_uuid()),
description=s['description'],
tenant_id=tenant_id,
name=s['name'])
context.session.add(security_group_db)
if s.get('name') == 'default':
for ethertype in ext_sg.sg_supported_ethertypes:
# Allow intercommunication
db = SecurityGroupRule(
id=uuidutils.generate_uuid(), tenant_id=tenant_id,
security_group=security_group_db,
direction='ingress',
ethertype=ethertype,
source_group=security_group_db)
context.session.add(db)
return self._make_security_group_dict(security_group_db)
开发者ID:Frostman,项目名称:quantum,代码行数:30,代码来源:securitygroups_db.py
示例3: get_random_params
def get_random_params(self):
"""create random parameters for portinfo test"""
tenant = uuidutils.generate_uuid()
network = uuidutils.generate_uuid()
port = uuidutils.generate_uuid()
_filter = uuidutils.generate_uuid()
none = uuidutils.generate_uuid()
return tenant, network, port, _filter, none
开发者ID:StackOps,项目名称:quantum,代码行数:8,代码来源:test_ofc_manager.py
示例4: get_ofc_item_random_params
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test"""
tenant_id = uuidutils.generate_uuid()
network_id = uuidutils.generate_uuid()
port_id = uuidutils.generate_uuid()
portinfo = nmodels.PortInfo(
id=port_id, datapath_id="0x123456789", port_no=1234, vlan_id=321, mac="11:22:33:44:55:66"
)
return tenant_id, network_id, portinfo
开发者ID:rosstaylor,项目名称:quantum,代码行数:9,代码来源:test_trema_driver.py
示例5: get_portinfo_random_params
def get_portinfo_random_params(self):
"""create random parameters for portinfo test"""
port_id = uuidutils.generate_uuid()
datapath_id = hex(random.randint(0, 0xffffffff))
port_no = random.randint(1, 100)
vlan_id = random.randint(0, 4095)
mac = ':'.join(["%02x" % random.randint(0, 0xff) for x in range(6)])
none = uuidutils.generate_uuid()
return port_id, datapath_id, port_no, vlan_id, mac, none
开发者ID:AmirolAhmad,项目名称:quantum,代码行数:9,代码来源:test_db.py
示例6: get_ofc_item_random_params
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test."""
tenant_id = uuidutils.generate_uuid()
network_id = uuidutils.generate_uuid()
port_id = uuidutils.generate_uuid()
mac = ':'.join(['%x' % random.randint(0, 255) for i in xrange(6)])
portinfo = nmodels.PortInfo(id=port_id, datapath_id="0x123456789",
port_no=1234, vlan_id=321,
mac=mac)
return tenant_id, network_id, portinfo
开发者ID:Apsu,项目名称:quantum,代码行数:10,代码来源:test_trema_driver.py
示例7: test_get_port_groups_for_sg
def test_get_port_groups_for_sg(self):
sg_id = uuidutils.generate_uuid()
pg_id = uuidutils.generate_uuid()
self.mock_api_cfg.port_groups_in = [
_create_test_port_group(sg_id, 'test-sg', pg_id, self._tenant_id)]
pg = self.client.get_port_groups_for_sg(self._tenant_id, sg_id)
self.assertIsNotNone(pg)
self.assertEqual(pg.get_id(), pg_id)
开发者ID:Apsu,项目名称:quantum,代码行数:10,代码来源:test_midonet_lib.py
示例8: test_get_ready_devices_multiple_vips_and_pools
def test_get_ready_devices_multiple_vips_and_pools(self):
ctx = context.get_admin_context()
# add 3 pools and 2 vips directly to DB
# to create 2 "ready" devices and one pool without vip
pools = []
for i in xrange(0, 3):
pools.append(
ldb.Pool(
id=uuidutils.generate_uuid(),
subnet_id=self._subnet_id,
protocol="HTTP",
lb_method="ROUND_ROBIN",
status=constants.ACTIVE,
admin_state_up=True,
)
)
ctx.session.add(pools[i])
vip0 = ldb.Vip(
id=uuidutils.generate_uuid(),
protocol_port=80,
protocol="HTTP",
pool_id=pools[0].id,
status=constants.ACTIVE,
admin_state_up=True,
connection_limit=3,
)
ctx.session.add(vip0)
pools[0].vip_id = vip0.id
vip1 = ldb.Vip(
id=uuidutils.generate_uuid(),
protocol_port=80,
protocol="HTTP",
pool_id=pools[1].id,
status=constants.ACTIVE,
admin_state_up=True,
connection_limit=3,
)
ctx.session.add(vip1)
pools[1].vip_id = vip1.id
ctx.session.flush()
self.assertEqual(ctx.session.query(ldb.Pool).count(), 3)
self.assertEqual(ctx.session.query(ldb.Vip).count(), 2)
ready = self.callbacks.get_ready_devices(ctx)
self.assertEqual(len(ready), 2)
self.assertIn(pools[0].id, ready)
self.assertIn(pools[1].id, ready)
self.assertNotIn(pools[2].id, ready)
# cleanup
ctx.session.query(ldb.Pool).delete()
ctx.session.query(ldb.Vip).delete()
开发者ID:NCU-PDCLAB,项目名称:quantum,代码行数:55,代码来源:test_plugin.py
示例9: test_delete_quantum_ports
def test_delete_quantum_ports(self):
port1 = ovs_lib.VifPort('tap1234', 1, uuidutils.generate_uuid(),
'ca:fe:de:ad:be:ef', 'br')
port2 = ovs_lib.VifPort('tap5678', 2, uuidutils.generate_uuid(),
'ca:ee:de:ad:be:ef', 'br')
self.mox.StubOutWithMock(self.br, 'get_vif_ports')
self.br.get_vif_ports().AndReturn([port1, port2])
self.mox.StubOutWithMock(self.br, 'delete_port')
self.br.delete_port('tap1234')
self.br.delete_port('tap5678')
self.mox.ReplayAll()
self.br.delete_ports(all_ports=False)
self.mox.VerifyAll()
开发者ID:DestinyOneSystems,项目名称:quantum,代码行数:13,代码来源:test_ovs_lib.py
示例10: test_collect_quantum_ports
def test_collect_quantum_ports(self):
port1 = ovs_lib.VifPort('tap1234', 1, uuidutils.generate_uuid(),
'11:22:33:44:55:66', 'br')
port2 = ovs_lib.VifPort('tap5678', 2, uuidutils.generate_uuid(),
'77:88:99:aa:bb:cc', 'br')
port3 = ovs_lib.VifPort('tap90ab', 3, uuidutils.generate_uuid(),
'99:00:aa:bb:cc:dd', 'br')
ports = [[port1, port2], [port3]]
portnames = [p.port_name for p in itertools.chain(*ports)]
with mock.patch('quantum.agent.linux.ovs_lib.OVSBridge') as ovs:
ovs.return_value.get_vif_ports.side_effect = ports
bridges = ['br-int', 'br-ex']
ret = util.collect_quantum_ports(bridges, 'dummy_sudo')
self.assertEqual(ret, portnames)
开发者ID:StackOps,项目名称:quantum,代码行数:14,代码来源:test_agent_ovs_cleanup.py
示例11: create_security_group
def create_security_group(self, context, security_group, default_sg=False):
"""Create security group.
If default_sg is true that means we are a default security group for
a given tenant if it does not exist.
"""
s = security_group['security_group']
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
if (cfg.CONF.SECURITYGROUP.proxy_mode and not s.get('external_id')):
raise ext_sg.SecurityGroupProxyMode()
if not cfg.CONF.SECURITYGROUP.proxy_mode and s.get('external_id'):
raise ext_sg.SecurityGroupNotProxyMode()
tenant_id = self._get_tenant_id_for_create(context, s)
# if in proxy mode a default security group will be created by source
if not default_sg and not cfg.CONF.SECURITYGROUP.proxy_mode:
self._ensure_default_security_group(context, tenant_id,
security_group)
if s.get('external_id'):
try:
# Check if security group already exists
sg = self.get_security_group(context, s.get('external_id'))
if sg:
raise ext_sg.SecurityGroupAlreadyExists(
name=sg.get('name', ''),
external_id=s.get('external_id'))
except ext_sg.SecurityGroupNotFound:
pass
with context.session.begin(subtransactions=True):
security_group_db = SecurityGroup(id=s.get('id') or (
uuidutils.generate_uuid()),
description=s['description'],
tenant_id=tenant_id,
name=s['name'],
external_id=s.get('external_id'))
context.session.add(security_group_db)
if s.get('name') == 'default':
for ethertype in ext_sg.sg_supported_ethertypes:
# Allow intercommunication
db = SecurityGroupRule(
id=uuidutils.generate_uuid(), tenant_id=tenant_id,
security_group=security_group_db,
direction='ingress',
ethertype=ethertype,
source_group=security_group_db)
context.session.add(db)
return self._make_security_group_dict(security_group_db)
开发者ID:aristanetworks,项目名称:arista-ovs-quantum,代码行数:50,代码来源:securitygroups_db.py
示例12: create_member
def create_member(self, context, member):
v = member["member"]
tenant_id = self._get_tenant_id_for_create(context, v)
with context.session.begin(subtransactions=True):
pool = None
try:
qry = context.session.query(Pool)
pool = qry.filter_by(id=v["pool_id"]).one()
except exc.NoResultFound:
raise loadbalancer.PoolNotFound(pool_id=v["pool_id"])
member_db = Member(
id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
pool_id=v["pool_id"],
address=v["address"],
protocol_port=v["protocol_port"],
weight=v["weight"],
admin_state_up=v["admin_state_up"],
status=constants.PENDING_CREATE,
)
context.session.add(member_db)
return self._make_member_dict(member_db)
开发者ID:vglafirov,项目名称:quantum,代码行数:25,代码来源:loadbalancer_db.py
示例13: create_port
def create_port(self, ofc_network_id, portinfo, port_id=None):
#NOTE: This Driver create slices with Port-MAC Based bindings on Trema
# Sliceable. It's REST API requires Port Based binding before you
# define Port-MAC Based binding.
ofc_port_id = port_id or uuidutils.generate_uuid()
dummy_port_id = "dummy-%s" % ofc_port_id
path = self.ports_path % {'network': ofc_network_id}
body = {'id': dummy_port_id,
'datapath_id': portinfo.datapath_id,
'port': str(portinfo.port_no),
'vid': str(portinfo.vlan_id)}
self.client.post(path, body=body)
path = self.attachments_path % {'network': ofc_network_id,
'port': dummy_port_id}
body = {'id': ofc_port_id, 'mac': portinfo.mac}
self.client.post(path, body=body)
path = self.port_path % {'network': ofc_network_id,
'port': dummy_port_id}
self.client.delete(path)
return self.attachment_path % {'network': ofc_network_id,
'port': dummy_port_id,
'attachment': ofc_port_id}
开发者ID:AmirolAhmad,项目名称:quantum,代码行数:26,代码来源:trema.py
示例14: test_fields
def test_fields(self):
return_value = {"name": "net1", "admin_state_up": True, "subnets": []}
instance = self.plugin.return_value
instance.get_network.return_value = return_value
self.api.get(_get_path("networks", id=uuidutils.generate_uuid()))
开发者ID:dani4571,项目名称:quantum,代码行数:7,代码来源:test_api_v2.py
示例15: create_dummy
def create_dummy(self, context, dummy):
d = dummy['dummy']
d['id'] = uuidutils.generate_uuid()
self.dummys[d['id']] = d
self.svctype_mgr.increase_service_type_refcount(context,
d['service_type'])
return d
开发者ID:Apsu,项目名称:quantum,代码行数:7,代码来源:dummy_plugin.py
示例16: create_vip
def create_vip(self, context, vip):
v = vip['vip']
tenant_id = self._get_tenant_id_for_create(context, v)
with context.session.begin(subtransactions=True):
if v['address'] is attributes.ATTR_NOT_SPECIFIED:
address = None
else:
address = v['address']
vip_db = Vip(id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=v['name'],
description=v['description'],
subnet_id=v['subnet_id'],
address=address,
port=v['port'],
protocol=v['protocol'],
pool_id=v['pool_id'],
connection_limit=v['connection_limit'],
admin_state_up=v['admin_state_up'],
status=constants.PENDING_CREATE)
vip_id = vip_db['id']
session_info = v['session_persistence']
if session_info:
s_p = self._create_session_persistence_db(session_info, vip_id)
vip_db.session_persistence = s_p
context.session.add(vip_db)
self._update_pool_vip_info(context, v['pool_id'], vip_id)
vip_db = self._get_resource(context, Vip, vip_id)
return self._make_vip_dict(vip_db)
开发者ID:StackOps,项目名称:quantum,代码行数:34,代码来源:loadbalancer_db.py
示例17: test_get_bridge
def test_get_bridge(self):
bridge_id = uuidutils.generate_uuid()
bridge = self.client.get_bridge(bridge_id)
self.assertIsNotNone(bridge)
self.assertEqual(bridge.get_id(), bridge_id)
开发者ID:Apsu,项目名称:quantum,代码行数:7,代码来源:test_midonet_lib.py
示例18: _test_list
def _test_list(self, req_tenant_id, real_tenant_id):
env = {}
if req_tenant_id:
env = {'quantum.context': context.Context('', req_tenant_id)}
input_dict = {'id': uuidutils.generate_uuid(),
'name': 'net1',
'admin_state_up': True,
'status': "ACTIVE",
'tenant_id': real_tenant_id,
'shared': False,
'subnets': []}
return_value = [input_dict]
instance = self.plugin.return_value
instance.get_networks.return_value = return_value
res = self.api.get(_get_path('networks'), extra_environ=env)
self.assertTrue('networks' in res.json)
if not req_tenant_id or req_tenant_id == real_tenant_id:
# expect full list returned
self.assertEqual(len(res.json['networks']), 1)
output_dict = res.json['networks'][0]
input_dict['shared'] = False
self.assertEqual(len(input_dict), len(output_dict))
for k, v in input_dict.iteritems():
self.assertEqual(v, output_dict[k])
else:
# expect no results
self.assertEqual(len(res.json['networks']), 0)
开发者ID:FreescaleSemiconductor,项目名称:quantum,代码行数:28,代码来源:test_api_v2.py
示例19: _test_list
def _test_list(self, req_tenant_id, real_tenant_id):
env = {}
if req_tenant_id:
env = {"quantum.context": context.Context("", req_tenant_id)}
input_dict = {
"id": uuidutils.generate_uuid(),
"name": "net1",
"admin_state_up": True,
"status": "ACTIVE",
"tenant_id": real_tenant_id,
"shared": False,
"subnets": [],
}
return_value = [input_dict]
instance = self.plugin.return_value
instance.get_networks.return_value = return_value
res = self.api.get(_get_path("networks"), extra_environ=env)
self.assertTrue("networks" in res.json)
if not req_tenant_id or req_tenant_id == real_tenant_id:
# expect full list returned
self.assertEqual(len(res.json["networks"]), 1)
output_dict = res.json["networks"][0]
input_dict["shared"] = False
self.assertEqual(len(input_dict), len(output_dict))
for k, v in input_dict.iteritems():
self.assertEqual(v, output_dict[k])
else:
# expect no results
self.assertEqual(len(res.json["networks"]), 0)
开发者ID:dani4571,项目名称:quantum,代码行数:30,代码来源:test_api_v2.py
示例20: setUp
def setUp(self):
super(MetaQuantumPluginV2Test, self).setUp()
db._ENGINE = None
db._MAKER = None
self.fake_tenant_id = uuidutils.generate_uuid()
self.context = context.get_admin_context()
db.configure_db()
setup_metaplugin_conf()
self.mox = mox.Mox()
self.stubs = stubout.StubOutForTesting()
args = ['--config-file', etcdir('quantum.conf.test')]
self.client_cls_p = mock.patch('quantumclient.v2_0.client.Client')
client_cls = self.client_cls_p.start()
self.client_inst = mock.Mock()
client_cls.return_value = self.client_inst
self.client_inst.create_network.return_value = \
{'id': 'fake_id'}
self.client_inst.create_port.return_value = \
{'id': 'fake_id'}
self.client_inst.create_subnet.return_value = \
{'id': 'fake_id'}
self.client_inst.update_network.return_value = \
{'id': 'fake_id'}
self.client_inst.update_port.return_value = \
{'id': 'fake_id'}
self.client_inst.update_subnet.return_value = \
{'id': 'fake_id'}
self.client_inst.delete_network.return_value = True
self.client_inst.delete_port.return_value = True
self.client_inst.delete_subnet.return_value = True
self.plugin = MetaPluginV2(configfile=None)
开发者ID:abhiraut,项目名称:quantum,代码行数:34,代码来源:test_metaplugin.py
注:本文中的quantum.openstack.common.uuidutils.generate_uuid函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论