Commits

Tim Savage committed a6bf7b7 Draft

Changes to utils to simplify writing test cases for commands. Added test coverage over the django and django.south modules.
Changes to django calls to make more consistent and some rethought of appropriate user to use.

  • Participants
  • Parent commits f3cce1f

Comments (0)

Files changed (13)

denim/django/__init__.py

 from denim.constants import DeployUser
 
 
-def manage(cmd, args='', revision=None, noinput=True, use_sudo=True, user=None):
+def manage(cmd, args=None, revision=None, noinput=True, use_sudo=True, user=DeployUser):
     """
     Run a django manage.py command.
 
     :param revision: version name that is being worked on.
     :param noinput: Do not ask for input.
     """
+    args = args or []
+    if noinput:
+        args.insert(0, '--noinput')
+    args.insert(0, cmd)
+
     with paths.cd_package(revision):
-        utils.run_as('python manage.py %(cmd)s %(args)s' % {
-            'cmd': cmd,
-            'args': ('--noinput ' if noinput else '') + args
-        }, use_sudo, user)
+        utils.run_as('python manage.py ' + ' '.join(args), use_sudo, user)
 
 
 def test_deploy(revision=None):
     """
     Call manage.py validate to ensure deployment is working correctly.
     """
-    manage('validate', '', revision, noinput=False)
+    manage('validate', revision=revision, noinput=False)
 
 
-def collectstatic(revision=None, use_sudo=True, user=DeployUser):
+def collectstatic(revision=None, user=None):
     """
     Collect static files.
     """
-    manage('collectstatic', '', revision, use_sudo=use_sudo, user=user)
+    manage('collectstatic',
+        revision=revision, user=user)
 
 
-@task
-def syncdb(revision=None, user=DeployUser, *args, **kwargs):
+def syncdb(revision=None, **kwargs):
     """
     Run a database sync
     """
-    manage('syncdb', '', revision, user=user, *args, **kwargs)
+    manage('syncdb', revision=revision, **kwargs)
 
 
-@task
-def createsuperuser(username='', revision=None, user=DeployUser, *args, **kwargs):
+def createsuperuser(username='', revision=None, **kwargs):
     """
     Run a database sync and migrate operation.
     """
-    manage('createsuperuser', username, revision, noinput=False, user=user, *args, **kwargs)
+    manage('createsuperuser', [username], revision=revision, noinput=False, **kwargs)
 
 
-def link_settings(revision=None, use_sudo=True, user=None):
+def link_settings(revision=None, user=None, **kwargs):
     """
     Put correct settings in place.
     """
     system.create_symlink(
         paths.package_path(revision, '%(package_name)s/deployment/settings_%(deploy_env)s.py' % env),
         paths.package_path(revision, '%(package_name)s/local_settings.py' % env),
-        use_sudo=use_sudo, user=user
+        user=user, **kwargs
     )

denim/django/south.py

 __all__ = ('show_migrations', 'migrate',)
 
 
-@task
 def show_migrations(revision=None, non_applied_only=False):
     """
     Print report of migrations.
         else:
             print(colors.magenta('No migrations defined/or not using south.'))
     else:
-        django.manage('migrate', '--list', revision=revision, use_sudo=False)
+        django.manage('migrate', ['--list'], revision=revision, use_sudo=False)
 
 
-@task
-def migrate(revision=None, user=DeployUser, *args, **kwargs):
+def migrate(migration=None, revision=None, user=DeployUser, **kwargs):
     """
     Apply migrations.
 
     :param revision: revision of the application to run show migrations from.
+    :param migration: name of the migration to revert to, leave None to apply
+        all migrations.
 
     """
-    django.manage('migrate', '', revision, user=user, *args, **kwargs)
+    args = [migration] if migration else None
+    django.manage('migrate', args, revision, user=user, **kwargs)
 
 
 
 def install_requirements(revision=None, path_to_requirements=None,
-                         upgrade=True, use_sudo=True, user=None):
+                         upgrade=True, use_sudo=True, user=None,
+                         path_to_log=None):
     """
     Install requirements with PIP.
 
         parameters.append('--proxy=%s' % env.proxy)
     if upgrade:
         parameters.append('--upgrade')
+    if not path_to_log:
+        path_to_log = paths.deploy_path('var/pip.log')
     parameters.append('-r %s' % path_to_requirements)
     utils.run_as('pip ' + ' '.join(parameters), use_sudo, user)
 

denim/scaffold/__init__.py

 
 template_environment = Environment(loader=PackageLoader('denim.scaffold'))
 
