# under the License.
import datetime
-import uuid
from oslo.config import cfg
from heat.tests import generic_resource
from heat.tests import fakes
from heat.tests.common import HeatTestCase
-from heat.tests.utils import stack_delete_after
-from heat.tests.utils import setup_dummy_db
+from heat.tests import utils
from heat.common import context
from heat.common import exception
'''
-class UUIDStub(object):
- def __init__(self, value):
- self.value = value
-
- def __enter__(self):
- self.uuid4 = uuid.uuid4
- uuid_stub = lambda: self.value
- uuid.uuid4 = uuid_stub
-
- def __exit__(self, *exc_info):
- uuid.uuid4 = self.uuid4
-
-
class SignalTest(HeatTestCase):
def setUp(self):
super(SignalTest, self).setUp()
- setup_dummy_db()
+ utils.setup_dummy_db()
resource._register_class('SignalResourceType',
generic_resource.SignalResource)
disable_rollback=True)
# Stub out the stack ID so we have a known value
- with UUIDStub(self.stack_id):
+ with utils.UUIDStub(self.stack_id):
stack.store()
if stub:
self.fc)
return stack
- @stack_delete_after
+ @utils.stack_delete_after
def test_FnGetAtt_Alarm_Url(self):
self.stack = self.create_stack()
self.assertEqual(expected_url, rsrc.FnGetAtt('AlarmUrl'))
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_signal(self):
test_d = {'Data': 'foo', 'Reason': 'bar',
'Status': 'SUCCESS', 'UniqueId': '123'}
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_signal_wrong_resource(self):
# assert that we get the correct exception when calling a
# resource.signal() that does not have a handle_signal()
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_signal_reception_wrong_state(self):
# assert that we get the correct exception when calling a
# resource.signal() that is in having a destructive action.
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_signal_reception_failed_call(self):
# assert that we get the correct exception from resource.signal()
# when resource.handle_signal() raises an exception.
import datetime
import time
import json
-import uuid
from oslo.config import cfg
from heat.tests.common import HeatTestCase
from heat.tests import fakes
-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.tests import utils
import heat.db.api as db_api
from heat.common import template_format
'''
-class UUIDStub(object):
- def __init__(self, value):
- self.value = value
-
- def __enter__(self):
- self.uuid4 = uuid.uuid4
- uuid_stub = lambda: self.value
- uuid.uuid4 = uuid_stub
-
- def __exit__(self, *exc_info):
- uuid.uuid4 = self.uuid4
-
-
class WaitConditionTest(HeatTestCase):
def setUp(self):
super(WaitConditionTest, self).setUp()
- setup_dummy_db()
+ utils.setup_dummy_db()
self.m.StubOutWithMock(wc.WaitConditionHandle,
'get_status')
stub=True):
temp = template_format.parse(template)
template = parser.Template(temp)
- ctx = dummy_context(tenant_id='test_tenant')
+ ctx = utils.dummy_context(tenant_id='test_tenant')
stack = parser.Stack(ctx, stack_name, template,
environment.Environment(params),
disable_rollback=True)
# Stub out the stack ID so we have a known value
- with UUIDStub(self.stack_id):
+ with utils.UUIDStub(self.stack_id):
stack.store()
if stub:
return stack
- @stack_delete_after
+ @utils.stack_delete_after
def test_post_success_to_handle(self):
self.stack = self.create_stack()
wc.WaitConditionHandle.get_status().AndReturn([])
self.assertEqual(r.name, 'WaitHandle')
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_post_failure_to_handle(self):
self.stack = self.create_stack()
wc.WaitConditionHandle.get_status().AndReturn([])
self.assertEqual(r.name, 'WaitHandle')
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_post_success_to_handle_count(self):
self.stack = self.create_stack(template=test_template_wc_count)
wc.WaitConditionHandle.get_status().AndReturn([])
self.assertEqual(r.name, 'WaitHandle')
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_post_failure_to_handle_count(self):
self.stack = self.create_stack(template=test_template_wc_count)
wc.WaitConditionHandle.get_status().AndReturn([])
self.assertEqual(r.name, 'WaitHandle')
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_timeout(self):
st = time.time()
rsrc.handle_update, {}, {}, {})
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_FnGetAtt(self):
self.stack = self.create_stack()
wc.WaitConditionHandle.get_status().AndReturn(['SUCCESS'])
self.assertEqual(wc_att, u'{"123": "foo", "456": "dog"}')
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_validate_handle_url_bad_stackid(self):
self.m.ReplayAll()
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_validate_handle_url_bad_stackname(self):
self.m.ReplayAll()
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_validate_handle_url_bad_tenant(self):
self.m.ReplayAll()
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_validate_handle_url_bad_resource(self):
self.m.ReplayAll()
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_validate_handle_url_bad_resource_type(self):
self.m.ReplayAll()
'http://127.0.0.1:8000/v1/waitcondition')
self.fc = fakes.FakeKeystoneClient()
- setup_dummy_db()
+ utils.setup_dummy_db()
self.stack = self.create_stack()
def create_stack(self, stack_name='test_stack2', params={}):
temp = template_format.parse(test_template_waitcondition)
template = parser.Template(temp)
- ctx = dummy_context(tenant_id='test_tenant')
+ ctx = utils.dummy_context(tenant_id='test_tenant')
stack = parser.Stack(ctx, stack_name, template,
environment.Environment(params),
disable_rollback=True)
# Stub out the UUID for this test, so we can get an expected signature
- with UUIDStub('STACKABCD1234'):
+ with utils.UUIDStub('STACKABCD1234'):
stack.store()
# Stub waitcondition status so all goes CREATE_COMPLETE
return stack
- @stack_delete_after
+ @utils.stack_delete_after
def test_handle(self):
created_time = datetime.datetime(2012, 11, 29, 13, 49, 37)
rsrc.handle_update, {}, {}, {})
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_metadata_update(self):
rsrc = self.stack.resources['WaitHandle']
self.assertEqual(rsrc.state, (rsrc.CREATE, rsrc.COMPLETE))
self.assertEqual(rsrc.metadata, handle_metadata)
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_metadata_update_invalid(self):
rsrc = self.stack.resources['WaitHandle']
self.assertEqual(rsrc.state, (rsrc.CREATE, rsrc.COMPLETE))
new_metadata=err_metadata)
self.m.VerifyAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_get_status(self):
rsrc = self.stack.resources['WaitHandle']
self.assertEqual(rsrc.state, (rsrc.CREATE, rsrc.COMPLETE))
wc.WaitConditionHandle.keystone().MultipleTimes().AndReturn(self.fc)
self.m.ReplayAll()
- @stack_delete_after
+ @utils.stack_delete_after
def test_get_status_reason(self):
rsrc = self.stack.resources['WaitHandle']
self.assertEqual(rsrc.state, (rsrc.CREATE, rsrc.COMPLETE))