Merge pull request jazzband#78 from ZuluPro/basecommand

Added base command class & fixed tests for dbbackup command

dbbackup/management/commands/_base.py

@@ -0,0 +1,18 @@
+from optparse import make_option
+from django.core.management.base import BaseCommand, LabelCommand
+
+
+class BaseDbBackupCommand(LabelCommand):
+    option_list = BaseCommand.option_list + (
+        make_option("--noinput", action='store_false', dest='interactive', default=True,
+                    help='Tells Django to NOT prompt the user for input of any kind.'),
+        make_option('-q', "--quiet", action='store_true', default=False,
+                    help='Tells Django to NOT output other text than errors.')
+    )
+
+    verbosity = 1
+    quiet = False
+
+    def log(self, msg, level):
+        if not self.quiet and self.verbosity >= level:
+            self.stdout.write(msg)

dbbackup/management/commands/dbbackup.py

@@ -1,5 +1,5 @@
 """
-Save backup files to Dropbox.
+Save database.
 """
 from __future__ import (absolute_import, division,
                         print_function, unicode_literals)
@@ -9,22 +9,22 @@
 import tempfile
 import gzip
 from shutil import copyfileobj
+
 from django.conf import settings
-from django.core.management.base import BaseCommand
 from django.core.management.base import CommandError
-from django.core.management.base import LabelCommand
 from optparse import make_option
 
+from dbbackup.management.commands._base import BaseDbBackupCommand
 from dbbackup import utils
 from dbbackup.dbcommands import DBCommands
 from dbbackup.storage.base import BaseStorage
 from dbbackup.storage.base import StorageError
 from dbbackup import settings as dbbackup_settings
 
 
-class Command(LabelCommand):
+class Command(BaseDbBackupCommand):
     help = "dbbackup [-c] [-d <dbname>] [-s <servername>] [--compress] [--encrypt]"
