--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+import fixtures
+import mox
+import testtools
+
+
+class HeatTestCase(testtools.TestCase):
+
+ def setUp(self):
+ super(HeatTestCase, self).setUp()
+ self.m = mox.Mox()
+ self.addCleanup(self.m.UnsetStubs)
+ self.useFixture(fixtures.FakeLogger())
# under the License.
-import unittest
+from heat.tests.common import HeatTestCase
from nose.plugins.attrib import attr
from heat.api.aws import utils as api_utils
@attr(tag=['unit', 'api-aws', 'AWSCommon'])
@attr(speed='fast')
-class AWSCommon(unittest.TestCase):
+class AWSCommon(HeatTestCase):
'''
Tests the api/aws common componenents
'''
expected = {"bar": 123}
result = api_utils.reformat_dict_keys(keymap, data)
self.assertEqual(result, expected)
-
- def setUp(self):
- print "setup complete"
-
- def tearDown(self):
- print "teardown complete"
import json
import os
-import unittest
-import mox
from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.rpc import api as rpc_api
from heat.api.aws import exception
import heat.api.cfn.v1.stacks as stacks
+from heat.tests.common import HeatTestCase
+
+policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
@attr(tag=['unit', 'api-cfn-v1-stacks', 'StackController'])
@attr(speed='fast')
-class CfnStackControllerTest(unittest.TestCase):
+class CfnStackControllerTest(HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
params = {'Action': 'ListStacks'}
dummy_req = self._dummy_GET_request(params)
dummy_req.context.roles = ['heat_stack_user']
- self.controller.policy.policy_path = (self.policy_path +
+ self.controller.policy.policy_path = (policy_path +
'deny_stack_user.json')
self.assertRaises(exception.HeatAccessDeniedError,
self.controller._enforce, dummy_req, 'ListStacks')
).AndRaise(AttributeError)
self.m.ReplayAll()
- self.controller.policy.policy_path = (self.policy_path +
+ self.controller.policy.policy_path = (policy_path +
'deny_stack_user.json')
self.assertRaises(exception.HeatInternalFailureError,
self.controller._enforce, dummy_req, 'ListStacks')
self.m.VerifyAll()
def setUp(self):
- self.maxDiff = None
- self.m = mox.Mox()
+ super(CfnStackControllerTest, self).setUp()
- self.path = os.path.dirname(os.path.realpath(__file__))
- self.policy_path = self.path + "/policy/"
opts = [
- cfg.StrOpt('config_dir', default=self.policy_path),
+ cfg.StrOpt('config_dir', default=policy_path),
cfg.StrOpt('config_file', default='foo'),
cfg.StrOpt('project', default='heat'),
]
bind_port = 8000
cfgopts = DummyConfig()
self.controller = stacks.StackController(options=cfgopts)
- print "setup complete"
-
- def tearDown(self):
- self.m.UnsetStubs()
- print "teardown complete"
# under the License.
import os
-import unittest
-import mox
from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.common.wsgi import Request
from heat.api.aws import exception
import heat.api.cloudwatch.watch as watches
-from heat.engine import api as engine_api
-from heat.rpc import api as rpc_api
+from heat.rpc import api as engine_api
+from heat.tests.common import HeatTestCase
@attr(tag=['unit', 'api-cloudwatch', 'WatchController'])
@attr(speed='fast')
-class WatchControllerTest(unittest.TestCase):
+class WatchControllerTest(HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
self.assert_(type(result) == exception.HeatInvalidParameterValueError)
def setUp(self):
- self.maxDiff = None
- self.m = mox.Mox()
-
+ super(WatchControllerTest, self).setUp()
self.path = os.path.dirname(os.path.realpath(__file__))
self.policy_path = self.path + "/policy/"
opts = [
]
cfg.CONF.register_opts(opts)
cfg.CONF.set_default('host', 'host')
- self.topic = rpc_api.ENGINE_TOPIC
+ self.topic = engine_api.ENGINE_TOPIC
self.api_version = '1.0'
# Create WSGI controller instance
cfgopts = DummyConfig()
self.controller = watches.WatchController(options=cfgopts)
self.controller.policy.policy_path = None
- print "setup complete"
def tearDown(self):
- self.m.UnsetStubs()
self.m.VerifyAll()
- print "teardown complete"
+ super(WatchControllerTest, self).tearDown()
from heat.common import policy
from heat.common import exception
+policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
+
@attr(tag=['unit', 'common-policy', 'Enforcer'])
@attr(speed='fast')
"PutMetricAlarm", "PutMetricData", "SetAlarmState")
def setUp(self):
- self.path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
self.m = mox.Mox()
opts = [
- cfg.StrOpt('config_dir', default=self.path),
+ cfg.StrOpt('config_dir', default=policy_path),
cfg.StrOpt('config_file', default='foo'),
cfg.StrOpt('project', default='heat'),
]
cfg.CONF.register_opts(opts)
- print "setup complete"
def tearDown(self):
self.m.UnsetStubs()
- print "teardown complete"
def test_policy_cfn_default(self):
enforcer = policy.Enforcer(scope='cloudformation')
enforcer.enforce(ctx, action, {})
def test_policy_cfn_notallowed(self):
- pf = self.path + 'notallowed.json'
+ pf = policy_path + 'notallowed.json'
self.m.StubOutWithMock(policy.Enforcer, '_find_policy_file')
policy.Enforcer._find_policy_file().MultipleTimes().AndReturn(pf)
self.m.ReplayAll()
self.m.VerifyAll()
def test_policy_cfn_deny_stack_user(self):
- pf = self.path + 'deny_stack_user.json'
+ pf = policy_path + 'deny_stack_user.json'
self.m.StubOutWithMock(policy.Enforcer, '_find_policy_file')
policy.Enforcer._find_policy_file().MultipleTimes().AndReturn(pf)
self.m.ReplayAll()
self.m.VerifyAll()
def test_policy_cfn_allow_non_stack_user(self):
- pf = self.path + 'deny_stack_user.json'
+ pf = policy_path + 'deny_stack_user.json'
self.m.StubOutWithMock(policy.Enforcer, '_find_policy_file')
policy.Enforcer._find_policy_file().MultipleTimes().AndReturn(pf)
self.m.ReplayAll()
self.m.VerifyAll()
def test_policy_cw_deny_stack_user(self):
- pf = self.path + 'deny_stack_user.json'
+ pf = policy_path + 'deny_stack_user.json'
self.m.StubOutWithMock(policy.Enforcer, '_find_policy_file')
policy.Enforcer._find_policy_file().MultipleTimes().AndReturn(pf)
self.m.ReplayAll()
self.m.VerifyAll()
def test_policy_cw_allow_non_stack_user(self):
- pf = self.path + 'deny_stack_user.json'
+ pf = policy_path + 'deny_stack_user.json'
self.m.StubOutWithMock(policy.Enforcer, '_find_policy_file')
policy.Enforcer._find_policy_file().MultipleTimes().AndReturn(pf)
self.m.ReplayAll()
import os
-import unittest
import mox
from nose.plugins.attrib import attr
from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import dbinstance as dbi
+from heat.tests.common import HeatTestCase
@attr(tag=['unit', 'resource'])
@attr(speed='fast')
-class DBInstanceTest(unittest.TestCase):
+class DBInstanceTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(DBInstanceTest, self).setUp()
self.m.StubOutWithMock(dbi.DBInstance, 'create_with_template')
self.m.StubOutWithMock(dbi.DBInstance, 'nested')
- def tearDown(self):
- self.m.UnsetStubs()
- print "DBInstanceTest teardown complete"
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# under the License.
+import mox
import re
import os
-import unittest
-import mox
-
from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.engine.resources import loadbalancer as lb
from heat.engine.resources import wait_condition as wc
from heat.engine.resource import Metadata
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.v1_1 import fakes
from heat.tests import fakes as test_fakes
@attr(tag=['unit', 'resource'])
@attr(speed='fast')
-class LoadBalancerTest(unittest.TestCase):
+class LoadBalancerTest(HeatTestCase):
def setUp(self):
+ super(LoadBalancerTest, self).setUp()
config.register_engine_opts()
- self.m = mox.Mox()
self.fc = fakes.FakeClient()
self.m.StubOutWithMock(lb.LoadBalancer, 'nova')
self.m.StubOutWithMock(instance.Instance, 'nova')
'http://127.0.0.1:8000/v1/waitcondition')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
- print "LoadBalancerTest teardown complete"
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# under the License.
-import unittest
-import mox
-
from nose.plugins.attrib import attr
from heat.common import context
from heat.engine import parser
from heat.engine.resources import stack as nested_stack
from heat.common import urlfetch
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
@attr(tag=['unit', 'resource'])
@attr(speed='fast')
-class NestedStackTest(unittest.TestCase):
+class NestedStackTest(HeatTestCase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
'''
def setUp(self):
- self.m = mox.Mox()
+ super(NestedStackTest, self).setUp()
self.m.StubOutWithMock(urlfetch, 'get')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
- print "NestedStackTest teardown complete"
-
def create_stack(self, template):
t = template_format.parse(template)
stack = self.parse_stack(t)
import os
import eventlet
-import mox
-import unittest
from nose.plugins.attrib import attr
from heat.engine import scheduler
from heat.engine.resources import volume as vol
from heat.engine import clients
+from heat.tests.common import HeatTestCase
from heat.tests.v1_1 import fakes
from heat.tests.utils import setup_dummy_db
@attr(tag=['unit', 'resource', 'volume'])
@attr(speed='fast')
-class VolumeTest(unittest.TestCase):
+class VolumeTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(VolumeTest, self).setUp()
self.fc = fakes.FakeClient()
self.m.StubOutWithMock(clients.OpenStackClients, 'cinder')
self.m.StubOutWithMock(clients.OpenStackClients, 'nova')
self.m.StubOutWithMock(eventlet, 'sleep')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
- print "VolumeTest teardown complete"
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
to ensure tests clean up their stacks regardless of test success/failure
"""
def wrapped_test(test_cls):
- #print "Running test", test_fn.__name__
try:
test_fn(test_cls)
finally:
try:
test_cls.stack.delete()
except AttributeError:
- print "Could not delete stack (already deleted?)"
- #print "Exited", test_fn.__name__
+ pass
return wrapped_test