Add internal testing for the stress test framework
This adds a unit test for StressAction class and a end2end test
case for the whole stress test framework.
Change-Id: I4d30a131c75736fa05ba63f71cd731374f7fb9ba
diff --git a/tempest/tests/stress/__init__.py b/tempest/tests/stress/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tempest/tests/stress/__init__.py
diff --git a/tempest/tests/stress/test_stress.py b/tempest/tests/stress/test_stress.py
new file mode 100644
index 0000000..4d7de9d
--- /dev/null
+++ b/tempest/tests/stress/test_stress.py
@@ -0,0 +1,57 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Deutsche Telekom AG
+# All Rights Reserved.
+#
+# 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 shlex
+import subprocess
+
+import tempest.cli as cli
+from tempest.openstack.common import log as logging
+import tempest.test
+
+LOG = logging.getLogger(__name__)
+
+
+class StressFrameworkTest(tempest.test.BaseTestCase):
+ """Basic test for the stress test framework.
+ """
+
+ def _cmd(self, cmd, param):
+ """Executes specified command."""
+ cmd = ' '.join([cmd, param])
+ LOG.info("running: '%s'" % cmd)
+ cmd_str = cmd
+ cmd = shlex.split(cmd)
+ result = ''
+ result_err = ''
+ try:
+ stdout = subprocess.PIPE
+ stderr = subprocess.PIPE
+ proc = subprocess.Popen(
+ cmd, stdout=stdout, stderr=stderr)
+ result, result_err = proc.communicate()
+ if proc.returncode != 0:
+ LOG.debug('error of %s:\n%s' % (cmd_str, result_err))
+ raise cli.CommandFailed(proc.returncode,
+ cmd,
+ result)
+ finally:
+ LOG.debug('output of %s:\n%s' % (cmd_str, result))
+ return proc.returncode
+
+ def test_help_function(self):
+ result = self._cmd("python", "-m tempest.stress.run_stress -h")
+ self.assertEqual(0, result)
diff --git a/tempest/tests/stress/test_stressaction.py b/tempest/tests/stress/test_stressaction.py
new file mode 100644
index 0000000..3d2901e
--- /dev/null
+++ b/tempest/tests/stress/test_stressaction.py
@@ -0,0 +1,65 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Deutsche Telekom AG
+# All Rights Reserved.
+#
+# 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 tempest.stress.stressaction as stressaction
+import tempest.test
+
+
+class FakeStressAction(stressaction.StressAction):
+ def __init__(self, manager, max_runs=None, stop_on_error=False):
+ super(self.__class__, self).__init__(manager, max_runs, stop_on_error)
+ self._run_called = False
+
+ def run(self):
+ self._run_called = True
+
+ @property
+ def run_called(self):
+ return self._run_called
+
+
+class FakeStressActionFailing(stressaction.StressAction):
+ def run(self):
+ raise Exception('FakeStressActionFailing raise exception')
+
+
+class TestStressAction(tempest.test.BaseTestCase):
+ def _bulid_stats_dict(self, runs=0, fails=0):
+ return {'runs': runs, 'fails': fails}
+
+ def testStressTestRun(self):
+ stressAction = FakeStressAction(manager=None, max_runs=1)
+ stats = self._bulid_stats_dict()
+ stressAction.execute(stats)
+ self.assertTrue(stressAction.run_called)
+ self.assertEqual(stats['runs'], 1)
+ self.assertEqual(stats['fails'], 0)
+
+ def testStressMaxTestRuns(self):
+ stressAction = FakeStressAction(manager=None, max_runs=500)
+ stats = self._bulid_stats_dict(runs=499)
+ stressAction.execute(stats)
+ self.assertTrue(stressAction.run_called)
+ self.assertEqual(stats['runs'], 500)
+ self.assertEqual(stats['fails'], 0)
+
+ def testStressTestRunWithException(self):
+ stressAction = FakeStressActionFailing(manager=None, max_runs=1)
+ stats = self._bulid_stats_dict()
+ stressAction.execute(stats)
+ self.assertEqual(stats['runs'], 1)
+ self.assertEqual(stats['fails'], 1)