Commits

Tim Savage committed df6a598 Draft

Simplified test framework code. Added testcases for supervisor, added local support to ApiWrapper

Comments (0)

Files changed (5)

denim/service/supervisor.py

 # -*- encoding:utf8 -*-
 from fabric import colors
-from fabric.api import sudo, put, env
+from fabric.api import put, env
 from denim import paths, utils
 
 SERVICE_NAME = 'supervisor'
     Start service manager daemon.
 
     """
-    sudo('/etc/init.d/supervisor start')
+    utils.run_as('/etc/init.d/supervisor start', use_sudo=True)
 
 
 def manager_stop():
     Stop service manager daemon.
 
     """
-    sudo('/etc/init.d/supervisor stop')
+    utils.run_as('/etc/init.d/supervisor stop', use_sudo=True)
 
 
 def manager_restart():
     Reload service manager daemon.
 
     """
-    sudo('supervisorctl reread')
+    utils.run_as('supervisorctl reread', use_sudo=True)
 
 
 def manager_status():
     """
     if not service_name:
         service_name = env.project_name
-    sudo('supervisorctl start %s' % service_name)
+    utils.run_as('supervisorctl start %s' % service_name, use_sudo=True)
 
 
 def stop(service_name=None):
     """
     if not service_name:
         service_name = env.project_name
-    sudo('supervisorctl stop %s' % service_name)
+    utils.run_as('supervisorctl stop %s' % service_name, use_sudo=True)
 
 
 def restart(service_name=None):
     """
     if not service_name:
         service_name = env.project_name
-    sudo('supervisorctl restart %s' % service_name)
+    utils.run_as('supervisorctl restart %s' % service_name, use_sudo=True)
 
 
 def status(service_name=None):
     """
     if service_name is None:
         service_name = env.project_name
-    sudo('supervisorctl status %s' % service_name)
+    utils.run_as('supervisorctl status %s' % service_name, use_sudo=True)
     An easily replaceable wrapper around api commands to allow for easy
     testing.
     """
-    def sudo(self, command, *args, **kwargs):
-        return api.sudo(command, *args, **kwargs)
+    def sudo(self, command, **kwargs):
+        return api.sudo(command, **kwargs)
 
-    def run(self, command, *args, **kwargs):
-        return api.run(command, *args, **kwargs)
+    def run(self, command, **kwargs):
+        return api.run(command, **kwargs)
 
-    def local(self, command, *args, **kwargs):
-        return api.local(command, *args, **kwargs)
+    def local(self, command, **kwargs):
+        return api.local(command, **kwargs)
 
 
 __api_wrapper = ApiWrapper()
         return api.run(command, **kwargs)
 
 
