openstack/heat
Code Issues Proposed changes

Merge "Rename admin flag on SqlAlchemy#stack_get"

Browse Source
This commit is contained in:
Jenkins 2013-11-01 18:03:40 +00:00 committed by Gerrit Code Review
commit 13ec482b62
4 changed files with 24 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
heat/db/api.py
View File

@ -102,8 +102,9 @@ def resource_get_by_physical_resource_id(context, physical_resource_id):
physical_resource_id)
def stack_get(context, stack_id, admin=False, show_deleted=False):
return IMPL.stack_get(context, stack_id, admin, show_deleted=show_deleted)
def stack_get(context, stack_id, show_deleted=False, tenant_safe=True):
return IMPL.stack_get(context, stack_id, show_deleted=show_deleted,
tenant_safe=tenant_safe)
def stack_get_by_name(context, stack_name, owner_id=None):

9
heat/db/sqlalchemy/api.py
View File

@ -221,18 +221,13 @@ def stack_get_by_name(context, stack_name, owner_id=None):
return query.first()
def stack_get(context, stack_id, admin=False, show_deleted=False):
def stack_get(context, stack_id, show_deleted=False, tenant_safe=True):
result = model_query(context, models.Stack).get(stack_id)
if result is None or result.deleted_at is not None and not show_deleted:
return None
# If the admin flag is True, we allow retrieval of a specific
# stack without the tenant scoping
if admin:
return result
if (result is not None and context is not None and
if (tenant_safe and result is not None and context is not None and
result.tenant != context.tenant_id):
return None

4
heat/engine/service.py
View File

@ -696,11 +696,11 @@ class EngineService(service.Service):
def _check_stack_watches(self, sid):
# Retrieve the stored credentials & create context
# Require admin=True to the stack_get to defeat tenant
# Require tenant_safe=False to the stack_get to defeat tenant
# scoping otherwise we fail to retrieve the stack
logger.debug("Periodic watcher task for stack %s" % sid)
admin_context = context.get_admin_context()
stack = db_api.stack_get(admin_context, sid, admin=True)
stack = db_api.stack_get(admin_context, sid, tenant_safe=False)
if not stack:
logger.error("Unable to retrieve stack %s for periodic task" %
sid)

24
heat/tests/test_sqlalchemy_api.py
View File

@ -533,7 +533,8 @@ class DBAPIStackTest(HeatTestCase):
stack_id = stack.id
resource = create_resource(self.ctx, stack)
db_api.stack_delete(self.ctx, stack_id)
self.assertIsNone(db_api.stack_get(self.ctx, stack_id, False, False))
self.assertIsNone(db_api.stack_get(self.ctx, stack_id,
show_deleted=False))
self.assertRaises(exception.NotFound, db_api.resource_get,
self.ctx, resource.id)
@ -541,7 +542,7 @@ class DBAPIStackTest(HeatTestCase):
self.ctx, stack_id)
#Testing soft delete
ret_stack = db_api.stack_get(self.ctx, stack_id, False, True)
ret_stack = db_api.stack_get(self.ctx, stack_id, show_deleted=True)
self.assertIsNotNone(ret_stack)
self.assertEqual(stack_id, ret_stack.id)
self.assertEqual('db_test_stack_name', ret_stack.name)
@ -570,19 +571,28 @@ class DBAPIStackTest(HeatTestCase):
self.assertRaises(exception.NotFound, db_api.stack_update, self.ctx,
UUID2, values)
def test_stack_get(self):
def test_stack_get_returns_a_stack(self):
stack = create_stack(self.ctx, self.template, self.user_creds)
ret_stack = db_api.stack_get(self.ctx, stack.id, False, False)
ret_stack = db_api.stack_get(self.ctx, stack.id, show_deleted=False)
self.assertIsNotNone(ret_stack)
self.assertEqual(stack.id, ret_stack.id)
self.assertEqual('db_test_stack_name', ret_stack.name)
self.assertIsNone(db_api.stack_get(self.ctx, UUID1, False, False))
def test_stack_get_returns_none_if_stack_does_not_exist(self):
stack = db_api.stack_get(self.ctx, UUID1, show_deleted=False)
self.assertIsNone(stack)
def test_stack_get_returns_none_if_tenant_id_does_not_match(self):
stack = create_stack(self.ctx, self.template, self.user_creds)
self.ctx.tenant_id = 'abc'
self.assertIsNone(db_api.stack_get(self.ctx, stack.id, False, False))
stack = db_api.stack_get(self.ctx, UUID1, show_deleted=False)
self.assertIsNone(stack)
ret_stack = db_api.stack_get(self.ctx, stack.id, True, False)
def test_stack_get_can_return_a_stack_from_different_tenant(self):
stack = create_stack(self.ctx, self.template, self.user_creds)
self.ctx.tenant_id = 'abc'
ret_stack = db_api.stack_get(self.ctx, stack.id,
show_deleted=False, tenant_safe=False)
self.assertEqual(stack.id, ret_stack.id)
self.assertEqual('db_test_stack_name', ret_stack.name)