14
14
# under the License.
17
from oslo_context import context as oslo_context
17
18
from testtools import matchers
19
20
from neutron import context
20
from neutron.openstack.common import local
21
21
from neutron.tests import base
130
130
def test_neutron_context_overwrite(self):
131
131
ctx1 = context.Context('user_id', 'tenant_id')
132
self.assertEqual(ctx1.request_id, local.store.context.request_id)
132
self.assertEqual(ctx1.request_id,
133
oslo_context.get_current().request_id)
134
135
# If overwrite is not specified, request_id should be updated.
135
136
ctx2 = context.Context('user_id', 'tenant_id')
136
137
self.assertNotEqual(ctx2.request_id, ctx1.request_id)
137
self.assertEqual(ctx2.request_id, local.store.context.request_id)
138
self.assertEqual(ctx2.request_id,
139
oslo_context.get_current().request_id)
139
141
# If overwrite is specified, request_id should be kept.
140
142
ctx3 = context.Context('user_id', 'tenant_id', overwrite=False)
141
143
self.assertNotEqual(ctx3.request_id, ctx2.request_id)
142
self.assertEqual(ctx2.request_id, local.store.context.request_id)
144
self.assertEqual(ctx2.request_id,
145
oslo_context.get_current().request_id)
144
147
def test_neutron_context_get_admin_context_not_update_local_store(self):
145
148
ctx = context.Context('user_id', 'tenant_id')
146
req_id_before = local.store.context.request_id
149
req_id_before = oslo_context.get_current().request_id
147
150
self.assertEqual(ctx.request_id, req_id_before)
149
152
ctx_admin = context.get_admin_context()
150
self.assertEqual(req_id_before, local.store.context.request_id)
153
self.assertEqual(req_id_before, oslo_context.get_current().request_id)
151
154
self.assertNotEqual(req_id_before, ctx_admin.request_id)