Use oslo_context.from_dict() for context generation

Use RequestContext.from_dict in oslo_context to generate context, and
add/update needed attr such as user_id, tenant_id, tenant_name and
timestamp.

Closes-bug: #1933802
Change-Id: I0527eb5fa8d32d97ca45e44d1b154b6529b3f847
This commit is contained in:
Kiran Pawar 2022-09-29 08:40:30 +00:00
parent d9284ee76f
commit 9ecd5995b6
2 changed files with 28 additions and 9 deletions

View File

@ -106,15 +106,13 @@ class ContextBase(oslo_context.RequestContext):
@classmethod @classmethod
def from_dict(cls, values): def from_dict(cls, values):
return cls(user_id=values.get('user_id', values.get('user')), cls_obj = super().from_dict(values)
tenant_id=values.get('tenant_id', values.get('project_id')), if values.get('timestamp'):
is_admin=values.get('is_admin'), cls_obj.timestamp = values['timestamp']
roles=values.get('roles'), cls_obj.user_id = values.get('user_id', values.get('user'))
timestamp=values.get('timestamp'), cls_obj.tenant_id = values.get('tenant_id', values.get('project_id'))
request_id=values.get('request_id'), cls_obj.tenant_name = values.get('tenant_name')
tenant_name=values.get('tenant_name'), return cls_obj
user_name=values.get('user_name'),
auth_token=values.get('auth_token'))
def elevated(self): def elevated(self):
"""Return a version of this context with admin flag set.""" """Return a version of this context with admin flag set."""

View File

@ -87,6 +87,27 @@ class TestNeutronContext(_base.BaseTestCase):
self.assertEqual(owner['user_id'], ctx.user_id) self.assertEqual(owner['user_id'], ctx.user_id)
self.assertEqual(owner['tenant_id'], ctx.tenant_id) self.assertEqual(owner['tenant_id'], ctx.tenant_id)
def test_neutron_context_from_dict_validate(self):
ctx = context.Context(user_id='user_id1',
user_name='user',
tenant_id='tenant_id1',
tenant_name='project1',
request_id='request_id1',
auth_token='auth_token1')
values = {'user_id': 'user_id1',
'user_name': 'user',
'tenant_id': 'tenant_id1',
'tenant_name': 'project1',
'request_id': 'request_id1',
'auth_token': 'auth_token1'}
ctx2 = context.Context.from_dict(values)
ctx_dict = ctx.to_dict()
ctx_dict.pop('timestamp')
ctx2_dict = ctx2.to_dict()
ctx2_dict.pop('timestamp')
self.assertEqual(ctx_dict, ctx2_dict)
def test_neutron_context_to_dict(self): def test_neutron_context_to_dict(self):
ctx = context.Context('user_id', 'tenant_id') ctx = context.Context('user_id', 'tenant_id')
ctx_dict = ctx.to_dict() ctx_dict = ctx.to_dict()