+
+def single(template_file, output_name, context):
+    """
+    Generate a single file.
+    :param template_file:
+    :param output_name:
+    :param context:
+    :return:
+
+    """
+    template = template_environment.get_template(template_file)
+    print template.render(**context)
+
+
+def environment(template_file, output_name, context):
+    """
+    Generate multiple files based on the from the env list.
+    :param template_file:
+    :param output_name:
+    :param context:
+    :return:
+
+    """
+    envs = context.env
+    for env in envs:
+        context['env'] = env
+        single(template_file, output_name, context)
+
+
+#   Name: (Template, Target, Generation method, Required parameters)
+SCAFFOLDS = {
+    'nginx': ('nginx.conf.txt', 'conf/nginx/%(env)s.conf', environment, ('env', )),
+    'django.fabric': ('django/fabfile.py.txt', 'fabfile.py', single, ('env', ('scm', 'hg'))),
+    'django.supervisor': ('django/supervisor.conf.txt', 'conf/supervisor.conf', single, None),
+}
+
+
+def generate_scaffold(scaffold_code):
+    scaffold = SCAFFOLDS.get(scaffold_code)
+    if not scaffold:
+        raise NotImplementedError('This scaffold does not exist')
+
+
+
 #template = template_environment.get_template('django/fabfile.py.txt')
 #context = {
 #    'deploy_scm': 'git',
 #    }]
 #}
 #print template.render(**context)
+

denim/scaffold/templates/django/supervisor.conf.txt

+[program:{{ project_name }}.gunicorn]
+command={{ deploy_path }}/gunicorn_django --pid {{ deploy_path }}/var/gunicorn.pid --bind unix:{{ deploy_path }}/var/wsgi.sock --workers 8 --log-file {{ log_path }}/gunicorn.log
+directory={{ directory }}
+#environment=PATH="/var/www/django/bin"
+process_name=%(program_name)s
+user={{ user }}
+autostart=true
+autorestart=true
+redirect_stderr=true

denim/scaffold/templates/supervisor.conf.txt

-[program:{{ project_name }}.gunicorn]
-command={{ deploy_path }}/gunicorn_django -p {{ deploy_path }}/var/gunicorn.pid -b unix:{{ deploy_path }}/var/wsgi.sock -w 8
-directory={{ directory }}
-#environment=PATH="/var/www/django/bin"
-process_name=%(program_name)s
-user={{ user }}
-autostart=true
-autorestart=true
-redirect_stderr=True
 # -*- encoding:utf8 -*-
 from datetime import date
-from fabric.api import run, sudo, settings, hide
+from fabric import api
+from fabric.api import settings, hide
 from denim.constants import RootUser
 
-__all__ = ('run_as', 'generate_version')
+__all__ = ('run_as', 'run_test', 'generate_version')
+
+
+class ApiWrapper(object):
+    """
+    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 run(self, command, *args, **kwargs):
+        return api.run(command, *args, **kwargs)
+
+    def local(self, command, *args, **kwargs):
+        return api.local(command, *args, **kwargs)
+
+
+__api_wrapper = ApiWrapper()
+
+def set_api_wrapper(wrapper):
+    """
+    Hook to replace API wrapper. Used for testing.
+    :param wrapper: Wrapper class to replace with.
+    """
+    global __api_wrapper
+    __api_wrapper = wrapper
+
+
+def api_wrapper():
+    """
+    Get instance of the API wrapper.
+    :return: Current instance of API wrapper.
+    """
+    global __api_wrapper
+    return __api_wrapper
 
 
 def run_as(command, use_sudo=False, user=RootUser, **kwargs):
     :param user: if using sudo run command as this user; default None (root).
 
     """
+    api = api_wrapper()
     if hasattr(user, 'sudo_identity'):
         user = user.sudo_identity()
 
     if use_sudo:
-        return sudo(command, user=user, **kwargs)
+        return api.sudo(command, user=user, **kwargs)
     else:
-        return run(command, **kwargs)
-
-
-def generate_version(increment=None):
-    """
-    Generate a version number based on today's date and an optional increment.
-
-    Version string is in the format %Y.%m.%d.increment.
-
-    """
-    version = date.today().strftime('%Y.%m.%d')
-    if increment:
-        try:
-            version += '.{0}'.format(int(increment))
-        except ValueError:
-            raise ValueError("Increment must be an integer value.")
-    return version
+        return api.run(command, **kwargs)
 
 
 def run_test(command, hide_groups=('warnings', ), use_sudo=False, user=None):
     with settings(warn_only=True):
         with hide(*hide_groups):
             return run_as(command, use_sudo, user)