-    option_list = BaseCommand.option_list + (
+    option_list = BaseDbBackupCommand.option_list + (
         make_option("-c", "--clean", help="Clean up old backup files", action="store_true", default=False),
         make_option("-d", "--database", help="Database to backup (default: everything)"),
         make_option("-s", "--servername", help="Specify server name to include in backup filename"),
@@ -35,6 +35,8 @@ class Command(LabelCommand):
     @utils.email_uncaught_exception
     def handle(self, **options):
         """ Django command handler. """
+        self.verbosity = options.get('verbosity')
+        self.quiet = options.get('quiet')
         try:
             self.clean = options.get('clean')
             self.clean_keep = getattr(settings, 'DBBACKUP_CLEANUP_KEEP', 10)
@@ -55,39 +57,40 @@ def handle(self, **options):
         except StorageError as err:
             raise CommandError(err)
 
-    def save_new_backup(self, database, database_name):
+    def save_new_backup(self, database):
         """ Save a new backup file. """
-        print("Backing Up Database: %s" % database['NAME'])
+        self.log("Backing Up Database: %s" % database['NAME'], 1)
         filename = self.dbcommands.filename(self.servername)
         outputfile = tempfile.SpooledTemporaryFile(
             max_size=10 * 1024 * 1024,
             dir=dbbackup_settings.TMP_DIR)
         self.dbcommands.run_backup_commands(outputfile)
+        outputfile.name = filename
         if self.compress:
             compressed_file = self.compress_file(outputfile)
             outputfile.close()
             outputfile = compressed_file
         if self.encrypt:
             encrypted_file = utils.encrypt_file(outputfile)
             outputfile = encrypted_file
-        print("  Backup tempfile created: %s" % (utils.handle_size(outputfile)))
-        print("  Writing file to %s: %s, filename: %s" % (self.storage.name, self.storage.backup_dir, filename))
+        self.log("  Backup tempfile created: %s" % (utils.handle_size(outputfile)), 1)
+        self.log("  Writing file to %s: %s, filename: %s" % (self.storage.name, self.storage.backup_dir, filename), 1)
         self.storage.write_file(outputfile, filename)
 
     def cleanup_old_backups(self, database):
         """ Cleanup old backups, keeping the number of backups specified by
             DBBACKUP_CLEANUP_KEEP and any backups that occur on first of the month.
         """
         if self.clean:
-            print("Cleaning Old Backups for: %s" % database['NAME'])
+            self.log("Cleaning Old Backups for: %s" % database['NAME'], 1)
             filepaths = self.storage.list_directory()
             filepaths = self.dbcommands.filter_filepaths(filepaths)
             for filepath in sorted(filepaths[0:-self.clean_keep]):
                 regex = r'^%s' % self.dbcommands.filename_match(self.servername, '(.*?)')
                 datestr = re.findall(regex, os.path.basename(filepath))[0]
                 dateTime = datetime.datetime.strptime(datestr, dbbackup_settings.DATE_FORMAT)
                 if int(dateTime.strftime("%d")) != 1:
-                    print("  Deleting: %s" % filepath)
+                    self.log("  Deleting: %s" % filepath, 1)
                     self.storage.delete_file(filepath)
 
     def compress_file(self, inputfile):

dbbackup/management/commands/dbrestore.py

@@ -1,6 +1,5 @@
 """
-Restore pgdump files from Dropbox.
-See __init__.py for a list of options.
+Restore database.
 """
 from __future__ import (absolute_import, division,
                         print_function, unicode_literals)
@@ -10,26 +9,25 @@
 import sys
 from getpass import getpass
 
-from dbbackup import utils
-from dbbackup.dbcommands import DBCommands
-from dbbackup.storage.base import BaseStorage, StorageError
-from dbbackup import settings as dbbackup_settings
-
 from django.conf import settings
-from django.core.management.base import BaseCommand
 from django.core.management.base import CommandError
-from django.core.management.base import LabelCommand
 from django.utils import six
 from django.db import connection
 
 from optparse import make_option
 
+from dbbackup.management.commands._base import BaseDbBackupCommand
+from dbbackup import utils
+from dbbackup.dbcommands import DBCommands
+from dbbackup.storage.base import BaseStorage, StorageError
+from dbbackup import settings as dbbackup_settings
+
 input = raw_input if six.PY2 else input  # @ReservedAssignment
 
 
-class Command(LabelCommand):
+class Command(BaseDbBackupCommand):
     help = "dbrestore [-d <dbname>] [-f <filename>] [-s <servername>]"
-    option_list = BaseCommand.option_list + (
+    option_list = BaseDbBackupCommand.option_list + (
         make_option("-d", "--database", help="Database to restore"),
         make_option("-f", "--filepath", help="Specific file to backup from"),
         make_option("-x", "--backup-extension", help="The extension to use when scanning for files to restore from."),
@@ -42,6 +40,8 @@ class Command(LabelCommand):
 
     def handle(self, **options):
         """ Django command handler. """
+        self.verbosity = options.get('verbosity')
+        self.quiet = options.get('quiet')
         try:
             connection.close()
             self.filepath = options.get('filepath')
@@ -72,17 +72,17 @@ def _get_database(self, options):
 
     def restore_backup(self):
         """ Restore the specified database. """
-        self.stdout.write("Restoring backup for database: %s" % self.database['NAME'])
+        self.log("Restoring backup for database: %s" % self.database['NAME'], 1)
         # Fetch the latest backup if filepath not specified
         if not self.filepath:
-            self.stdout.write("  Finding latest backup")
+            self.log("  Finding latest backup", 1)
             filepaths = self.storage.list_directory()
             filepaths = [f for f in filepaths if f.endswith('.' + self.backup_extension)]
             if not filepaths:
                 raise CommandError("No backup files found in: /%s" % self.storage.backup_dir)
             self.filepath = filepaths[-1]
         # Restore the specified filepath backup
-        self.stdout.write("  Restoring: %s" % self.filepath)
+        self.log("  Restoring: %s" % self.filepath, 1)
         input_filename = self.filepath
         inputfile = self.storage.read_file(input_filename)
         if self.decrypt:
@@ -94,10 +94,10 @@ def restore_backup(self):
             uncompressed_file = self.uncompress_file(inputfile)
             inputfile.close()
             inputfile = uncompressed_file
-        self.stdout.write("  Restore tempfile created: %s" % utils.handle_size(inputfile))
+        self.log("  Restore tempfile created: %s" % utils.handle_size(inputfile), 1)
         answer = input("Are you sure you want to continue? [Y/n]")
         if answer.lower() not in ('y', 'yes', ''):
-            self.stdout.write("Quitting")
+            self.log("Quitting", 1)
             sys.exit(0)
         inputfile.seek(0)
         self.dbcommands.run_restore_commands(inputfile)
@@ -155,8 +155,8 @@ def get_passphrase():
 
     def list_backups(self):
         """ List backups in the backup directory. """
-        self.stdout.write("Listing backups on %s in /%s:" % (self.storage.name, self.storage.backup_dir))
+        self.log("Listing backups on %s in /%s:" % (self.storage.name, self.storage.backup_dir), 1)
         for filepath in self.storage.list_directory():
-            self.stdout.write("  %s" % os.path.basename(filepath))
+            self.log("  %s" % os.path.basename(filepath), 1)
             # TODO: Implement filename_details method
             # print(utils.filename_details(filepath))

dbbackup/management/commands/mediabackup.py

@@ -1,3 +1,6 @@
+"""
+Save media files.
+"""
 from __future__ import (absolute_import, division,
                         print_function, unicode_literals)
 import os
@@ -9,28 +12,23 @@
 import re
 
 from django.conf import settings
-from django.core.management.base import BaseCommand
 from django.core.management.base import CommandError
 
+from dbbackup.management.commands._base import BaseDbBackupCommand
 from dbbackup import utils
 from dbbackup.storage.base import BaseStorage
 from dbbackup.storage.base import StorageError
 from dbbackup import settings as dbbackup_settings
 
-class Command(BaseCommand):
+
+class Command(BaseDbBackupCommand):
     help = "backup_media [--encrypt] [--clean] [--no-compress] " \
     "--servername SERVER_NAME"
-    option_list = BaseCommand.option_list + (
+    option_list = BaseDbBackupCommand.option_list + (
         make_option("-c", "--clean", help="Clean up old backup files", action="store_true", default=False),
         make_option("-s", "--servername", help="Specify server name to include in backup filename"),
         make_option("-e", "--encrypt", help="Encrypt the backup files", action="store_true", default=False),
-        make_option(
-            "-x",
-            "--no-compress",
-            help="Do not compress the archive",
-            action="store_true",
-            default=False
-        ),
+        make_option("-x", "--no-compress", help="Do not compress the archive", action="store_true", default=False),
     )
 
     @utils.email_uncaught_exception
@@ -52,9 +50,9 @@ def handle(self, *args, **options):
     def backup_mediafiles(self, encrypt, compress):
         source_dir = self.get_source_dir()
         if not source_dir:
-            print("No media source dir configured.")
+            self.stderr.write("No media source dir configured.")
             sys.exit(0)
-        print("Backing up media files in %s" % source_dir)
+        self.log("Backing up media files in %s" % source_dir, 1)
         output_file = self.create_backup_file(
             source_dir,
             self.get_backup_basename(
@@ -67,8 +65,8 @@ def backup_mediafiles(self, encrypt, compress):
             encrypted_file = utils.encrypt_file(output_file)
             output_file = encrypted_file
 
-        print("  Backup tempfile created: %s (%s)" % (output_file.name, utils.handle_size(output_file)))
-        print("  Writing file to %s: %s" % (self.storage.name, self.storage.backup_dir))
+        self.log("  Backup tempfile created: %s (%s)" % (output_file.name, utils.handle_size(output_file)), 1)
+        self.log("  Writing file to %s: %s" % (self.storage.name, self.storage.backup_dir), 1)
         self.storage.write_file(
             output_file,
             self.get_backup_basename(
@@ -120,13 +118,13 @@ def cleanup_old_backups(self):
         """ Cleanup old backups, keeping the number of backups specified by
         DBBACKUP_CLEANUP_KEEP and any backups that occur on first of the month.
         """
-        print("Cleaning Old Backups for media files")
+        self.log("Cleaning Old Backups for media files", 1)
 
         file_list = self.get_backup_file_list()
 
         for backup_date, filename in file_list[0:-dbbackup_settings.CLEANUP_KEEP_MEDIA]:
             if int(backup_date.strftime("%d")) != 1:
-                print("  Deleting: %s" % filename)
+                self.log("  Deleting: %s" % filename, 1)
                 self.storage.delete_file(filename)
 
     def get_backup_file_list(self):

dbbackup/tests/commands/test_base.py

@@ -0,0 +1,31 @@
+try:
+    from StringIO import StringIO
+except ImportError:  # Py3
+    from io import StringIO
+from django.test import TestCase
+from dbbackup.management.commands._base import BaseDbBackupCommand
+
+
+class BaseDbBackupCommandLogTest(TestCase):
+    def setUp(self):
+        self.command = BaseDbBackupCommand()
+        self.command.stdout = StringIO()
+
+    def test_less_level(self):
+        self.command.verbosity = 1
+        self.command.log("foo", 2)
+        self.command.stdout.seek(0)
+        self.assertFalse(self.command.stdout.read())
+
+    def test_more_level(self):
+        self.command.verbosity = 1
+        self.command.log("foo", 0)
+        self.command.stdout.seek(0)
+        self.assertEqual('foo', self.command.stdout.read())
+
+    def test_quiet(self):
+        self.command.quiet = True
+        self.command.verbosity = 1
+        self.command.log("foo", 0)
+        self.command.stdout.seek(0)
+        self.assertFalse(self.command.stdout.read())

tests/runtests.py

@@ -42,6 +42,7 @@ def main():
         call_command('shell')
         sys.exit(0)
     else:
+        # call_command('test', *sys.argv[2:])
         from django.test.runner import DiscoverRunner
         runner = DiscoverRunner(failfast=True, verbosity=int(os.environ.get('DJANGO_DEBUG', 1)))
         failures = runner.run_tests(['dbbackup'], interactive=True)