from oslo.config import cfg
-from heat.common import context
from heat.common import identifier
from heat.common import policy
from heat.openstack.common import rpc
from heat.api.aws import exception
import heat.api.cfn.v1.stacks as stacks
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
- # Utility functions
- def _create_context(self, user='api_test_user'):
- ctx = context.get_admin_context()
- self.m.StubOutWithMock(ctx, 'username')
- ctx.username = user
- self.m.StubOutWithMock(ctx, 'tenant_id')
- ctx.tenant_id = 't'
- return ctx
def _dummy_GET_request(self, params={}):
# Mangle the params dict into a query string
qs = "&".join(["=".join([k, str(params[k])]) for k in params])
environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': qs}
req = Request(environ)
- req.context = self._create_context()
+ req.context = dummy_context()
return req
# The tests
from oslo.config import cfg
-from heat.common import context
from heat.common import policy
from heat.openstack.common import rpc
from heat.common.wsgi import Request
import heat.api.cloudwatch.watch as watches
from heat.rpc import api as engine_api
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
class WatchControllerTest(HeatTestCase):
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
- # Utility functions
- def _create_context(self, user='api_test_user'):
- ctx = context.get_admin_context()
- self.m.StubOutWithMock(ctx, 'username')
- ctx.username = user
- return ctx
def _dummy_GET_request(self, params={}):
# Mangle the params dict into a query string
qs = "&".join(["=".join([k, str(params[k])]) for k in params])
environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': qs}
req = Request(environ)
- req.context = self._create_context()
+ req.context = dummy_context()
return req
# The tests
from oslo.config import cfg
import webob.exc
-from heat.common import context
from heat.common import identifier
from heat.openstack.common import rpc
import heat.openstack.common.rpc.common as rpc_common
import heat.api.openstack.v1.resources as resources
import heat.api.openstack.v1.events as events
import heat.api.openstack.v1.actions as actions
+from heat.tests.utils import dummy_context
class InstantiationDataTest(HeatTestCase):
self.api_version = '1.0'
self.tenant = 't'
- def _create_context(self, user='api_test_user'):
- ctx = context.get_admin_context()
- self.m.StubOutWithMock(ctx, 'username')
- ctx.username = user
- self.m.StubOutWithMock(ctx, 'tenant_id')
- ctx.tenant_id = self.tenant
- return ctx
-
def _environ(self, path):
return {
'SERVER_NAME': 'heat.example.com',
environ['REQUEST_METHOD'] = method
req = Request(environ)
- req.context = self._create_context()
+ req.context = dummy_context('api_test_user', self.tenant)
return req
def _get(self, path):
environ['REQUEST_METHOD'] = method
req = Request(environ)
- req.context = self._create_context()
+ req.context = dummy_context('api_test_user', self.tenant)
req.body = data
return req
from heat.engine.resources.rackspace import clouddatabase
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
stack_name = '%s_stack' % name
t = template_format.parse(wp_template)
template = parser.Template(t)
- stack = parser.Stack(None,
+ stack = parser.Stack(dummy_context(),
stack_name,
template,
environment.Environment({'InstanceName': 'Test',
from oslo.config import cfg
-from heat.common import context
from heat.common import policy
from heat.common import exception
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
def test_policy_cfn_default(self):
enforcer = policy.Enforcer(scope='cloudformation')
- ctx = context.RequestContext(roles=[])
+ ctx = dummy_context(roles=[])
for action in self.cfn_actions:
# Everything should be allowed
enforcer.enforce(ctx, action, {})
enforcer = policy.Enforcer(scope='cloudformation')
- ctx = context.RequestContext(roles=[])
+ ctx = dummy_context(roles=[])
for action in self.cfn_actions:
# Everything should raise the default exception.Forbidden
self.assertRaises(exception.Forbidden, enforcer.enforce, ctx,
enforcer = policy.Enforcer(scope='cloudformation')
- ctx = context.RequestContext(roles=['heat_stack_user'])
+ ctx = dummy_context(roles=['heat_stack_user'])
for action in self.cfn_actions:
# Everything apart from DescribeStackResource should be Forbidden
if action == "DescribeStackResource":
enforcer = policy.Enforcer(scope='cloudformation')
- ctx = context.RequestContext(roles=['not_a_stack_user'])
+ ctx = dummy_context(roles=['not_a_stack_user'])
for action in self.cfn_actions:
# Everything should be allowed
enforcer.enforce(ctx, action, {})
enforcer = policy.Enforcer(scope='cloudwatch')
- ctx = context.RequestContext(roles=['heat_stack_user'])
+ ctx = dummy_context(roles=['heat_stack_user'])
for action in self.cw_actions:
# Everything apart from PutMetricData should be Forbidden
if action == "PutMetricData":
enforcer = policy.Enforcer(scope='cloudwatch')
- ctx = context.RequestContext(roles=['not_a_stack_user'])
+ ctx = dummy_context(roles=['not_a_stack_user'])
for action in self.cw_actions:
# Everything should be allowed
enforcer.enforce(ctx, action, {})
# License for the specific language governing permissions and limitations
# under the License.
-from heat.common import context
import heat.engine.api as api
from heat.engine import parser
from heat.engine import resource
from heat.rpc import api as rpc_api
from heat.tests.common import HeatTestCase
from heat.tests import generic_resource as generic_rsrc
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
def setUp(self):
super(FormatTest, self).setUp()
setup_dummy_db()
- ctx = context.get_admin_context()
- self.m.StubOutWithMock(ctx, 'user')
- ctx.user = 'test_user'
- ctx.tenant_id = 'test_tenant'
template = parser.Template({
'Resources': {
})
resource._register_class('GenericResourceType',
generic_rsrc.GenericResource)
- self.stack = parser.Stack(ctx, 'test_stack', template,
+ self.stack = parser.Stack(dummy_context(), 'test_stack', template,
stack_id=uuidutils.generate_uuid())
def test_format_stack_resource(self):
from oslo.config import cfg
-from heat.common import context
from heat.engine import environment
from heat.common import exception
from heat.tests.v1_1 import fakes
from heat.openstack.common import threadgroup
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
def create_context(mocks, user='stacks_test_user',
- tenant='test_admin', password='stacks_test_password'):
- ctx = context.get_admin_context()
+ tenant_id='test_admin', password='stacks_test_password'):
+ ctx = dummy_context()
mocks.StubOutWithMock(ctx, 'username')
mocks.StubOutWithMock(ctx, 'tenant_id')
mocks.StubOutWithMock(ctx, 'password')
ctx.username = user
- ctx.tenant_id = tenant
+ ctx.tenant_id = tenant_id
ctx.password = password
return ctx
self.assertNotEqual(stack.resources['WebServer'].ipaddress, '0.0.0.0')
def test_wordpress_single_instance_stack_delete(self):
- ctx = create_context(self.m, tenant='test_delete_tenant')
+ ctx = create_context(self.m)
stack = get_wordpress_stack('test_stack', ctx)
fc = setup_mocks(self.m, stack)
self.m.ReplayAll()
def setUp(self):
super(StackServiceCreateUpdateDeleteTest, self).setUp()
- self.username = 'stack_service_create_test_user'
- self.tenant = 'stack_service_create_test_tenant'
setup_dummy_db()
- self.ctx = create_context(self.m, self.username, self.tenant)
+ self.ctx = create_context(self.m)
self.man = service.EngineService('a-host', 'a-topic')
def setUp(self):
super(StackServiceSuspendResumeTest, self).setUp()
- self.username = 'stack_service_suspend_test_user'
- self.tenant = 'stack_service_suspend_test_tenant'
setup_dummy_db()
- self.ctx = create_context(self.m, self.username, self.tenant)
+ self.ctx = create_context(self.m)
self.man = service.EngineService('a-host', 'a-topic')
def setUp(self):
super(StackServiceTest, self).setUp()
- self.username = 'stack_service_test_user'
- self.tenant = 'stack_service_test_tenant'
-
- self.ctx = create_context(self.m, self.username, self.tenant)
+ self.ctx = create_context(self.m,
+ tenant_id='stack_service_test_tenant')
self.eng = service.EngineService('a-host', 'a-topic')
cfg.CONF.set_default('heat_stack_user_role', 'stack_user_role')
self.assertEqual(self.stack.id,
db_api.stack_get_by_name(self.ctx,
self.stack.name).id)
- ctx2 = create_context(self.m, self.username,
- 'stack_service_test_tenant2')
+ ctx2 = create_context(self.m, tenant_id='stack_service_test_tenant2')
self.assertEqual(None, db_api.stack_get_by_name(ctx2, self.stack.name))
@stack_context('service_event_list_test_stack')
service.EngineService._get_stack(self.ctx,
self.stack.identifier()).AndReturn(s)
self.m.StubOutWithMock(instances.Instance, 'metadata_update')
+ self.m.StubOutWithMock(db_api, 'user_creds_get')
instances.Instance.metadata_update(new_metadata=test_metadata)
+ db_api.user_creds_get(mox.IgnoreArg()).MultipleTimes().AndReturn(
+ self.ctx.to_dict())
self.m.ReplayAll()
result = self.eng.metadata_update(self.ctx,
# under the License.
-from heat.common import context
import heat.db.api as db_api
from heat.engine import parser
from heat.engine import resource
from heat.engine import event
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
from heat.tests import generic_resource as generic_rsrc
self.username = 'event_test_user'
setup_dummy_db()
- self.ctx = context.get_admin_context()
- self.m.StubOutWithMock(self.ctx, 'username')
- self.ctx.username = self.username
+ self.ctx = dummy_context()
self.m.ReplayAll()
import mox
-from heat.common import context
from heat.common import heat_keystoneclient
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
class KeystoneClientTest(HeatTestCase):
# verify all the things
self.addCleanup(self.m.VerifyAll)
- def _create_context(self, user='stacks_test_user',
- tenant='test_admin', password='test_password',
- auth_url="auth_url", tenant_id='tenant_id', ctx=None):
- """
- :returns: A test context
- """
- ctx = ctx or context.get_admin_context()
- ctx.auth_url = auth_url
- ctx.username = user
- ctx.password = password
- ctx.tenant_id = tenant_id
- ctx.tenant = tenant
- return ctx
-
def test_username_length(self):
"""Test that user names >64 characters are properly truncated."""
# long_user_name, keystone was actually called with a truncated
# user name
heat_ks_client = heat_keystoneclient.KeystoneClient(
- self._create_context())
+ dummy_context())
heat_ks_client.create_stack_user(long_user_name, password='password')
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
def _setup_test_stack(self, stack_name):
t = template_format.parse(wp_template)
template = parser.Template(t)
- stack = parser.Stack(None, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
environment.Environment({'KeyName': 'test'}),
stack_id=uuidutils.generate_uuid())
return (t, stack)
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
kwargs = {'KeyName': 'test',
'InstanceType': 'm1.large',
'SubnetId': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861'}
- stack = parser.Stack(None, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
environment.Environment(kwargs),
stack_id=uuidutils.generate_uuid())
kwargs = {'KeyName': 'test',
'InstanceType': 'm1.large',
'SubnetId': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861'}
- stack = parser.Stack(None, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
environment.Environment(kwargs),
stack_id=uuidutils.generate_uuid())
from oslo.config import cfg
from heat.tests import fakes
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import stack_delete_after
+from heat.db import api as db_api
from heat.engine import environment
from heat.common import identifier
from heat.common import template_format
from heat.engine import scheduler
from heat.engine import service
from heat.engine.resources import instance
-from heat.common import context
from heat.engine.resources import wait_condition as wc
def create_stack(self, stack_name='test_stack', params={}):
temp = template_format.parse(test_template_metadata)
template = parser.Template(temp)
- ctx = context.get_admin_context()
- ctx.tenant_id = 'test_tenant'
+ ctx = dummy_context()
stack = parser.Stack(ctx, stack_name, template,
environment.Environment(params),
disable_rollback=True)
def setUp(self):
super(WaitCondMetadataUpdateTest, self).setUp()
setup_dummy_db()
- self.ctx = context.get_admin_context()
- self.ctx.tenant_id = 'test_tenant'
self.fc = fakes.FakeKeystoneClient()
self.man = service.EngineService('a-host', 'a-topic')
cfg.CONF.set_default('heat_waitcondition_server_url',
def create_stack(self, stack_name='test_stack'):
temp = template_format.parse(test_template_waitcondition)
template = parser.Template(temp)
- stack = parser.Stack(self.ctx, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
disable_rollback=True)
self.stack_id = stack.store()
self.m.StubOutWithMock(wc.WaitConditionHandle, 'keystone')
wc.WaitConditionHandle.keystone().MultipleTimes().AndReturn(self.fc)
- id = identifier.ResourceIdentifier('test_tenant', stack.name,
+ id = identifier.ResourceIdentifier('test_tenant_id', stack.name,
stack.id, '', 'WH')
self.m.StubOutWithMock(wc.WaitConditionHandle, 'identifier')
wc.WaitConditionHandle.identifier().MultipleTimes().AndReturn(id)
self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ self.m.StubOutWithMock(db_api, 'user_creds_get')
return stack
self.assertEqual(inst.metadata['test'], None)
def update_metadata(id, data, reason):
- self.man.metadata_update(self.ctx,
+ self.man.metadata_update(dummy_context(),
dict(self.stack.identifier()),
'WH',
{'Data': data, 'Reason': reason,
scheduler.TaskRunner._sleep(mox.IsA(int)).WithSideEffects(check_empty)
scheduler.TaskRunner._sleep(mox.IsA(int)).WithSideEffects(post_success)
+ db_api.user_creds_get(mox.IgnoreArg()).MultipleTimes().AndReturn(
+ self.stack.context.to_dict())
scheduler.TaskRunner._sleep(mox.IsA(int)).AndReturn(None)
self.m.ReplayAll()
# under the License.
-from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.common import urlfetch
from heat.tests import generic_resource as generic_rsrc
from heat.tests import utils
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
return stack
def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'tenant_id': 'aaaa',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
+ ctx = dummy_context('test_username', 'aaaa', 'password')
stack_name = 'test_stack'
tmpl = parser.Template(t)
stack = parser.Stack(ctx, stack_name, tmpl)
import time
import uuid
-from heat.common import context
from heat.engine import environment
from heat.common import exception
from heat.common import template_format
from heat.engine import template
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
from heat.tests.v1_1 import fakes
from heat.tests.utils import stack_delete_after
class TemplateTest(HeatTestCase):
+
+ def setUp(self):
+ super(TemplateTest, self).setUp()
+ self.ctx = dummy_context()
+
def test_defaults(self):
empty = parser.Template({})
try:
def test_get_azs_with_stack(self):
snippet = {"Fn::GetAZs": ""}
- stack = parser.Stack(None, 'test_stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}))
self.m.StubOutWithMock(clients.OpenStackClients, 'nova')
fc = fakes.FakeClient()
clients.OpenStackClients.nova().MultipleTimes().AndReturn(fc)
parent_resource.metadata = '{"foo": "bar"}'
parent_resource.t = {'DeletionPolicy': 'Retain',
'UpdatePolicy': '{"foo": "bar"}'}
- stack = parser.Stack(None, 'test_stack',
+ stack = parser.Stack(self.ctx, 'test_stack',
parser.Template({}),
parent_resource=parent_resource)
self.assertEqual(
def test_resource_facade_invalid_arg(self):
snippet = {'Fn::ResourceFacade': 'wibble'}
- stack = parser.Stack(None, 'test_stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}))
self.assertRaises(ValueError,
parser.Template.resolve_resource_facade,
snippet,
parent_resource = DummyClass()
parent_resource.metadata = '{"foo": "bar"}'
parent_resource.t = {}
- stack = parser.Stack(None, 'test_stack',
+ stack = parser.Stack(self.ctx, 'test_stack',
parser.Template({}),
parent_resource=parent_resource)
self.assertRaises(KeyError,
self.username = 'parser_stack_test_user'
setup_dummy_db()
- self.ctx = context.get_admin_context()
- self.m.StubOutWithMock(self.ctx, 'user')
- self.ctx.user = self.username
- self.ctx.tenant_id = 'test_tenant'
+ self.ctx = dummy_context()
resource._register_class('GenericResourceType',
generic_rsrc.GenericResource)
self.m.ReplayAll()
def test_state_defaults(self):
- stack = parser.Stack(None, 'test_stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}))
self.assertEqual(stack.state, (None, None))
self.assertEqual(stack.status_reason, '')
def test_state(self):
- stack = parser.Stack(None, 'test_stack', parser.Template({}),
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}),
action=parser.Stack.CREATE,
status=parser.Stack.IN_PROGRESS)
self.assertEqual(stack.state,
(parser.Stack.DELETE, parser.Stack.COMPLETE))
def test_state_bad(self):
- stack = parser.Stack(None, 'test_stack', parser.Template({}),
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}),
action=parser.Stack.CREATE,
status=parser.Stack.IN_PROGRESS)
self.assertEqual(stack.state,
parser.Stack.CREATE, 'oops', 'test')
def test_status_reason(self):
- stack = parser.Stack(None, 'test_stack', parser.Template({}),
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}),
status_reason='quux')
self.assertEqual(stack.status_reason, 'quux')
stack.state_set(parser.Stack.CREATE, parser.Stack.IN_PROGRESS,
self.m.ReplayAll()
self.stack = parser.Stack(self.ctx, 'param_arn_test',
parser.Template({}))
- exp_prefix = 'arn:openstack:heat::test_tenant:stacks/param_arn_test/'
+ exp_prefix = ('arn:openstack:heat::test_tenant_id'
+ ':stacks/param_arn_test/')
self.assertEqual(self.stack.parameters['AWS::StackId'],
exp_prefix + 'None')
self.stack.store()
self.m.StubOutWithMock(scheduler.DependencyTaskGroup, '__call__')
self.m.StubOutWithMock(scheduler, 'wallclock')
- stack = parser.Stack(None, 's', parser.Template({}))
+ stack = parser.Stack(self.ctx, 's', parser.Template({}))
def dummy_task():
while True:
self.m.VerifyAll()
def test_stack_name_valid(self):
- stack = parser.Stack(None, 's', parser.Template({}))
- stack = parser.Stack(None, 'stack123', parser.Template({}))
- stack = parser.Stack(None, 'test.stack', parser.Template({}))
- stack = parser.Stack(None, 'test_stack', parser.Template({}))
- stack = parser.Stack(None, 'TEST', parser.Template({}))
- stack = parser.Stack(None, 'test-stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 's', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'stack123', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test.stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'TEST', parser.Template({}))
+ stack = parser.Stack(self.ctx, 'test-stack', parser.Template({}))
def test_stack_name_invalid(self):
- self.assertRaises(ValueError, parser.Stack, None, '_foo',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '_foo',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '1bad',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '1bad',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '.kcats',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '.kcats',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, ' teststack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, ' teststack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '^-^',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '^-^',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '\"stack\"',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '\"stack\"',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '1234',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '1234',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'cat|dog',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'cat|dog',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '$(foo)',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '$(foo)',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test/stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test/stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test\stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test\stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test::stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test::stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test;stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test;stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, 'test~stack',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, 'test~stack',
parser.Template({}))
- self.assertRaises(ValueError, parser.Stack, None, '#test',
+ self.assertRaises(ValueError, parser.Stack, self.ctx, '#test',
parser.Template({}))
@stack_delete_after
from heat.tests import generic_resource as generic_rsrc
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
def test_to_parameters(self):
"""Tests property conversion to parameter values."""
setup_dummy_db()
- stack = parser.Stack(None, 'test_stack', parser.Template({}),
+ stack = parser.Stack(dummy_context(), 'test_stack',
+ parser.Template({}),
stack_id=uuidutils.generate_uuid())
class DummyResource(object):
import itertools
from eventlet.support import greenlets as greenlet
-from heat.common import context
from heat.common import exception
from heat.engine import parser
from heat.engine import resource
from heat.tests import generic_resource as generic_rsrc
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
def setUp(self):
super(ResourceTest, self).setUp()
setup_dummy_db()
- self.stack = parser.Stack(None, 'test_stack', parser.Template({}),
+ self.stack = parser.Stack(dummy_context(), 'test_stack',
+ parser.Template({}),
stack_id=uuidutils.generate_uuid())
resource._register_class('GenericResourceType',
tmpl1 = {'Type': 'Foo'}
tmpl2 = {'Type': 'Foo'}
tmpl3 = {'Type': 'Bar'}
- stack2 = parser.Stack(None, 'test_stack', parser.Template({}),
- stack_id=-1)
+ stack2 = parser.Stack(dummy_context(), 'test_stack',
+ parser.Template({}), stack_id=-1)
res1 = generic_rsrc.GenericResource('test_resource', tmpl1, self.stack)
res2 = generic_rsrc.GenericResource('test_resource', tmpl2, stack2)
res3 = generic_rsrc.GenericResource('test_resource2', tmpl3, stack2)
'Metadata': {'Test': 'Initial metadata'}
}
setup_dummy_db()
- ctx = context.get_admin_context()
- self.m.StubOutWithMock(ctx, 'username')
- ctx.username = 'metadata_test_user'
- self.stack = parser.Stack(ctx, 'test_stack', parser.Template({}))
+ self.stack = parser.Stack(dummy_context(),
+ 'test_stack', parser.Template({}))
self.stack.store()
self.res = generic_rsrc.GenericResource('metadata_resource',
tmpl, self.stack)
import stubout
import testtools
-from heat.common import context
from heat.common import identifier
from heat.rpc import api as rpc_api
from heat.rpc import client as rpc_client
from heat.openstack.common import rpc
+from heat.tests.utils import dummy_context
class EngineRpcAPITestCase(testtools.TestCase):
def setUp(self):
- self.context = context.get_admin_context()
+ self.context = dummy_context()
cfg.CONF.set_default('rpc_backend',
'heat.openstack.common.rpc.impl_fake')
cfg.CONF.set_default('verbose', True)
super(EngineRpcAPITestCase, self).setUp()
def _test_engine_api(self, method, rpc_method, **kwargs):
- ctxt = context.RequestContext('fake_user', 'fake_project')
+ ctxt = dummy_context()
if 'rpcapi_class' in kwargs:
rpcapi_class = kwargs['rpcapi_class']
del kwargs['rpcapi_class']
import collections
from heat.engine import clients
-from heat.common import context
from heat.common import template_format
from heat.engine import parser
from heat.engine import resource
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
from heat.tests.v1_1 import fakes
from heat.tests import utils
return self.stack
def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
stack_name = 'test_stack'
tmpl = parser.Template(t)
- stack = parser.Stack(ctx, stack_name, tmpl)
+ stack = parser.Stack(dummy_context(), stack_name, tmpl)
stack.store()
return stack
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
stack_name = 'tag_test'
t = template_format.parse(instance_template)
template = parser.Template(t)
- stack = parser.Stack(None, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
environment.Environment({'KeyName': 'test'}),
stack_id=uuidutils.generate_uuid())
stack_name = 'tag_test'
t = template_format.parse(group_template)
template = parser.Template(t)
- stack = parser.Stack(None, stack_name, template,
+ stack = parser.Stack(dummy_context(), stack_name, template,
environment.Environment({'KeyName': 'test'}),
stack_id=uuidutils.generate_uuid())
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import dummy_context
wp_template = '''
def _setup_test_stack(self, stack_name):
t = template_format.parse(wp_template)
template = parser.Template(t)
- stack = parser.Stack(None, stack_name, template,
+ ctx = dummy_context()
+ stack = parser.Stack(ctx, stack_name, template,
environment.Environment({'KeyName': 'test'}),
stack_id=uuidutils.generate_uuid())
return (t, stack)
# under the License.
from heat.common import template_format
-from heat.common import context
from heat.common import exception
from heat.engine import environment
from heat.engine import parser
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests import generic_resource as generic_rsrc
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import stack_delete_after
MyStackResource)
t = parser.Template({template.RESOURCES:
{"provider_resource": ws_res_snippet}})
- self.parent_stack = parser.Stack(None, 'test_stack', t,
+ self.parent_stack = parser.Stack(dummy_context(), 'test_stack', t,
stack_id=uuidutils.generate_uuid())
self.parent_resource = MyStackResource('test',
ws_res_snippet,
self.parent_stack)
- self.parent_resource.context = context.get_admin_context()
self.templ = template_format.parse(wp_template)
@stack_delete_after
from heat.engine import clients
from heat.engine import environment
from heat.tests.v1_1 import fakes
-from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.engine import resources
import heat.db.api as db_api
from heat.engine import parser
from heat.tests.common import HeatTestCase
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
test_template_volumeattach = '''
self.fc = fakes.FakeClient()
resources.initialise()
setup_dummy_db()
- self.ctx = context.get_admin_context()
+ self.ctx = dummy_context()
def test_validate_volumeattach_valid(self):
t = template_format.parse(test_template_volumeattach % 'vdq')
- stack = parser.Stack(None, 'test_stack', parser.Template(t))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template(t))
self.m.StubOutWithMock(db_api, 'resource_get_by_name_and_stack')
db_api.resource_get_by_name_and_stack(None, 'test_resource_name',
def test_validate_volumeattach_invalid(self):
t = template_format.parse(test_template_volumeattach % 'sda')
- stack = parser.Stack(None, 'test_stack', parser.Template(t))
+ stack = parser.Stack(self.ctx, 'test_stack', parser.Template(t))
self.m.StubOutWithMock(db_api, 'resource_get_by_name_and_stack')
db_api.resource_get_by_name_and_stack(None, 'test_resource_name',
t = template_format.parse(test_unregistered_key)
template = parser.Template(t)
params = {'KeyName': 'not_registered'}
- stack = parser.Stack(None, 'test_stack', template,
+ stack = parser.Stack(self.ctx, 'test_stack', template,
environment.Environment(params))
self.m.StubOutWithMock(instances.Instance, 'nova')
def test_invalid_security_groups_with_nics(self):
t = template_format.parse(test_template_invalid_secgroups)
template = parser.Template(t)
- stack = parser.Stack(None, 'test_stack', template,
+ stack = parser.Stack(self.ctx, 'test_stack', template,
environment.Environment({'KeyName': 'test'}))
self.m.StubOutWithMock(instances.Instance, 'nova')
def test_invalid_security_group_ids_with_nics(self):
t = template_format.parse(test_template_invalid_secgroupids)
template = parser.Template(t)
- stack = parser.Stack(None, 'test_stack', template,
+ stack = parser.Stack(self.ctx, 'test_stack', template,
environment.Environment({'KeyName': 'test'}))
self.m.StubOutWithMock(instances.Instance, 'nova')
from testtools import skipIf
-from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.engine import parser
from heat.engine import resource
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
try:
return stack
def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
stack_name = 'test_stack'
tmpl = parser.Template(t)
- stack = parser.Stack(ctx, stack_name, tmpl)
+ stack = parser.Stack(dummy_context(), stack_name, tmpl)
stack.store()
return stack
from heat.tests.common import HeatTestCase
from heat.tests import fakes
-from heat.tests.utils import stack_delete_after
+from heat.tests.utils import dummy_context
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import stack_delete_after
import heat.db.api as db_api
from heat.common import template_format
from heat.engine import resource
from heat.engine import scheduler
from heat.engine.resources import wait_condition as wc
-from heat.common import context
test_template_waitcondition = '''
{
stub=True):
temp = template_format.parse(template)
template = parser.Template(temp)
- ctx = context.get_admin_context()
- ctx.tenant_id = 'test_tenant'
+ ctx = dummy_context(tenant_id='test_tenant')
stack = parser.Stack(ctx, stack_name, template,
environment.Environment(params),
disable_rollback=True)
def create_stack(self, stack_name='test_stack2', params={}):
temp = template_format.parse(test_template_waitcondition)
template = parser.Template(temp)
- ctx = context.get_admin_context()
- ctx.tenant_id = 'test_tenant'
+ ctx = dummy_context(tenant_id='test_tenant')
stack = parser.Stack(ctx, stack_name, template,
environment.Environment(params),
disable_rollback=True)
import datetime
import mox
-from heat.common import context
import heat.db.api as db_api
from heat.common import exception
from heat.engine import parser
from heat.tests.common import HeatTestCase
from heat.tests import utils
+from heat.tests.utils import dummy_context
class WatchData:
# Create a dummy stack in the DB as WatchRule instances
# must be associated with a stack
utils.setup_dummy_db()
- ctx = context.get_admin_context()
- ctx.username = 'dummyuser'
- ctx.tenant_id = '123456'
+ ctx = dummy_context()
+ ctx.auth_token = 'abcd1234'
empty_tmpl = {"template": {}}
tmpl = parser.Template(empty_tmpl)
stack_name = 'dummystack'
self.setUpDatabase()
self.username = 'watchrule_test_user'
- self.ctx = context.get_admin_context()
- self.ctx.username = self.username
- self.ctx.tenant_id = u'123456'
+ self.ctx = dummy_context()
+ self.ctx.auth_token = 'abcd1234'
self.m.ReplayAll()