"""Context: context for security/db session."""
import copy
-
import datetime
+from oslo_context import context as oslo_context
+
from neutron.db import api as db_api
-from neutron.openstack.common import context as common_context
from neutron.openstack.common import local
from neutron.openstack.common import log as logging
from neutron import policy
LOG = logging.getLogger(__name__)
-class ContextBase(common_context.RequestContext):
+class ContextBase(oslo_context.RequestContext):
"""Security context and request information.
Represents the user taking a given action within the system.
+++ /dev/null
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2011 OpenStack Foundation.
-# 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
-#
-# 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.
-
-"""
-Simple class that stores security context information in the web request.
-
-Projects should subclass this class if they wish to enhance the request
-context or provide additional information in their specific WSGI pipeline.
-"""
-
-import itertools
-
-from neutron.openstack.common import uuidutils
-
-
-def generate_request_id():
- return 'req-%s' % uuidutils.generate_uuid()
-
-
-class RequestContext(object):
-
- """Helper class to represent useful information about a request context.
-
- Stores information about the security context under which the user
- accesses the system, as well as additional request information.
- """
-
- def __init__(self, auth_token=None, user=None, tenant=None, is_admin=False,
- read_only=False, show_deleted=False, request_id=None):
- self.auth_token = auth_token
- self.user = user
- self.tenant = tenant
- self.is_admin = is_admin
- self.read_only = read_only
- self.show_deleted = show_deleted
- if not request_id:
- request_id = generate_request_id()
- self.request_id = request_id
-
- def to_dict(self):
- return {'user': self.user,
- 'tenant': self.tenant,
- 'is_admin': self.is_admin,
- 'read_only': self.read_only,
- 'show_deleted': self.show_deleted,
- 'auth_token': self.auth_token,
- 'request_id': self.request_id}
-
-
-def get_admin_context(show_deleted="no"):
- context = RequestContext(None,
- tenant=None,
- is_admin=True,
- show_deleted=show_deleted)
- return context
-
-
-def get_context_from_function_and_args(function, args, kwargs):
- """Find an arg of type RequestContext and return it.
-
- This is useful in a couple of decorators where we don't
- know much about the function we're wrapping.
- """
-
- for arg in itertools.chain(kwargs.values(), args):
- if isinstance(arg, RequestContext):
- return arg
-
- return None
from oslo.config import cfg
from oslo import messaging
from oslo.utils import importutils
+from oslo_context import context as oslo_context
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.db import securitygroups_rpc_base as sg_db_rpc
from neutron.extensions import portbindings
from neutron.extensions import securitygroup as ext_sg
-from neutron.openstack.common import context
from neutron.i18n import _LE, _LI
from neutron.openstack.common import log as logging
from neutron.plugins.brocade.db import models as brocade_db
physical_interface)
self.base_binding_dict = self._get_base_binding_dict()
portbindings_base.register_port_dict_function()
- self.ctxt = context.get_admin_context()
+ self.ctxt = oslo_context.get_admin_context()
self.ctxt.session = db.get_session()
self._vlan_bitmap = vbm.VlanBitmap(self.ctxt)
self._setup_rpc()
# RPC support
self.service_topics = {svc_constants.CORE: topics.PLUGIN,
svc_constants.L3_ROUTER_NAT: topics.L3PLUGIN}
- self.rpc_context = context.RequestContext('neutron', 'neutron',
+ self.rpc_context = oslo_context.RequestContext('neutron', 'neutron',
is_admin=False)
self.conn = n_rpc.create_connection(new=True)
self.endpoints = [BridgeRpcCallbacks(),
Test vlans alloc/dealloc.
"""
+from oslo_context import context as oslo_context
+
from neutron.db import api as db
-from neutron.openstack.common import context
from neutron.plugins.brocade import vlanbm as vlan_bitmap
from neutron.tests.unit import testlib_api
def setUp(self):
super(TestVlanBitmap, self).setUp()
- self.context = context.get_admin_context()
+ self.context = oslo_context.get_admin_context()
self.context.session = db.get_session()
def test_vlan(self):
"""
import contextlib
+
import mock
+from oslo_context import context as oslo_context
from neutron.agent import rpc as agent_rpc
from neutron.common import topics
-from neutron.openstack.common import context
from neutron.plugins.hyperv import agent_notifier_api as ana
from neutron.plugins.hyperv.common import constants
from neutron.tests import base
def _test_hyperv_neutron_api(
self, rpcapi, topic, method, rpc_method, **kwargs):
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
expected_retval = 'foo' if rpc_method == 'call' else None
expected_version = kwargs.pop('version', None)
fanout = kwargs.pop('fanout', False)
import mock
from oslo.config import cfg
from oslo.utils import importutils
+from oslo_context import context as oslo_context
from neutron.db import api as db
-from neutron.openstack.common import context
from neutron.openstack.common import log as logging
from neutron.tests.unit import test_l3_plugin
LOG.info(_("rbridge id %s"), self._switch['rbridge_id'])
self._driver = mock.MagicMock()
- self.context = context.get_admin_context()
+ self.context = oslo_context.get_admin_context()
self.context.session = db.get_session()
self.l3_plugin = importutils.import_object(L3_SVC_PLUGIN)
with mock.patch.object(self.l3_plugin,
import contextlib
import mock
+from oslo_context import context as oslo_context
from neutron.agent import rpc as agent_rpc
from neutron.common import constants
from neutron.common import exceptions
from neutron.common import topics
-from neutron.openstack.common import context
from neutron.plugins.ml2.drivers import type_tunnel
from neutron.plugins.ml2 import rpc as plugin_rpc
from neutron.tests import base
class RpcApiTestCase(base.BaseTestCase):
def _test_rpc_api(self, rpcapi, topic, method, rpc_method, **kwargs):
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
expected_retval = 'foo' if rpc_method == 'call' else None
expected_version = kwargs.pop('version', None)
fanout = kwargs.pop('fanout', False)
import mock
from oslo.config import cfg
+from oslo_context import context as oslo_context
from neutron.agent import rpc as agent_rpc
from neutron.common import topics
-from neutron.openstack.common import context
from neutron.plugins.mlnx import agent_notify_api
from neutron.tests import base
class rpcApiTestCase(base.BaseTestCase):
def _test_mlnx_api(self, rpcapi, topic, method, rpc_method, **kwargs):
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
expected_retval = 'foo' if rpc_method == 'call' else None
expected_version = kwargs.pop('version', None)
fanout = kwargs.pop('fanout', False)
# under the License.
import contextlib
+
import mock
from oslo import messaging
+from oslo_context import context as oslo_context
from neutron.agent import rpc
-from neutron.openstack.common import context
from neutron.tests import base
class AgentRPCPluginApi(base.BaseTestCase):
def _test_rpc_call(self, method):
agent = rpc.PluginApi('fake_topic')
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
expect_val = 'foo'
with contextlib.nested(
mock.patch.object(agent.client, 'call'),
def test_devices_details_list_unsupported(self):
agent = rpc.PluginApi('fake_topic')
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
expect_val_get_device_details = 'foo'
expect_val = [expect_val_get_device_details]
with contextlib.nested(
mock_call, mock_cast, mock_prepare
):
mock_prepare.return_value = reportStateAPI.client
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
reportStateAPI.report_state(ctxt, expected_agent_state,
use_call=True)
self.assertEqual(mock_call.call_args[0][0], ctxt)
mock_call, mock_cast, mock_prepare
):
mock_prepare.return_value = reportStateAPI.client
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = oslo_context.RequestContext('fake_user', 'fake_project')
reportStateAPI.report_state(ctxt, expected_agent_state)
self.assertEqual(mock_cast.call_args[0][0], ctxt)
self.assertEqual(mock_cast.call_args[0][1], 'report_state')
[DEFAULT]
# The list of modules to copy from oslo-incubator.git
module=cache
-module=context
module=eventlet_backdoor
module=fileutils
module=fixture
six>=1.7.0
stevedore>=1.1.0 # Apache-2.0
oslo.config>=1.4.0 # Apache-2.0
+oslo.context>=0.1.0 # Apache-2.0
oslo.db>=1.1.0 # Apache-2.0
oslo.i18n>=1.0.0 # Apache-2.0
oslo.messaging>=1.4.0,!=1.5.0
Code Review / openstack-build / neutron-build.git / commitdiff