"""Utilities and helper functions."""
-
import os
import signal
-import uuid
from eventlet.green import subprocess
from quantum.openstack.common import cfg
from quantum.openstack.common import log as logging
+
TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
LOG = logging.getLogger(__name__)
return cfg_file
-def str_uuid():
- """Return a uuid as a string"""
- return str(uuid.uuid4())
-
-
def _subprocess_setup():
# Python installs a SIGPIPE handler by default. This is usually not what
# non-Python subprocesses expect.
-# Copyright (c) 2012 OpenStack, LLC.
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2012 OpenStack LLC.
+# All Rights Reserved.
#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
#
-# http://www.apache.org/licenses/LICENSE-2.0
+# http://www.apache.org/licenses/LICENSE-2.0
#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
import datetime
import random
from quantum.api.v2 import attributes
from quantum.common import constants
from quantum.common import exceptions as q_exc
-from quantum.common import utils
from quantum.db import api as db
from quantum.db import models_v2
from quantum.openstack.common import cfg
from quantum.openstack.common import log as logging
from quantum.openstack.common import timeutils
+from quantum.openstack.common import uuidutils
from quantum import quantum_plugin_base_v2
# unneeded db action if the operation raises
tenant_id = self._get_tenant_id_for_create(context, n)
with context.session.begin(subtransactions=True):
- network = models_v2.Network(tenant_id=tenant_id,
- id=n.get('id') or utils.str_uuid(),
- name=n['name'],
- admin_state_up=n['admin_state_up'],
- shared=n['shared'],
- status=constants.NET_STATUS_ACTIVE)
+ args = {'tenant_id': tenant_id,
+ 'id': n.get('id') or uuidutils.generate_uuid(),
+ 'name': n['name'],
+ 'admin_state_up': n['admin_state_up'],
+ 'shared': n['shared'],
+ 'status': constants.NET_STATUS_ACTIVE}
+ network = models_v2.Network(**args)
context.session.add(network)
return self._make_network_dict(network)
self._validate_subnet_cidr(context, network, s['cidr'])
# The 'shared' attribute for subnets is for internal plugin
# use only. It is not exposed through the API
- subnet = models_v2.Subnet(tenant_id=tenant_id,
- id=s.get('id') or utils.str_uuid(),
- name=s['name'],
- network_id=s['network_id'],
- ip_version=s['ip_version'],
- cidr=s['cidr'],
- enable_dhcp=s['enable_dhcp'],
- gateway_ip=s['gateway_ip'],
- shared=network.shared)
+ args = {'tenant_id': tenant_id,
+ 'id': s.get('id') or uuidutils.generate_uuid(),
+ 'name': s['name'],
+ 'network_id': s['network_id'],
+ 'ip_version': s['ip_version'],
+ 'cidr': s['cidr'],
+ 'enable_dhcp': s['enable_dhcp'],
+ 'gateway_ip': s['gateway_ip'],
+ 'shared': network.shared}
+ subnet = models_v2.Subnet(**args)
# perform allocate pools first, since it might raise an error
pools = self._allocate_pools_for_subnet(context, s)
def create_port(self, context, port):
p = port['port']
- port_id = p.get('id') or utils.str_uuid()
+ port_id = p.get('id') or uuidutils.generate_uuid()
network_id = p['network_id']
mac_address = p['mac_address']
# NOTE(jkoelker) Get the tenant_id outside of the session to avoid
-"""
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 Nicira Networks, Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
#
# @author: Dan Wendlandt, Nicira, Inc
#
-"""
import netaddr
import sqlalchemy as sa
from quantum.api.v2 import attributes
from quantum.common import exceptions as q_exc
-from quantum.common import utils
from quantum.db import db_base_plugin_v2
from quantum.db import model_base
from quantum.db import models_v2
from quantum.extensions import l3
from quantum.openstack.common import cfg
from quantum.openstack.common import log as logging
-
+from quantum.openstack.common import uuidutils
from quantum import policy
with context.session.begin(subtransactions=True):
# pre-generate id so it will be available when
# configuring external gw port
- router_db = Router(id=utils.str_uuid(),
+ router_db = Router(id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=r['name'],
admin_state_up=r['admin_state_up'],
def create_floatingip(self, context, floatingip):
fip = floatingip['floatingip']
tenant_id = self._get_tenant_id_for_create(context, fip)
- fip_id = utils.str_uuid()
+ fip_id = uuidutils.generate_uuid()
f_net_id = fip['floating_network_id']
if not self._network_is_external(context, f_net_id):
-# Copyright (c) 2012 OpenStack, LLC.
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2012 OpenStack LLC.
+# All Rights Reserved.
#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
#
-# http://www.apache.org/licenses/LICENSE-2.0
+# http://www.apache.org/licenses/LICENSE-2.0
#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
import sqlalchemy as sa
from sqlalchemy import orm
-from quantum.common import utils
from quantum.db import model_base
+from quantum.openstack.common import uuidutils
class HasTenant(object):
class HasId(object):
"""id mixin, add to subclasses that have an id."""
- id = sa.Column(sa.String(36), primary_key=True, default=utils.str_uuid)
+ id = sa.Column(sa.String(36),
+ primary_key=True,
+ default=uuidutils.generate_uuid)
class IPAvailabilityRange(model_base.BASEV2):
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 Nicira Networks, Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
from sqlalchemy.orm import exc
from sqlalchemy.orm import scoped_session
-from quantum.common import utils
from quantum.db import model_base
from quantum.db import models_v2
from quantum.extensions import securitygroup as ext_sg
with context.session.begin(subtransactions=True):
security_group_db = SecurityGroup(id=s.get('id') or (
- utils.str_uuid()),
+ uuidutils.generate_uuid()),
description=s['description'],
tenant_id=tenant_id,
name=s['name'],
for ethertype in self.sg_supported_ethertypes:
# Allow intercommunication
db = SecurityGroupRule(
- id=utils.str_uuid(), tenant_id=tenant_id,
+ id=uuidutils.generate_uuid(), tenant_id=tenant_id,
security_group=security_group_db,
direction='ingress',
ethertype=ethertype,
rule = rule_dict['security_group_rule']
tenant_id = self._get_tenant_id_for_create(context, rule)
db = SecurityGroupRule(
- id=utils.str_uuid(), tenant_id=tenant_id,
+ id=uuidutils.generate_uuid(), tenant_id=tenant_id,
security_group_id=rule['security_group_id'],
direction=rule['direction'],
external_id=rule.get('external_id'),
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
from sqlalchemy.orm import exc
from quantum.api.v2 import attributes
-from quantum.common import utils
from quantum.db import db_base_plugin_v2
+from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import exceptions as q_exc
from quantum.plugins.nec.db import models as nmodels
super(NECPluginV2Base, self).get_port(context, pf['in_port'])
params = {'tenant_id': tenant_id,
- 'id': pf.get('id') or utils.str_uuid(),
+ 'id': pf.get('id') or uuidutils.generate_uuid(),
'network_id': pf['network_id'],
'priority': pf['priority'],
'action': pf['action'],
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
import random
import unittest
-from quantum.common import utils
+from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import exceptions as nexc
from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.db import models as nmodels
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test"""
- ofc_id = utils.str_uuid()
- quantum_id = utils.str_uuid()
- none = utils.str_uuid()
+ ofc_id = uuidutils.generate_uuid()
+ quantum_id = uuidutils.generate_uuid()
+ none = uuidutils.generate_uuid()
return ofc_id, quantum_id, none
def testa_add_ofc_item(self):
def get_portinfo_random_params(self):
"""create random parameters for portinfo test"""
- port_id = utils.str_uuid()
+ 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 = utils.str_uuid()
+ none = uuidutils.generate_uuid()
return port_id, datapath_id, port_no, vlan_id, mac, none
def testd_add_portinfo(self):
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
import unittest
-from quantum.common import utils
+from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import config
from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.db import models as nmodels
def get_random_params(self):
"""create random parameters for portinfo test"""
- tenant = utils.str_uuid()
- network = utils.str_uuid()
- port = utils.str_uuid()
- _filter = utils.str_uuid()
- none = utils.str_uuid()
+ 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
def testa_create_ofc_tenant(self):
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
import mox
import unittest
-from quantum.common import utils
+from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import ofc_client as ofc
from quantum.plugins.nec.db import models as nmodels
from quantum.plugins.nec import drivers
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test"""
- tenant_id = utils.str_uuid()
- network_id = utils.str_uuid()
- port_id = utils.str_uuid()
+ 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")
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 NEC Corporation. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
import mox
import unittest
-from quantum.common import utils
+from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import ofc_client
from quantum.plugins.nec.db import models as nmodels
from quantum.plugins.nec import drivers
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test"""
- tenant_id = utils.str_uuid()
- network_id = utils.str_uuid()
- port_id = utils.str_uuid()
+ 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")
"""create random parameters for ofc_item test"""
t, n, p = (super(TremaFilterDriverTest, self).
get_ofc_item_random_params())
- filter_id = utils.str_uuid()
+ filter_id = uuidutils.generate_uuid()
filter_dict = {'tenant_id': t,
'id': filter_id,
'network_id': n,
-"""
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
+
# Copyright 2012 Nicira Networks, Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
#
# @author: Dan Wendlandt, Nicira, Inc
#
-"""
import contextlib
import copy
from quantum.common import config
from quantum.common import exceptions as q_exc
from quantum.common.test_lib import test_config
-from quantum.common import utils
from quantum import context
from quantum.db import db_base_plugin_v2
from quantum.db import l3_db
from quantum.extensions import l3
from quantum import manager
from quantum.openstack.common import cfg
+from quantum.openstack.common import uuidutils
from quantum.tests.unit import test_api_v2
from quantum.tests.unit import test_db_plugin
from quantum.tests.unit import test_extensions
+
LOG = logging.getLogger(__name__)
_uuid = test_api_v2._uuid
def test_create_floatingip_invalid_floating_network_id_returns_400(self):
# API-level test - no need to create all objects for l3 plugin
res = self._create_floatingip('json', 'iamnotanuuid',
- utils.str_uuid(), '192.168.0.1')
+ uuidutils.generate_uuid(), '192.168.0.1')
self.assertEqual(res.status_int, 400)
def test_create_floatingip_invalid_floating_port_id_returns_400(self):
# API-level test - no need to create all objects for l3 plugin
- res = self._create_floatingip('json', utils.str_uuid(),
+ res = self._create_floatingip('json', uuidutils.generate_uuid(),
'iamnotanuuid', '192.168.0.1')
self.assertEqual(res.status_int, 400)
def test_create_floatingip_invalid_fixed_ip_address_returns_400(self):
# API-level test - no need to create all objects for l3 plugin
- res = self._create_floatingip('json', utils.str_uuid(),
- utils.str_uuid(), 'iamnotnanip')
+ res = self._create_floatingip('json', uuidutils.generate_uuid(),
+ uuidutils.generate_uuid(), 'iamnotnanip')
self.assertEqual(res.status_int, 400)
def test_list_nets_external(self):
Code Review / openstack-build / neutron-build.git / commitdiff