test_resource for functional tests
Introduces new resource plugin to be able to test various
functional test cases which includes Rollback, Update In Place,
Concurrent Update and can be customised furthur based on the needs.
Co-Authored-by: Anant Patil <anant.patil@hp.com>
Change-Id: I3b8a1d2928553c87abaac81ee687e0faa85c9c5e
diff --git a/common/test.py b/common/test.py
index b88c72b..331c922 100644
--- a/common/test.py
+++ b/common/test.py
@@ -276,7 +276,8 @@
def update_stack(self, stack_identifier, template, environment=None,
files=None, parameters=None,
- expected_status='UPDATE_COMPLETE'):
+ expected_status='UPDATE_COMPLETE',
+ disable_rollback=True):
env = environment or {}
env_files = files or {}
parameters = parameters or {}
@@ -286,11 +287,19 @@
stack_name=stack_name,
template=template,
files=env_files,
- disable_rollback=True,
+ disable_rollback=disable_rollback,
parameters=parameters,
environment=env
)
- self._wait_for_stack_status(stack_identifier, expected_status)
+ kwargs = {'stack_identifier': stack_identifier,
+ 'status': expected_status}
+ if expected_status in ['ROLLBACK_COMPLETE']:
+ self.addCleanup(self.client.stacks.delete, stack_name)
+ # To trigger rollback you would intentionally fail the stack
+ # Hence check for rollback failures
+ kwargs['failure_pattern'] = '^ROLLBACK_FAILED$'
+
+ self._wait_for_stack_status(**kwargs)
def assert_resource_is_a_stack(self, stack_identifier, res_name,
wait=False):
@@ -334,7 +343,7 @@
def stack_create(self, stack_name=None, template=None, files=None,
parameters=None, environment=None,
- expected_status='CREATE_COMPLETE'):
+ expected_status='CREATE_COMPLETE', disable_rollback=True):
name = stack_name or self._stack_rand_name()
templ = template or self.template
templ_files = files or {}
@@ -344,16 +353,23 @@
stack_name=name,
template=templ,
files=templ_files,
- disable_rollback=True,
+ disable_rollback=disable_rollback,
parameters=params,
environment=env
)
- self.addCleanup(self.client.stacks.delete, name)
+ if expected_status not in ['ROLLBACK_COMPLETE']:
+ self.addCleanup(self.client.stacks.delete, name)
stack = self.client.stacks.get(name)
stack_identifier = '%s/%s' % (name, stack.id)
+ kwargs = {'stack_identifier': stack_identifier,
+ 'status': expected_status}
if expected_status:
- self._wait_for_stack_status(stack_identifier, expected_status)
+ if expected_status in ['ROLLBACK_COMPLETE']:
+ # To trigger rollback you would intentionally fail the stack
+ # Hence check for rollback failures
+ kwargs['failure_pattern'] = '^ROLLBACK_FAILED$'
+ self._wait_for_stack_status(**kwargs)
return stack_identifier
def stack_adopt(self, stack_name=None, files=None,
diff --git a/common/test_resources/test_resource.py b/common/test_resources/test_resource.py
new file mode 100644
index 0000000..88f1745
--- /dev/null
+++ b/common/test_resources/test_resource.py
@@ -0,0 +1,134 @@
+#
+# 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 eventlet
+
+from heat.common.i18n import _
+from heat.engine import attributes
+from heat.engine import properties
+from heat.engine import resource
+from heat.engine import support
+from oslo_log import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+class TestResource(resource.Resource):
+ '''
+ A resource which stores the string value that was provided.
+
+ This resource is to be used only for testing.
+ It has control knobs such as 'update_replace', 'fail', 'wait_secs'
+
+ '''
+
+ support_status = support.SupportStatus(version='2014.1')
+
+ PROPERTIES = (
+ VALUE, UPDATE_REPLACE, FAIL, WAIT_SECS
+ ) = (
+ 'value', 'update_replace', 'fail', 'wait_secs'
+ )
+
+ ATTRIBUTES = (
+ OUTPUT,
+ ) = (
+ 'output',
+ )
+
+ properties_schema = {
+ VALUE: properties.Schema(
+ properties.Schema.STRING,
+ _('The input string to be stored.'),
+ default='test_string',
+ update_allowed=True
+ ),
+ FAIL: properties.Schema(
+ properties.Schema.BOOLEAN,
+ _('Value which can be set to fail the resource operation '
+ 'to test failure scenarios.'),
+ update_allowed=True,
+ default=False
+ ),
+ UPDATE_REPLACE: properties.Schema(
+ properties.Schema.BOOLEAN,
+ _('Value which can be set to trigger update replace for '
+ 'the particular resource'),
+ update_allowed=True,
+ default=False
+ ),
+ WAIT_SECS: properties.Schema(
+ properties.Schema.NUMBER,
+ _('Value which can be set for resource to wait after an action '
+ 'is performed.'),
+ update_allowed=True,
+ default=0,
+ ),
+ }
+
+ attributes_schema = {
+ OUTPUT: attributes.Schema(
+ _('The string that was stored. This value is '
+ 'also available by referencing the resource.'),
+ cache_mode=attributes.Schema.CACHE_NONE
+ ),
+ }
+
+ def handle_create(self):
+ value = self.properties.get(self.VALUE)
+ fail_prop = self.properties.get(self.FAIL)
+ sleep_secs = self.properties.get(self.WAIT_SECS)
+
+ self.data_set('value', value, redact=False)
+ self.resource_id_set(self.physical_resource_name())
+
+ # sleep for specified time
+ if sleep_secs:
+ LOG.debug("Resource %s sleeping for %s seconds",
+ self.name, sleep_secs)
+ eventlet.sleep(sleep_secs)
+
+ # emulate failure
+ if fail_prop:
+ raise Exception("Test Resource failed %s", self.name)
+
+ def handle_update(self, json_snippet=None, tmpl_diff=None, prop_diff=None):
+ value = prop_diff.get(self.VALUE)
+ new_prop = json_snippet._properties
+ if value:
+ update_replace = new_prop.get(self.UPDATE_REPLACE, False)
+ if update_replace:
+ raise resource.UpdateReplace(self.name)
+ else:
+ fail_prop = new_prop.get(self.FAIL, False)
+ sleep_secs = new_prop.get(self.WAIT_SECS, 0)
+ # emulate failure
+ if fail_prop:
+ raise Exception("Test Resource failed %s", self.name)
+ # update in place
+ self.data_set('value', value, redact=False)
+
+ if sleep_secs:
+ LOG.debug("Update of Resource %s sleeping for %s seconds",
+ self.name, sleep_secs)
+ eventlet.sleep(sleep_secs)
+
+ def _resolve_attribute(self, name):
+ if name == self.OUTPUT:
+ return self.data().get('value')
+
+
+def resource_mapping():
+ return {
+ 'OS::Heat::TestResource': TestResource,
+ }