Unit Test cases for Timerthread and cpulse Api

Partially implements: blueprint unit-tests

Change-Id: I90b9563dcf2de1985eda481508b02ad872ee6fd1

cloudpulse/config.py

@@ -12,7 +12,7 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
-"""Cloudpulse specific config handling."""
+"""Magnum specific config handling."""
 
 from oslo_config import cfg
 
@@ -25,15 +25,17 @@ server = {
 }
 
 # Pecan Application Configurations
-app = {'root': 'cloudpulse.api.controllers.root.RootController',
-       'modules': ['cloudpulse.api'],
-       'static_root': '%(confdir)s/public',
-       'template_path': '%(confdir)s/api/templates',
-       'debug': True,
-       'errors': {404: '/error/404',
-                  '__force_dict__': True
-                  }
-       }
+app = {
+    'root': 'cloudpulse.api.controllers.root.RootController',
+    'modules': ['cloudpulse.api'],
+    'static_root': '%(confdir)s/public',
+    'template_path': '%(confdir)s/api/templates',
+    'debug': True,
+    'errors': {
+        404: '/error/404',
+        '__force_dict__': True
+    }
+}
 
 logging = {
     'root': {'level': 'INFO', 'handlers': ['console']},

cloudpulse/tests/base.py

@@ -26,6 +26,7 @@ import mock
 from oslo_config import cfg
 from oslotest import base
 import pecan
+from pecan import testing
 
 CONF = cfg.CONF
 CONF.set_override('use_stderr', False)
@@ -34,6 +35,10 @@ CONF.set_override('use_stderr', False)
 class TestCase(base.BaseTestCase):
     def setUp(self):
         super(TestCase, self).setUp()
+        self.app = testing.load_test_app(os.path.join(
+            os.path.dirname(__file__),
+            'config.py'
+        ))
         token_info = {
             'token': {
                 'project': {

cloudpulse/tests/config.py

@@ -0,0 +1,40 @@
+#    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.
+
+from cloudpulse.api import hooks
+
+# Server Specific Configurations
+server = {
+    'port': '8080',
+    'host': '0.0.0.0'
+}
+
+# Pecan Application Configurations
+app = {
+    'root': 'cloudpulse.api.controllers.root.RootController',
+    'modules': ['cloudpulse.api'],
+    'debug': True,
+    'hooks': [
+        hooks.ContextHook(),
+        hooks.RPCHook()
+    ],
+    'acl_public_routes': [
+        '/'
+    ],
+}
+
+# Custom Configurations must be in Python dictionary format::
+#
+# foo = {'bar':'baz'}
+#
+# All configurations are accessible at::
+# pecan.conf

cloudpulse/tests/unit/api/controllers/v1/test_cpulse.py

@@ -0,0 +1,76 @@
+#    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.
+
+from cloudpulse import objects
+from cloudpulse.tests.unit.db import base as db_base
+from cloudpulse.tests.unit.db import utils
+
+import mock
+from mock import patch
+from webtest.app import AppError
+
+
+class TestCpulseController(db_base.DbTestCase):
+
+    @patch('cloudpulse.conductor.api.API.test_create')
+    def test_cpulse_create(self, mock_test_create):
+        params = ('{"name": "nova_endpoint"}')
+        response = self.app.post('/cpulse',
+                                 params=params,
+                                 content_type='application/json')
+        self.assertEqual(response.status_int, 201)
+        self.assertTrue(mock_test_create.called)
+
+    def test_cpulse_create_name_not_given(self):
+        params = ('{"name": ""}')
+        self.assertRaises(AppError, self.app.post, '/cpulse',
+                          params=params, content_type='application/json')
+
+    @patch('cloudpulse.conductor.api.API.test_show')
+    @patch('cloudpulse.objects.Cpulse.get_by_uuid')
+    def test_cpulse_getone_by_uuid(self, mock_get_by_uuid, mock_test_show):
+        cpulse = utils.get_cpulse_test()
+        cpulse_obj = objects.Cpulse(self.context, **cpulse)
+        mock_get_by_uuid.return_value = cpulse_obj
+        uuid = cpulse.get('uuid')
+        response = self.app.get('/cpulse/%s' % uuid)
+        self.assertEqual(response.status_int, 200)
+
+    @patch('cloudpulse.conductor.api.API.test_show')
+    @patch('cloudpulse.objects.Cpulse.list')
+    def test_cpulse_getall_tests(self, mock_test_list, mock_test_show):
+        cpulse = utils.get_cpulse_test()
+        cpulse_obj = [objects.Cpulse(self.context, **cpulse)]
+        mock_test_list.return_value = cpulse_obj
+        mock_test_show.return_value = cpulse_obj[0]
+        response = self.app.get('/cpulse')
+        self.assertEqual(response.status_int, 200)
+        unittest_cpulses = response.json['cpulses']
+        self.assertEqual(len(unittest_cpulses), 1)
+        self.assertEqual(unittest_cpulses[0].get('uuid'),
+                         cpulse['uuid'])
+
+    @patch('cloudpulse.conductor.api.API.test_delete')
+    def test_cpulse_delete_not_found(self, mock_test_delete):
+        uuid = "123"
+        self.assertRaises(AppError, self.app.delete, '/cpulse/%s' % uuid)
+
+    @patch('cloudpulse.conductor.api.API.test_delete')
+    @patch('cloudpulse.objects.Cpulse.get_by_uuid')
+    def test_cpulse_delete_by_uuid(self, mock_get_by_uuid, mock_test_delete):
+        cpulse = utils.get_cpulse_test()
+        cpulse_obj = objects.Cpulse(self.context, **cpulse)
+        mock_get_by_uuid.return_value = cpulse_obj
+        uuid = cpulse.get('uuid')
+        response = self.app.delete('/cpulse/%s' % uuid)
+        mock_test_delete.assert_called_once_with(mock.ANY, uuid)
+        self.assertEqual(response.status_int, 204)

cloudpulse/tests/unit/common/test_timerthread.py

@@ -0,0 +1,47 @@
+#    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.
+from cloudpulse.common import timerthread
+from cloudpulse import objects
+from cloudpulse.tests.unit.db import base as db_base
+from cloudpulse.tests.unit.db import utils
+
+from mock import patch
+import time
+
+
+class Test_Cpulse_TimerThread(db_base.DbTestCase):
+
+    @patch('cloudpulse.objects.Cpulse.list')
+    def test_timerfunc_no_test(self, mock_test_list):
+        timerthread.timerfunc()
+
+    @patch('cloudpulse.TestManager.TestManager.TestManager.run')
+    @patch('cloudpulse.objects.Cpulse.list')
+    def test_timerfunc_list(self, mock_test_list, mock_TestManager_run):
+        cpulse = utils.get_cpulse_test()
+        cpulse_obj = [objects.Cpulse(self.context, **cpulse)]
+        mock_test_list.return_value = cpulse_obj
+        timerthread.timerfunc()
+        time.sleep(1)
+        self.assertTrue(mock_TestManager_run.not_called)
+
+    @patch('cloudpulse.TestManager.TestManager.TestManager.run')
+    @patch('cloudpulse.objects.Cpulse.list')
+    def test_timerfunc_list_manual_test(self, mock_test_list,
+                                        mock_TestManager_run):
+        cpulse = utils.get_cpulse_test()
+        cpulse['testtype'] = 'manual'
+        cpulse_obj = [objects.Cpulse(self.context, **cpulse)]
+        mock_test_list.return_value = cpulse_obj
+        timerthread.timerfunc()
+        time.sleep(1)
+        mock_TestManager_run.assert_called_once_with(test=cpulse_obj[0])

cloudpulse/tests/unit/db/utils.py

@@ -20,7 +20,7 @@ def get_cpulse_test(**kw):
     return {
         'id': kw.get('id', 32),
         'uuid': kw.get('uuid', 'e74c40e0-d825-11e2-a28f-0800200c9a66'),
-        'name': kw.get('name', 'nova_endpoint'),
+        'name': kw.get('name', 'dummy_cloudtest'),
         'state': kw.get('state', 'created'),
         'result': kw.get('state', 'success'),
         'testtype': kw.get('testtype', 'periodic'),