neutron/tests/unit/test_context.py

   1 # Copyright 2012 VMware, Inc.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17 from oslo_context import context as oslo_context
  18 from testtools import matchers
  19
  20 from neutron import context
  21 from neutron.tests import base
  22
  23
  24 class TestNeutronContext(base.BaseTestCase):
  25
  26     def setUp(self):
  27         super(TestNeutronContext, self).setUp()
  28         db_api = 'neutron.db.api.get_session'
  29         self._db_api_session_patcher = mock.patch(db_api)
  30         self.db_api_session = self._db_api_session_patcher.start()
  31
  32     def test_neutron_context_create(self):
  33         ctx = context.Context('user_id', 'tenant_id')
  34         self.assertEqual('user_id', ctx.user_id)
  35         self.assertEqual('tenant_id', ctx.project_id)
  36         self.assertEqual('tenant_id', ctx.tenant_id)
  37         request_id = ctx.request_id
  38         if isinstance(request_id, bytes):
  39             request_id = request_id.decode('utf-8')
  40         self.assertThat(request_id, matchers.StartsWith('req-'))
  41         self.assertEqual('user_id', ctx.user)
  42         self.assertEqual('tenant_id', ctx.tenant)
  43         self.assertIsNone(ctx.user_name)
  44         self.assertIsNone(ctx.tenant_name)
  45         self.assertIsNone(ctx.auth_token)
  46
  47     def test_neutron_context_create_with_name(self):
  48         ctx = context.Context('user_id', 'tenant_id',
  49                               tenant_name='tenant_name', user_name='user_name')
  50         # Check name is set
  51         self.assertEqual('user_name', ctx.user_name)
  52         self.assertEqual('tenant_name', ctx.tenant_name)
  53         # Check user/tenant contains its ID even if user/tenant_name is passed
  54         self.assertEqual('user_id', ctx.user)
  55         self.assertEqual('tenant_id', ctx.tenant)
  56
  57     def test_neutron_context_create_with_request_id(self):
  58         ctx = context.Context('user_id', 'tenant_id', request_id='req_id_xxx')
  59         self.assertEqual('req_id_xxx', ctx.request_id)
  60
  61     def test_neutron_context_create_with_auth_token(self):
  62         ctx = context.Context('user_id', 'tenant_id',
  63                               auth_token='auth_token_xxx')
  64         self.assertEqual('auth_token_xxx', ctx.auth_token)
  65
  66     def test_neutron_context_to_dict(self):
  67         ctx = context.Context('user_id', 'tenant_id')
  68         ctx_dict = ctx.to_dict()
  69         self.assertEqual('user_id', ctx_dict['user_id'])
  70         self.assertEqual('tenant_id', ctx_dict['project_id'])
  71         self.assertEqual(ctx.request_id, ctx_dict['request_id'])
  72         self.assertEqual('user_id', ctx_dict['user'])
  73         self.assertEqual('tenant_id', ctx_dict['tenant'])
  74         self.assertIsNone(ctx_dict['user_name'])
  75         self.assertIsNone(ctx_dict['tenant_name'])
  76         self.assertIsNone(ctx_dict['project_name'])
  77         self.assertIsNone(ctx_dict['auth_token'])
  78
  79     def test_neutron_context_to_dict_with_name(self):
  80         ctx = context.Context('user_id', 'tenant_id',
  81                               tenant_name='tenant_name', user_name='user_name')
  82         ctx_dict = ctx.to_dict()
  83         self.assertEqual('user_name', ctx_dict['user_name'])
  84         self.assertEqual('tenant_name', ctx_dict['tenant_name'])
  85         self.assertEqual('tenant_name', ctx_dict['project_name'])
  86
  87     def test_neutron_context_to_dict_with_auth_token(self):
  88         ctx = context.Context('user_id', 'tenant_id',
  89                               auth_token='auth_token_xxx')
  90         ctx_dict = ctx.to_dict()
  91         self.assertEqual('auth_token_xxx', ctx_dict['auth_token'])
  92
  93     def test_neutron_context_admin_to_dict(self):
  94         self.db_api_session.return_value = 'fakesession'
  95         ctx = context.get_admin_context()
  96         ctx_dict = ctx.to_dict()
  97         self.assertIsNone(ctx_dict['user_id'])
  98         self.assertIsNone(ctx_dict['tenant_id'])
  99         self.assertIsNone(ctx_dict['auth_token'])
 100         self.assertIsNotNone(ctx.session)
 101         self.assertNotIn('session', ctx_dict)
 102
 103     def test_neutron_context_admin_without_session_to_dict(self):
 104         ctx = context.get_admin_context_without_session()
 105         ctx_dict = ctx.to_dict()
 106         self.assertIsNone(ctx_dict['user_id'])
 107         self.assertIsNone(ctx_dict['tenant_id'])
 108         self.assertIsNone(ctx_dict['auth_token'])
 109         self.assertFalse(hasattr(ctx, 'session'))
 110
 111     def test_neutron_context_elevated_retains_request_id(self):
 112         ctx = context.Context('user_id', 'tenant_id')
 113         self.assertFalse(ctx.is_admin)
 114         req_id_before = ctx.request_id
 115
 116         elevated_ctx = ctx.elevated()
 117         self.assertTrue(elevated_ctx.is_admin)
 118         self.assertEqual(req_id_before, elevated_ctx.request_id)
 119
 120     def test_neutron_context_overwrite(self):
 121         ctx1 = context.Context('user_id', 'tenant_id')
 122         self.assertEqual(ctx1.request_id,
 123                          oslo_context.get_current().request_id)
 124
 125         # If overwrite is not specified, request_id should be updated.
 126         ctx2 = context.Context('user_id', 'tenant_id')
 127         self.assertNotEqual(ctx2.request_id, ctx1.request_id)
 128         self.assertEqual(ctx2.request_id,
 129                          oslo_context.get_current().request_id)
 130
 131         # If overwrite is specified, request_id should be kept.
 132         ctx3 = context.Context('user_id', 'tenant_id', overwrite=False)
 133         self.assertNotEqual(ctx3.request_id, ctx2.request_id)
 134         self.assertEqual(ctx2.request_id,
 135                          oslo_context.get_current().request_id)
 136
 137     def test_neutron_context_get_admin_context_not_update_local_store(self):
 138         ctx = context.Context('user_id', 'tenant_id')
 139         req_id_before = oslo_context.get_current().request_id
 140         self.assertEqual(ctx.request_id, req_id_before)
 141
 142         ctx_admin = context.get_admin_context()
 143         self.assertEqual(req_id_before, oslo_context.get_current().request_id)
 144         self.assertNotEqual(req_id_before, ctx_admin.request_id)