return rsrc
def test_loadbalancer(self):
+ self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ scheduler.TaskRunner._sleep(mox.IsA(int)).MultipleTimes()
+
self.m.StubOutWithMock(user.User, 'keystone')
user.User.keystone().AndReturn(self.fkc)
self.m.StubOutWithMock(user.AccessKey, 'keystone')
# License for the specific language governing permissions and limitations
# under the License.
+import mox
import uuid
from heat.engine import resource
from heat.engine import parser
from heat.engine import parameters
+from heat.engine import scheduler
from heat.engine import template
from heat.tests.common import HeatTestCase
self.stack = parser.Stack(self.ctx, 'update_test_stack',
template.Template(tmpl))
+
+ self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ scheduler.TaskRunner._sleep(mox.IsA(int)).AndReturn(None)
+ self.m.ReplayAll()
+
self.stack.store()
self.stack.create()
+ self.m.VerifyAll()
self.assertEqual(self.stack.state, parser.Stack.CREATE_COMPLETE)
self.assertEqual(self.stack['AResource'].properties['Foo'], 'abc')
self.assertEqual(self.stack['BResource'].properties['Foo'],
self.stack = parser.Stack(self.ctx, 'update_test_stack',
template.Template(tmpl),
disable_rollback=False)
+
+ self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ scheduler.TaskRunner._sleep(mox.IsA(int)).AndReturn(None)
+ self.m.ReplayAll()
+
self.stack.store()
self.stack.create()
+ self.m.VerifyAll()
+
self.assertEqual(self.stack.state, parser.Stack.CREATE_COMPLETE)
self.assertEqual(self.stack['AResource'].properties['Foo'], 'abc')
self.assertEqual(self.stack['BResource'].properties['Foo'],
self.stack = parser.Stack(self.ctx, 'update_test_stack',
template.Template(tmpl),
disable_rollback=False)
+
+ self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ scheduler.TaskRunner._sleep(mox.IsA(int)).AndReturn(None)
+ self.m.ReplayAll()
+
self.stack.store()
self.stack.create()
+ self.m.VerifyAll()
+
self.assertEqual(self.stack.state, parser.Stack.CREATE_COMPLETE)
self.assertEqual(self.stack['AResource'].properties['Foo'], 'abc')
self.assertEqual(self.stack['BResource'].properties['Foo'],
from testtools import skipIf
+import mox
+
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.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
self.m.StubOutWithMock(
quantumclient.Client, 'delete_security_group_rule')
+ def stub_sleep(self):
+ self.m.StubOutWithMock(scheduler.TaskRunner, '_sleep')
+ scheduler.TaskRunner._sleep(mox.IsA(int)).MultipleTimes()
+ mox.Replay(scheduler.TaskRunner._sleep)
+
def create_stack(self, template):
t = template_format.parse(template)
stack = self.parse_stack(t)
+ self.stub_sleep()
self.assertEqual(None, stack.create())
return stack
Properties: {CidrBlock: '10.0.0.0/16'}
'''
+ def stub_sleep(self):
+ pass
+
def test_vpc(self):
self.mock_create_network()
self.mock_delete_network()