+
+
+def generate_version(increment=None):
+    """
+    Generate a version number based on today's date and an optional increment.
+
+    Version string is in the format %Y.%m.%d.increment.
+
+    """
+    version = date.today().strftime('%Y.%m.%d')
+    if increment:
+        try:
+            version += '.{0}'.format(int(increment))
+        except ValueError:
+            raise ValueError("Increment must be an integer value.")
+    return version

tests/__init__.py

+# Some default settings
+from fabric.api import env
+
+env.deploy_user = 'test'
+env.project_name = 'test-project'
+env.package_name = 'test_project'
+env.deploy_env = 'test'
+import unittest
+from denim.utils import set_api_wrapper, api_wrapper
+
+
+class TestApiWrapper(object):
+    """
+    An easily replaceable wrapper around api commands to allow for easy
+    testing.
+    """
+    def __init__(self):
+        self.reset()
+
+    def reset(self):
+        """
+        Clear all commands.
+        """
+        self.sudo_commands = []
+        self.run_commands = []
+
+    def sudo(self, command, *args, **kwargs):
+        return self.sudo_commands.append((command, args, kwargs))
+
+    def run(self, command, *args, **kwargs):
+        return self.run_commands.append((command, args, kwargs))
+
+set_api_wrapper(TestApiWrapper())
+
+
+class ApiTestCase(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        super(ApiTestCase, self).__init__(*args, **kwargs)
+        self.api = api_wrapper()
+
+    def setUp(self):
+        self.api.reset()
+
+    def assertCommand(self, expected_command, actual_command):
+        ec, ea, ekwa = expected_command
+        ac, aa, akwa = 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))
+
+    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 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])

tests/django/__init__.py

Empty file added.

tests/django/django.py

+from denim import django
+from denim.constants import DeployUser, RootUser
+from tests._utils import ApiTestCase
+
+
+class TestDjango(ApiTestCase):
+    def test_django_manage_simple_command(self):
+        django.manage('test')
+
+        self.assertSudo('python manage.py test --noinput',
+            user=DeployUser.sudo_identity())
+
+    def test_django_manage_command_with_args(self):
+        django.manage('test', ['--arg1 234', '--arg2 foo'])
+
+        self.assertSudo('python manage.py test --noinput --arg1 234 --arg2 foo',
+            user=DeployUser.sudo_identity())
+
+    def test_django_manage_command_with_input(self):
+        django.manage('test', ['--arg1 234', '--arg2 foo'], noinput=False)
+
+        self.assertSudo('python manage.py test --arg1 234 --arg2 foo',
+            user=DeployUser.sudo_identity())
+
+    def test_django_manage_dont_use_sudo(self):
+        django.manage('test', use_sudo=False)
+
+        self.assertRun('python manage.py test --noinput')
+
+    def test_django_manage_use_sudo_with_user(self):
+        django.manage('test', use_sudo=True, user='dave')
+
+        self.assertSudo('python manage.py test --noinput', user='dave')
+
+    def test_django_test_deploy(self):
+        django.test_deploy()
+
+        self.assertSudo('python manage.py validate',
+            user=DeployUser.sudo_identity())
+
+    def test_django_collectstatic(self):
+        django.collectstatic()
+
+        self.assertSudo('python manage.py collectstatic --noinput',
+            user=RootUser.sudo_identity())
+
+    def test_django_syncdb(self):
+        django.syncdb()
+
+        self.assertSudo('python manage.py syncdb --noinput',
+            user=DeployUser.sudo_identity())
+
+    def test_django_createsuperuser(self):
+        django.createsuperuser()

tests/django/south.py

+from denim.constants import DeployUser
+from denim.django import south
+from tests._utils import ApiTestCase
+
+
+class TestDjangoSouth(ApiTestCase):
+    def test_south_migrate(self):
+        south.migrate()
+
+        self.assertSudo('python manage.py migrate --noinput',
+            user=DeployUser.sudo_identity())
+
+    def test_south_migrate_to_migration(self):
+        south.migrate('0003')
+
+        self.assertSudo('python manage.py migrate --noinput 0003',
+            user=DeployUser.sudo_identity())
+
 
 class PathTestCaseBase(unittest.TestCase):
     def setUp(self):
-        # Required by Denim
-        api.env.project_name = 'test-project'
-        api.env.package_name = 'test_project'
-        api.env.deploy_env = 'test'
-
         # Required for test cases
         api.env.real_fabfile = __file__