Some simple cases that do not use mox still use unittest directly.
Replacing unittest.TestCase with HeatTestCase has the added benefit of
capturing all logging for reporting during failure.
Change-Id: I76192bdb91822055f1e2f7d2e806bf094d399304
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(HeatTestCase):
'''
Tests the api/aws common componenents
import json
import os
-from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.common import context
policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
-@attr(tag=['unit', 'api-cfn-v1-stacks', 'StackController'])
-@attr(speed='fast')
class CfnStackControllerTest(HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
import os
-from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.common import context
from heat.tests.common import HeatTestCase
-@attr(tag=['unit', 'api-cloudwatch', 'WatchController'])
-@attr(speed='fast')
class WatchControllerTest(HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
# under the License.
import json
-import unittest
-import mox
-from nose.plugins.attrib import attr
from oslo.config import cfg
import webob.exc
from heat.common.wsgi import Request
from heat.common import urlfetch
from heat.rpc import api as rpc_api
+from heat.tests.common import HeatTestCase
import heat.api.openstack.v1 as api_v1
import heat.api.openstack.v1.stacks as stacks
import heat.api.openstack.v1.events as events
-@attr(tag=['unit', 'api-openstack-v1'])
-@attr(speed='fast')
-class InstantiationDataTest(unittest.TestCase):
-
- def setUp(self):
- self.m = mox.Mox()
-
- def tearDown(self):
- self.m.UnsetStubs()
+class InstantiationDataTest(HeatTestCase):
def test_format_parse(self):
data = {"key1": ["val1[0]", "val1[1]"], "key2": "val2"}
def __init__(self, *args, **kwargs):
super(ControllerTest, self).__init__(*args, **kwargs)
- self.maxDiff = None
- self.m = mox.Mox()
-
cfg.CONF.set_default('host', 'host')
self.topic = rpc_api.ENGINE_TOPIC
self.api_version = '1.0'
self.tenant = 't'
- def tearDown(self):
- self.m.UnsetStubs()
-
def _create_context(self, user='api_test_user'):
ctx = context.get_admin_context()
self.m.StubOutWithMock(ctx, 'username')
return 'http://%s%s' % (host, path)
-@attr(tag=['unit', 'api-openstack-v1', 'StackController'])
-@attr(speed='fast')
-class StackControllerTest(ControllerTest, unittest.TestCase):
+class StackControllerTest(ControllerTest, HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
+ super(StackControllerTest, self).setUp()
# Create WSGI controller instance
+
class DummyConfig():
bind_port = 8004
+
cfgopts = DummyConfig()
self.controller = stacks.StackController(options=cfgopts)
self.m.VerifyAll()
-@attr(tag=['unit', 'api-openstack-v1', 'ResourceController'])
-@attr(speed='fast')
-class ResourceControllerTest(ControllerTest, unittest.TestCase):
+class ResourceControllerTest(ControllerTest, HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
+ super(ResourceControllerTest, self).setUp()
# Create WSGI controller instance
+
class DummyConfig():
bind_port = 8004
+
cfgopts = DummyConfig()
self.controller = resources.ResourceController(options=cfgopts)
self.m.VerifyAll()
-@attr(tag=['unit', 'api-openstack-v1', 'EventController'])
-@attr(speed='fast')
-class EventControllerTest(ControllerTest, unittest.TestCase):
+class EventControllerTest(ControllerTest, HeatTestCase):
'''
Tests the API class which acts as the WSGI controller,
the endpoint processing API requests after they are routed
'''
def setUp(self):
+ super(EventControllerTest, self).setUp()
# Create WSGI controller instance
+
class DummyConfig():
bind_port = 8004
+
cfgopts = DummyConfig()
self.controller = events.EventController(options=cfgopts)
self.m.VerifyAll()
-class RoutesTest(unittest.TestCase):
+class RoutesTest(HeatTestCase):
def assertRoute(self, mapper, path, method, action, controller, params={}):
route = mapper.match(path, {'REQUEST_METHOD': method})
self.assertEqual(params, route)
def setUp(self):
+ super(RoutesTest, self).setUp()
self.m = api_v1.API({}).map
def test_template_handling(self):
import copy
import eventlet
-import unittest
import mox
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import template_format
from heat.engine.resources import autoscaling as asc
from heat.engine import scheduler
from heat.engine.resource import Metadata
from heat.openstack.common import timeutils
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class AutoScalingTest(unittest.TestCase):
+class AutoScalingTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(AutoScalingTest, self).setUp()
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# under the License.
-from nose.plugins.attrib import attr
import unittest
import heat
import os
basepath = os.path.join(heat.__path__[0], os.path.pardir)
-@attr(tag=['unit', 'cli'])
-@attr(speed='medium')
class CliTest(unittest.TestCase):
def test_bins(self):
import os.path
-import mox
-from nose.plugins.attrib import attr
from oslo.config import cfg
-import unittest
from heat.common import context
from heat.common import policy
from heat.common import exception
+from heat.tests.common import HeatTestCase
policy_path = os.path.dirname(os.path.realpath(__file__)) + "/policy/"
-@attr(tag=['unit', 'common-policy', 'Enforcer'])
-@attr(speed='fast')
-class TestPolicyEnforcer(unittest.TestCase):
+class TestPolicyEnforcer(HeatTestCase):
cfn_actions = ("ListStacks", "CreateStack", "DescribeStacks",
"DeleteStack", "UpdateStack", "DescribeStackEvents",
"ValidateTemplate", "GetTemplate",
"PutMetricAlarm", "PutMetricData", "SetAlarmState")
def setUp(self):
- self.m = mox.Mox()
+ super(TestPolicyEnforcer, self).setUp()
opts = [
cfg.StrOpt('config_dir', default=policy_path),
cfg.StrOpt('config_file', default='foo'),
]
cfg.CONF.register_opts(opts)
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_policy_cfn_default(self):
enforcer = policy.Enforcer(scope='cloudformation')
import mox
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.tests.common import HeatTestCase
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class DBInstanceTest(HeatTestCase):
def setUp(self):
super(DBInstanceTest, self).setUp()
import unittest
-from nose.plugins.attrib import attr
from heat.engine.dependencies import Dependencies
from heat.engine.dependencies import CircularDependencyException
-@attr(tag=['unit', 'dependencies'])
-@attr(speed='fast')
class dependenciesTest(unittest.TestCase):
def _dep_test(self, func, checkorder, deps):
import os
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import template_format
from heat.engine.resources import eip
from heat.engine import parser
from heat.engine import scheduler
+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'])
-@attr(speed='fast')
-class EIPTest(unittest.TestCase):
+class EIPTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(EIPTest, self).setUp()
self.fc = fakes.FakeClient()
self.m.StubOutWithMock(eip.ElasticIp, 'nova')
self.m.StubOutWithMock(eip.ElasticIpAssociation, 'nova')
self.m.StubOutWithMock(self.fc.servers, 'get')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
import unittest
-from nose.plugins.attrib import attr
import heat.engine.api as api
-@attr(tag=['unit', 'engine-api'])
-@attr(speed='fast')
class EngineApiTest(unittest.TestCase):
def test_timeout_extract(self):
p = {'timeout_mins': '5'}
import os
-import unittest
import json
import mox
-from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.common import context
from heat.engine.resources import instance as instances
from heat.engine import watchrule
from heat.openstack.common import threadgroup
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
pass
-@attr(tag=['unit', 'stack'])
-@attr(speed='slow')
-class stackCreateTest(unittest.TestCase):
+class stackCreateTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(stackCreateTest, self).setUp()
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_wordpress_single_instance_stack_create(self):
stack = get_wordpress_stack('test_stack', create_context(self.m))
setup_mocks(self.m, stack)
self.assertEqual(db_s.status, 'DELETE_COMPLETE')
-@attr(tag=['unit', 'engine-api', 'engine-service'])
-@attr(speed='fast')
-class stackServiceCreateUpdateDeleteTest(unittest.TestCase):
+class stackServiceCreateUpdateDeleteTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(stackServiceCreateUpdateDeleteTest, self).setUp()
self.username = 'stack_service_create_test_user'
self.tenant = 'stack_service_create_test_tenant'
setup_dummy_db()
self.man = service.EngineService('a-host', 'a-topic')
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_stack_create(self):
stack_name = 'service_create_test_stack'
params = {'foo': 'bar'}
self.m.VerifyAll()
-@attr(tag=['unit', 'engine-api', 'engine-service'])
-@attr(speed='fast')
-class stackServiceTest(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
+class stackServiceTestBase(HeatTestCase):
+
+ tenant = 'stack_service_test_tenant'
+
+ def tearDown(self):
+ super(stackServiceTestBase, self).tearDown()
+ # testtools runs cleanups *after* tearDown, but we need to mock some
+ # things now.
+ self.m.UnsetStubs()
+
+ m = mox.Mox()
+ create_context(m, self.username, self.tenant, ctx=self.stack.context)
+ fc = setup_mocks(m, self.stack)
+ m.StubOutWithMock(fc.client, 'get_servers_9999')
+ get = fc.client.get_servers_9999
+ get().AndRaise(service.clients.novaclient.exceptions.NotFound(404))
+ m.ReplayAll()
+
+ self.stack.delete()
+
+ m.UnsetStubs()
+
+ def setUp(self):
+ setup_dummy_db()
m = mox.Mox()
- cls.username = 'stack_service_test_user'
- cls.tenant = 'stack_service_test_tenant'
- ctx = create_context(m, cls.username, cls.tenant)
- cls.stack_name = 'service_test_stack'
+ self.username = 'stack_service_test_user'
+ ctx = create_context(m, self.username, self.tenant)
+ self.stack_name = 'service_test_stack'
cfg.CONF.set_default('heat_stack_user_role', 'stack_user_role')
- stack = get_wordpress_stack(cls.stack_name, ctx)
+ stack = get_wordpress_stack(self.stack_name, ctx)
setup_mocks(m, stack)
m.ReplayAll()
stack.store()
stack.create()
- cls.stack = stack
- cls.stack_identity = stack.identifier()
+ self.stack = stack
+ self.stack_identity = stack.identifier()
m.UnsetStubs()
- @classmethod
- def tearDownClass(cls):
- cls = cls
- m = mox.Mox()
- create_context(m, cls.username, cls.tenant, ctx=cls.stack.context)
- fc = setup_mocks(m, cls.stack)
- m.StubOutWithMock(fc.client, 'get_servers_9999')
- get = fc.client.get_servers_9999
- get().AndRaise(service.clients.novaclient.exceptions.NotFound(404))
- m.ReplayAll()
+ super(stackServiceTestBase, self).setUp()
+ self.m.UnsetStubs()
+ self.ctx = create_context(self.m, self.username, self.tenant)
+ setup_mocks(self.m, self.stack)
- cls.stack.delete()
- m.UnsetStubs()
+class stackServiceTest(stackServiceTestBase):
def setUp(self):
- self.m = mox.Mox()
- setup_dummy_db()
- self.ctx = create_context(self.m, self.username, self.tenant)
- setup_mocks(self.m, self.stack)
+ super(stackServiceTest, self).setUp()
self.m.ReplayAll()
self.man = service.EngineService('a-host', 'a-topic')
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_stack_identify(self):
identity = self.man.identify_stack(self.ctx, self.stack_name)
self.assertEqual(identity, self.stack_identity)
self.assertTrue('description' in s)
self.assertNotEqual(s['description'].find('WordPress'), -1)
- def test_stack_list_all_empty(self):
- self.tearDown()
- self.tenant = 'stack_list_all_empty_tenant'
- self.setUp()
-
- sl = self.man.list_stacks(self.ctx)
-
- self.assertEqual(len(sl), 0)
-
def test_stack_describe_nonexistent(self):
nonexist = dict(self.stack_identity)
nonexist['stack_name'] = 'wibble'
self.assertNotEqual(s['description'].find('WordPress'), -1)
self.assertTrue('parameters' in s)
- def test_stack_describe_all_empty(self):
- self.tearDown()
- self.tenant = 'stack_describe_all_empty_tenant'
- self.setUp()
-
- sl = self.man.show_stack(self.ctx, None)
-
- self.assertEqual(len(sl), 0)
-
def test_list_resource_types(self):
resources = self.man.list_resource_types(self.ctx)
self.assertTrue(isinstance(resources, list))
self.assertRaises(exception.WatchRuleNotFound,
self.man.set_watch_state,
self.ctx, watch_name="nonexistent", state=state)
+
+
+class stackServiceTestEmpty(stackServiceTestBase):
+
+ def setUp(self):
+ super(stackServiceTestEmpty, self).setUp()
+
+ # Change to a new, empty tenant context
+ self.ctx = create_context(self.m, self.username,
+ 'stack_list_all_empty_tenant')
+ self.m.ReplayAll()
+
+ self.man = service.EngineService('a-host', 'a-topic')
+
+ def test_stack_list_all_empty(self):
+ sl = self.man.list_stacks(self.ctx)
+
+ self.assertEqual(len(sl), 0)
+
+ def test_stack_describe_all_empty(self):
+ sl = self.man.show_stack(self.ctx, None)
+
+ self.assertEqual(len(sl), 0)
# under the License.
-from nose.plugins.attrib import attr
-import mox
-import unittest
-
from heat.common import context
import heat.db.api as db_api
from heat.engine import parser
from heat.engine import template
from heat.engine import event
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests import generic_resource as generic_rsrc
}
-@attr(tag=['unit', 'event'])
-@attr(speed='fast')
-class EventTest(unittest.TestCase):
+class EventTest(HeatTestCase):
def setUp(self):
+ super(EventTest, self).setUp()
self.username = 'event_test_user'
- self.m = mox.Mox()
-
setup_dummy_db()
self.ctx = context.get_admin_context()
self.m.StubOutWithMock(self.ctx, 'username')
def tearDown(self):
db_api.stack_delete(self.ctx, self.stack.id)
- self.m.UnsetStubs()
+ super(EventTest, self).tearDown()
def test_load(self):
self.resource.resource_id_set('resource_physical_id')
import unittest
-from nose.plugins.attrib import attr
from heat.common import identifier
-@attr(tag=['unit', 'identifier'])
-@attr(speed='fast')
class IdentifierTest(unittest.TestCase):
url_prefix = 'http://1.2.3.4/foo/'
self.assertEqual(hi._path_components(), ['p1', 'p2', 'p3'])
-@attr(tag=['unit', 'identifier'])
-@attr(speed='fast')
class ResourceIdentifierTest(unittest.TestCase):
def test_resource_init_no_path(self):
si = identifier.HeatIdentifier('t', 's', 'i')
't', 's', 'i', 'p', 'r/r')
-@attr(tag=['unit', 'identifier'])
-@attr(speed='fast')
class EventIdentifierTest(unittest.TestCase):
def test_event_init(self):
si = identifier.HeatIdentifier('t', 's', 'i')
import os
import copy
-import unittest
import mox
-from nose.plugins.attrib import attr
-
from heat.tests.v1_1 import fakes
from heat.engine.resources import instance as instances
from heat.common import template_format
from heat.engine import parser
from heat.engine import scheduler
from heat.openstack.common import uuidutils
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource', 'instance'])
-@attr(speed='fast')
-class instancesTest(unittest.TestCase):
+class instancesTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(instancesTest, self).setUp()
self.fc = fakes.FakeClient()
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_instance_create(self):
f = open("%s/WordPress_Single_Instance_gold.template" % self.path)
t = template_format.parse(f.read())
import os
import eventlet
-import unittest
import mox
-from nose.plugins.attrib import attr
-
from heat.tests.v1_1 import fakes
from heat.common import context
from heat.common import exception
from heat.engine.resources import loadbalancer
from heat.engine import parser
from heat.engine import scheduler
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class InstanceGroupTest(unittest.TestCase):
+class InstanceGroupTest(HeatTestCase):
def setUp(self):
+ super(InstanceGroupTest, self).setUp()
self.fc = fakes.FakeClient()
- self.m = mox.Mox()
self.m.StubOutWithMock(loadbalancer.LoadBalancer, 'reload')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
import re
import os
-from nose.plugins.attrib import attr
-
from oslo.config import cfg
from heat.common import exception
from heat.common import config
return ctx
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class LoadBalancerTest(HeatTestCase):
def setUp(self):
super(LoadBalancerTest, self).setUp()
# under the License.
import errno
-import mox
import os
import pkg_resources
import subprocess
-import unittest
import stat
import StringIO
-from nose.plugins.attrib import attr
-
from heat.cloudinit import loguserdata
+from heat.tests.common import HeatTestCase
class FakeCiVersion():
pass
-@attr(tag=['unit'])
-@attr(speed='fast')
-class LoguserdataTest(unittest.TestCase):
+class LoguserdataTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(LoguserdataTest, self).setUp()
self.m.StubOutWithMock(pkg_resources, 'get_distribution')
self.m.StubOutWithMock(subprocess, 'Popen')
self.m.StubOutWithMock(os, 'chmod')
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_ci_version(self):
# too old versions
pkg_resources.get_distribution('cloud-init').AndReturn(
# under the License.
-import mox
-
import eventlet
-import unittest
-from nose.plugins.attrib import attr
+import mox
from oslo.config import cfg
from heat.tests import fakes
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import stack_delete_after
'''
-@attr(tag=['unit', 'resource', 'Metadata'])
-@attr(speed='slow')
-class MetadataRefreshTest(unittest.TestCase):
+class MetadataRefreshTest(HeatTestCase):
'''
The point of the test is to confirm that metadata gets updated
when FnGetAtt() returns something different.
gets called.
'''
def setUp(self):
- self.m = mox.Mox()
+ super(MetadataRefreshTest, self).setUp()
self.m.StubOutWithMock(eventlet, 'sleep')
self.fc = fakes.FakeKeystoneClient()
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
# Note tests creating a stack should be decorated with @stack_delete_after
# to ensure the stack is properly cleaned up
def create_stack(self, stack_name='test_stack', params={}):
self.m.VerifyAll()
-@attr(tag=['unit', 'resource', 'Metadata'])
-@attr(speed='slow')
-class WaitCondMetadataUpdateTest(unittest.TestCase):
+class WaitCondMetadataUpdateTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(WaitCondMetadataUpdateTest, self).setUp()
setup_dummy_db()
self.ctx = context.get_admin_context()
self.ctx.tenant_id = 'test_tenant'
cfg.CONF.set_default('heat_waitcondition_server_url',
'http://127.0.0.1:8000/v1/waitcondition')
- def tearDown(self):
- self.m.UnsetStubs()
-
# Note tests creating a stack should be decorated with @stack_delete_after
# to ensure the stack is properly cleaned up
def create_stack(self, stack_name='test_stack'):
# under the License.
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class NestedStackTest(HeatTestCase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
import os
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
-
from heat.tests.v1_1 import fakes
from heat.engine.resources import instance as instances
from heat.common import template_format
from heat.engine import parser
from heat.engine import scheduler
from heat.openstack.common import uuidutils
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource', 'instance'])
-@attr(speed='fast')
-class nokeyTest(unittest.TestCase):
+class nokeyTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(nokeyTest, self).setUp()
self.fc = fakes.FakeClient()
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_nokey_create(self):
f = open("%s/WordPress_NoKey.template" % self.path)
t = template_format.parse(f.read())
import unittest
-from nose.plugins.attrib import attr
import json
from heat.engine import parameters
-@attr(tag=['unit', 'parameters'])
-@attr(speed='fast')
class ParameterTest(unittest.TestCase):
def test_new_string(self):
p = parameters.Parameter('p', {'Type': 'String'})
}''')
-@attr(tag=['unit', 'parameters'])
-@attr(speed='fast')
class ParametersTest(unittest.TestCase):
def test_pseudo_params(self):
params = parameters.Parameters('test_stack', {"Parameters": {}})
# under the License.
-import unittest
-from nose.plugins.attrib import attr
-import mox
import uuid
from heat.common import context
from heat.engine import parameters
from heat.engine import template
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import stack_delete_after
from heat.tests import generic_resource as generic_rsrc
return parser.Template.resolve_joins(raw)
-@attr(tag=['unit', 'parser'])
-@attr(speed='fast')
-class ParserTest(unittest.TestCase):
+class ParserTest(HeatTestCase):
def test_list(self):
raw = ['foo', 'bar', 'baz']
}''')
-@attr(tag=['unit', 'parser', 'template'])
-@attr(speed='fast')
-class TemplateTest(unittest.TestCase):
- def setUp(self):
- self.m = mox.Mox()
-
- def tearDown(self):
- self.m.UnsetStubs()
-
+class TemplateTest(HeatTestCase):
def test_defaults(self):
empty = parser.Template({})
try:
dict_snippet)
-@attr(tag=['unit', 'parser', 'stack'])
-@attr(speed='fast')
-class StackTest(unittest.TestCase):
+class StackTest(HeatTestCase):
def setUp(self):
- self.username = 'parser_stack_test_user'
+ super(StackTest, self).setUp()
- self.m = mox.Mox()
+ self.username = 'parser_stack_test_user'
setup_dummy_db()
self.ctx = context.get_admin_context()
self.m.ReplayAll()
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_state_defaults(self):
stack = parser.Stack(None, 'test_stack', parser.Template({}))
self.assertEqual(stack.state, None)
import pkgutil
import sys
import unittest
-from nose.plugins.attrib import attr
import heat.engine
from heat.common import plugin_loader
-@attr(tag=['unit', 'plugin_loader'])
-@attr(speed='fast')
class PluginLoaderTest(unittest.TestCase):
def test_module_name(self):
self.assertEqual(plugin_loader._module_name('foo.bar', 'blarg.wibble'),
import unittest
-from nose.plugins.attrib import attr
from heat.engine import properties
from heat.common import exception
-@attr(tag=['unit', 'properties'])
-@attr(speed='fast')
class PropertyTest(unittest.TestCase):
def test_required_default(self):
p = properties.Property({'Type': 'String'})
self.assertRaises(TypeError, p.validate_data, [42, 'fish'])
-@attr(tag=['unit', 'properties'])
-@attr(speed='fast')
class PropertiesTest(unittest.TestCase):
def setUp(self):
schema = {
self.assertEqual(self.props.get('foo', 'wibble'), 'wibble')
-@attr(tag=['unit', 'properties'])
-@attr(speed='fast')
class PropertiesValidationTest(unittest.TestCase):
def test_required(self):
schema = {'foo': {'Type': 'String', 'Required': True}}
import os
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
-from nose.exc import SkipTest
+from testtools import skipIf
from heat.common import context
from heat.common import exception
from heat.engine.resources.quantum import port
from heat.engine.resources.quantum.quantum import QuantumResource as qr
from heat.engine import parser
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
}}
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class QuantumTest(unittest.TestCase):
+class QuantumTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(QuantumTest, self).setUp()
self.m.StubOutWithMock(net.Net, 'quantum')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
'admin_state_up': False}, props)
def test_net(self):
- if net.clients.quantumclient is None:
- raise SkipTest
+ skipIf(net.clients.quantumclient is None, 'quantumclient unavailable')
fq = FakeQuantum()
net.Net.quantum().MultipleTimes().AndReturn(fq)
self.m.VerifyAll()
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class QuantumFloatingIPTest(unittest.TestCase):
+class QuantumFloatingIPTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(QuantumFloatingIPTest, self).setUp()
self.m.StubOutWithMock(floatingip.FloatingIP, 'quantum')
self.m.StubOutWithMock(floatingip.FloatingIPAssociation, 'quantum')
self.m.StubOutWithMock(port.Port, 'quantum')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self, name='Quantum'):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# under the License.
-import unittest
-from nose.plugins.attrib import attr
-import mox
-
from heat.common import context
from heat.common import exception
from heat.engine import parser
from heat.openstack.common import uuidutils
from heat.tests import generic_resource as generic_rsrc
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class ResourceTest(unittest.TestCase):
+class ResourceTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(ResourceTest, self).setUp()
setup_dummy_db()
self.stack = parser.Stack(None, 'test_stack', parser.Template({}),
stack_id=uuidutils.generate_uuid())
resource._register_class('GenericResourceType',
generic_rsrc.GenericResource)
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_get_class_ok(self):
cls = resource.get_class('GenericResourceType')
self.assertEqual(cls, generic_rsrc.GenericResource)
self.assertEqual(res.UPDATE_FAILED, res.state)
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class MetadataTest(unittest.TestCase):
+class MetadataTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(MetadataTest, self).setUp()
tmpl = {
'Type': 'Foo',
'Metadata': {'Test': 'Initial metadata'}
def tearDown(self):
self.stack.delete()
- self.m.UnsetStubs()
+ super(HeatTestCase, self).tearDown()
def test_read_initial(self):
self.assertEqual(self.res.metadata, {'Test': 'Initial metadata'})
"""
-from nose.plugins.attrib import attr
from oslo.config import cfg
import stubout
import unittest
from heat.openstack.common import rpc
-@attr(tag=['unit', 'rpcapi'])
class EngineRpcAPITestCase(unittest.TestCase):
def setUp(self):
'wordpress'))
super(EngineRpcAPITestCase, self).setUp()
- def tearDown(self):
- super(EngineRpcAPITestCase, self).tearDown()
-
def _test_engine_api(self, method, rpc_method, **kwargs):
ctxt = context.RequestContext('fake_user', 'fake_project')
if 'rpcapi_class' in kwargs:
import os
import re
-import unittest
import mox
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import template_format
from heat.openstack.common.importutils import try_import
from heat.engine.resources import s3
from heat.engine import parser
from heat.engine import scheduler
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import skip_if
swiftclient = try_import('swiftclient.client')
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class s3Test(unittest.TestCase):
+class s3Test(HeatTestCase):
@skip_if(swiftclient is None, 'unable to import swiftclient')
def setUp(self):
- self.m = mox.Mox()
+ super(s3Test, self).setUp()
self.m.CreateMock(swiftclient.Connection)
self.m.StubOutWithMock(swiftclient.Connection, 'put_container')
self.m.StubOutWithMock(swiftclient.Connection, 'delete_container')
self.container_pattern = 'test_stack-test_resource-[0-9a-z]+'
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# under the License.
import collections
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
from heat.engine import clients
from heat.common import context
from heat.common import template_format
from heat.engine import parser
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.v1_1 import fakes
]))
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class SecurityGroupTest(unittest.TestCase):
+class SecurityGroupTest(HeatTestCase):
test_template_nova = '''
HeatTemplateFormatVersion: '2012-12-12'
'''
def setUp(self):
- self.m = mox.Mox()
+ super(SecurityGroupTest, self).setUp()
self.fc = fakes.FakeClient()
self.m.StubOutWithMock(clients.OpenStackClients, 'nova')
self.m.StubOutWithMock(nova_sgr.SecurityGroupRuleManager, 'create')
self.m.StubOutWithMock(nova_sg.SecurityGroupManager, 'list')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def create_stack(self, template):
t = template_format.parse(template)
stack = self.parse_stack(t)
import unittest
-from nose.plugins.attrib import attr
from heat.common import short_id
import uuid
-@attr(tag=['unit', 'short_id'])
-@attr(speed='fast')
class ShortIdTest(unittest.TestCase):
def test_byte_string_8(self):
import os
import re
-import unittest
import mox
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import template_format
from heat.openstack.common.importutils import try_import
from heat.engine.resources import swift
from heat.engine import parser
from heat.engine import scheduler
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
from heat.tests.utils import skip_if
swiftclient = try_import('swiftclient.client')
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
-class swiftTest(unittest.TestCase):
+class swiftTest(HeatTestCase):
@skip_if(swiftclient is None, 'unable to import swiftclient')
def setUp(self):
- self.m = mox.Mox()
+ super(swiftTest, self).setUp()
self.m.CreateMock(swiftclient.Connection)
self.m.StubOutWithMock(swiftclient.Connection, 'put_container')
self.m.StubOutWithMock(swiftclient.Connection, 'delete_container')
self.container_pattern = 'test_stack-test_resource-[0-9a-z]+'
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
# License for the specific language governing permissions and limitations
# under the License.
-from nose.plugins.attrib import attr
-from nose.exc import SkipTest
+from testtools import skipIf
import os
import unittest
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit'])
class JsonToYamlTest(unittest.TestCase):
def setUp(self):
yield (json_str, yml_str, f.name)
-@attr(tag=['unit'])
class YamlMinimalTest(unittest.TestCase):
def test_minimal_yaml(self):
self.assertEqual(tpl1, tpl2)
-@attr(tag=['unit'])
class JsonYamlResolvedCompareTest(unittest.TestCase):
def setUp(self):
self.assertEqual(stack1.resources[key].t, stack2.resources[key].t)
def test_quantum_resolved(self):
- if clients.quantumclient is None:
- raise SkipTest
+ skipIf(clients.quantumclient is None, 'quantumclient unavailable')
self.compare_stacks('Quantum.template', 'Quantum.yaml', {})
def test_wordpress_resolved(self):
# License for the specific language governing permissions and limitations
# under the License.
-import mox
-from nose.plugins.attrib import attr
import StringIO
-import unittest
import urllib2
from heat.common import urlfetch
+from heat.tests.common import HeatTestCase
-@attr(tag=['unit', 'urlfetch'])
-@attr(speed='fast')
-class UrlFetchTest(unittest.TestCase):
+class UrlFetchTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(UrlFetchTest, self).setUp()
self.m.StubOutWithMock(urllib2, 'urlopen')
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_file_scheme(self):
self.m.ReplayAll()
self.assertRaises(IOError, urlfetch.get, 'file:///etc/profile')
import os
-import unittest
-import mox
-from nose.plugins.attrib import attr
from oslo.config import cfg
from heat.common import config
from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import user
+from heat.tests.common import HeatTestCase
from heat.tests import fakes
from heat.tests.utils import setup_dummy_db
import keystoneclient.exceptions
-class UserPolicyTestCase(unittest.TestCase):
+class UserPolicyTestCase(HeatTestCase):
def setUp(self):
+ super(UserPolicyTestCase, self).setUp()
config.register_engine_opts()
- self.m = mox.Mox()
self.fc = fakes.FakeKeystoneClient(username='test_stack.CfnUser')
cfg.CONF.set_default('heat_stack_user_role', 'stack_user_role')
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
-@attr(tag=['unit', 'resource', 'User'])
-@attr(speed='fast')
class UserTest(UserPolicyTestCase):
def load_template(self, template_name='Rails_Single_Instance.template'):
self.m.VerifyAll()
-@attr(tag=['unit', 'resource', 'AccessKey'])
-@attr(speed='fast')
class AccessKeyTest(UserPolicyTestCase):
def load_template(self):
self.m.VerifyAll()
-@attr(tag=['unit', 'resource', 'AccessPolicy'])
-@attr(speed='fast')
class AccessPolicyTest(UserPolicyTestCase):
def load_template(self):
# under the License.
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
-
from heat.tests.v1_1 import fakes
from heat.common import exception
from heat.common import template_format
from heat.engine import service
import heat.db.api as db_api
from heat.engine import parser
+from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
test_template_volumeattach = '''
'''
-@attr(tag=['unit', 'validate'])
-@attr(speed='fast')
-class validateTest(unittest.TestCase):
+class validateTest(HeatTestCase):
def setUp(self):
- self.m = mox.Mox()
+ super(validateTest, self).setUp()
self.fc = fakes.FakeClient()
setup_dummy_db()
- def tearDown(self):
- self.m.UnsetStubs()
-
def test_validate_volumeattach_valid(self):
t = template_format.parse(test_template_volumeattach % 'vdq')
stack = parser.Stack(None, 'test_stack', parser.Template(t))
import eventlet
-from nose.plugins.attrib import attr
-
from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.tests.utils import setup_dummy_db
-@attr(tag=['unit', 'resource', 'volume'])
-@attr(speed='fast')
class VolumeTest(HeatTestCase):
def setUp(self):
super(VolumeTest, self).setUp()
# License for the specific language governing permissions and limitations
# under the License.
-
-import unittest
-import mox
-
-from nose.plugins.attrib import attr
+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.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
try:
from quantumclient.common.exceptions import QuantumClientException
from quantumclient.v2_0 import client as quantumclient
except ImportError:
- from nose.exc import SkipTest
- raise SkipTest()
+ quantumclient = None
-class VPCTestBase(unittest.TestCase):
+class VPCTestBase(HeatTestCase):
def setUp(self):
+ super(VPCTestBase, self).setUp()
+ skipIf(quantumclient is None, 'quantumclient unavaialble')
setup_dummy_db()
- self.m = mox.Mox()
self.m.StubOutWithMock(quantumclient.Client, 'add_interface_router')
self.m.StubOutWithMock(quantumclient.Client, 'add_gateway_router')
self.m.StubOutWithMock(quantumclient.Client, 'create_network')
self.m.StubOutWithMock(quantumclient.Client, 'show_subnet')
self.m.StubOutWithMock(quantumclient.Client, 'show_network')
- def tearDown(self):
- self.m.UnsetStubs()
-
def create_stack(self, template):
t = template_format.parse(template)
stack = self.parse_stack(t)
self.assertEqual(metadata, dict(resource.metadata))
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class VPCTest(VPCTestBase):
test_template = '''
self.m.VerifyAll()
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class SubnetTest(VPCTestBase):
test_template = '''
self.m.VerifyAll()
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class NetworkInterfaceTest(VPCTestBase):
test_template = '''
self.m.VerifyAll()
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class InternetGatewayTest(VPCTestBase):
test_template = '''
self.m.VerifyAll()
-@attr(tag=['unit', 'resource'])
-@attr(speed='fast')
class RouteTableTest(VPCTestBase):
test_template = '''
# under the License.
-import mox
import uuid
import datetime
import json
import eventlet
-from nose.plugins.attrib import attr
from oslo.config import cfg
-import unittest
+from heat.tests.common import HeatTestCase
from heat.tests import fakes
from heat.tests.utils import stack_delete_after
from heat.tests.utils import setup_dummy_db
'''
-@attr(tag=['unit', 'resource', 'WaitCondition'])
-@attr(speed='slow')
-class WaitConditionTest(unittest.TestCase):
+class WaitConditionTest(HeatTestCase):
def setUp(self):
+ super(WaitConditionTest, self).setUp()
config.register_engine_opts()
setup_dummy_db()
- self.m = mox.Mox()
self.m.StubOutWithMock(wc.WaitConditionHandle,
'get_status')
self.m.StubOutWithMock(wc.WaitCondition,
self.fc = fakes.FakeKeystoneClient()
- def tearDown(self):
- self.m.UnsetStubs()
-
# Note tests creating a stack should be decorated with @stack_delete_after
# to ensure the stack is properly cleaned up
def create_stack(self, stack_name='test_stack',
self.m.VerifyAll()
-@attr(tag=['unit', 'resource', 'WaitConditionHandle'])
-@attr(speed='fast')
-class WaitConditionHandleTest(unittest.TestCase):
+class WaitConditionHandleTest(HeatTestCase):
def setUp(self):
+ super(WaitConditionHandleTest, self).setUp()
config.register_engine_opts()
- self.m = mox.Mox()
cfg.CONF.set_default('heat_waitcondition_server_url',
'http://127.0.0.1:8000/v1/waitcondition')
setup_dummy_db()
self.stack = self.create_stack()
- def tearDown(self):
- self.m.UnsetStubs()
-
def create_stack(self, stack_name='test_stack2', params={}):
temp = template_format.parse(test_template_waitcondition)
template = parser.Template(temp)
import datetime
import mox
-from nose.plugins.attrib import attr
-import unittest
from heat.common import context
import heat.db.api as db_api
from heat.openstack.common import timeutils
from heat.engine import watchrule
from heat.engine import parser
+from heat.tests.common import HeatTestCase
from heat.tests import utils
alarm = "DummyAction"
-@attr(tag=['unit', 'watchrule'])
-@attr(speed='fast')
-class WatchRuleTest(unittest.TestCase):
+class WatchRuleTest(HeatTestCase):
@classmethod
def setUpClass(cls):
cls.stack_id = dummy_stack.id
def setUp(self):
+ super(WatchRuleTest, self).setUp()
self.username = 'watchrule_test_user'
- self.m = mox.Mox()
-
self.ctx = context.get_admin_context()
self.ctx.username = self.username
self.ctx.tenant_id = u'123456'
self.m.ReplayAll()
- def tearDown(self):
- self.m.UnsetStubs()
-
def _action_set_stubs(self, now, action_expected=True):
# Setup stubs for the action tests
self.m.StubOutWithMock(timeutils, 'utcnow')