+def local(command, **kwargs):
+    """
+    A wrapper around the local command to push commands through the API
+    wrapper to capture the output.
+    :param command:
+
+    """
+    api = api_wrapper()
+    return api.local(command, **kwargs)
+
+
 def run_test(command, hide_groups=('warnings', ), use_sudo=False, user=None):
     """
     Helper method for performing commands where the result is going to be
         """
         Clear all commands.
         """
-        self.sudo_commands = []
-        self.run_commands = []
+        self.commands = []
 
-    def sudo(self, command, *args, **kwargs):
-        return self.sudo_commands.append((command, args, kwargs))
+    def sudo(self, command, **kwargs):
+        return self.commands.append(('sudo', command, kwargs))
 
-    def run(self, command, *args, **kwargs):
-        return self.run_commands.append((command, args, kwargs))
+    def run(self, command, **kwargs):
+        return self.commands.append(('run', command, kwargs))
+
+    def local(self, command, **kwargs):
+        return self.commands.append(('local', command, kwargs))
 
 set_api_wrapper(TestApiWrapper())
 
         self.api.reset()
 
     def assertCommand(self, expected_command, actual_command):
-        ec, ea, ekwa = expected_command
-        ac, aa, akwa = actual_command
+        e_scope, e_cmd, e_kwargs = expected_command
+        a_scope, a_cmd, a_kwargs = actual_command
 
-        self.assertEqual(ec, ac, msg='Command does not match. Expected "%s" got "%s"' % (ec, ac))
-        self.assertTupleEqual(ea, aa, msg='Command arguments do not match.\nExpected: %s\nActual: %s' % (ea, aa))
-        self.assertDictEqual(ekwa, akwa, msg='Command keyword arguments do not match.\nExpected: %s\nActual: %s' % (ekwa, akwa))
+        self.assertEqual(e_scope, a_scope,
+            msg='Command scope does not match. Expected "%s" got "%s"' % (e_scope, a_scope))
+        self.assertEqual(e_cmd, a_cmd,
+            msg='Command does not match. Expected "%s" got "%s"' % (e_cmd, a_cmd))
+        self.assertDictEqual(e_kwargs, a_kwargs,
+            msg='Command keyword arguments do not match.\nExpected: %s\nActual: %s' % (e_kwargs, a_kwargs))
 
-    def assertSudo(self, command, *args, **kwargs):
-        command_count = len(self.api.sudo_commands)
-        self.assertEqual(command_count, 1, msg="Expected a single sudo command got %s." % command_count)
-        self.assertCommand((command, args, kwargs), self.api.sudo_commands[0])
+    def assertSingeCommand(self, command, _scope, **kwargs):
+        user = kwargs.get('user')
+        if user and hasattr(user, 'sudo_identity'):
+            kwargs['user'] = user.sudo_identity()
+        command_count = len(self.api.commands)
+        self.assertEqual(command_count, 1, msg="Expected a single command got %s." % command_count)
+        self.assertCommand((_scope, command, kwargs), self.api.commands[0])
 
-    def assertRun(self, command, *args, **kwargs):
-        command_count = len(self.api.run_commands)
-        self.assertEqual(command_count, 1, msg="Expected a single run command got %s." % command_count)
-        self.assertCommand((command, args, kwargs), self.api.run_commands[0])
+    def assertSudo(self, command, **kwargs):
+        self.assertSingeCommand(command, _scope='sudo', **kwargs)
+
+    def assertRun(self, command, **kwargs):
+        self.assertSingeCommand(command, _scope='run', **kwargs)
+
+    def assertLocal(self, command, **kwargs):
+        self.assertSingeCommand(command, _scope='local', **kwargs)

tests/service/__init__.py

Empty file added.

tests/service/supervisor.py

+from denim.constants import RootUser
+from tests._utils import ApiTestCase
+from denim.service import supervisor
+
+
+class TestServiceSupervisor(ApiTestCase):
+    def test_manager_start(self):
+        supervisor.manager_start()
+
+        self.assertSudo('/etc/init.d/supervisor start', user=RootUser)
+
+    def test_manager_stop(self):
+        supervisor.manager_stop()
+
+        self.assertSudo('/etc/init.d/supervisor stop', user=RootUser)
+
+    def test_manager_restart(self):
+        supervisor.manager_restart()
+
+        self.assertSudo('/etc/init.d/supervisor restart', user=RootUser)
+
+    def test_manager_reload(self):
+        supervisor.manager_reload()
+
+        self.assertSudo('supervisorctl reread', user=RootUser)
+
+    def test_start_default_service(self):
+        supervisor.start()
+
+        self.assertSudo('supervisorctl start test-project', user=RootUser)
+
+    def test_start_custom_service(self):
+        supervisor.start('bar')
+
+        self.assertSudo('supervisorctl start bar', user=RootUser)
+
+    def test_stop_default_service(self):
+        supervisor.stop()
+
+        self.assertSudo('supervisorctl stop test-project', user=RootUser)
+
+    def test_stop_custom_service(self):
+        supervisor.stop('bar')
+
+        self.assertSudo('supervisorctl stop bar', user=RootUser)
+
+    def test_restart_default_service(self):
+        supervisor.restart()
+
+        self.assertSudo('supervisorctl restart test-project', user=RootUser)
+
+    def test_restart_custom_service(self):
+        supervisor.restart('bar')
+
+        self.assertSudo('supervisorctl restart bar', user=RootUser)
+
+    def test_status_default_service(self):
+        supervisor.status()
+
+        self.assertSudo('supervisorctl status test-project', user=RootUser)
+
+    def test_status_custom_service(self):
+        supervisor.status('bar')
+
+        self.assertSudo('supervisorctl status bar', user=RootUser)
+