forked from jazzband/django-dbbackup
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request jazzband#73 from ZuluPro/test_dbrestore
Added tests and fix dbrestore command
- Loading branch information
Showing
10 changed files
with
245 additions
and
28 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| import subprocess | ||
| from django.test import TestCase | ||
| from django.utils import six | ||
| from mock import patch | ||
| from dbbackup.management.commands.dbbackup import Command as DbbackupCommand | ||
| from dbbackup.dbcommands import DBCommands | ||
| from dbbackup.tests.utils import FakeStorage, TEST_DATABASE | ||
| from dbbackup.tests.utils import GPG_PUBLIC_PATH, DEV_NULL | ||
|
|
||
|
|
||
| @patch('dbbackup.settings.GPG_RECIPIENT', 'test@test') | ||
| @patch('sys.stdout', DEV_NULL) | ||
| class DbbackupCommandSaveNewBackupTest(TestCase): | ||
| def setUp(self): | ||
| if six.PY3: | ||
| self.skipTest("Compression isn't implemented in Python3") | ||
| open(TEST_DATABASE['NAME'], 'a+b').close() | ||
| self.command = DbbackupCommand() | ||
| self.command.servername = 'foo-server' | ||
| self.command.encrypt = False | ||
| self.command.compress = False | ||
| self.command.database = TEST_DATABASE['NAME'] | ||
| self.command.dbcommands = DBCommands(TEST_DATABASE) | ||
| self.command.storage = FakeStorage() | ||
| self.command.stdout = DEV_NULL | ||
|
|
||
| def test_func(self): | ||
| self.command.save_new_backup(TEST_DATABASE) | ||
|
|
||
| def test_compress(self): | ||
| self.command.compress = True | ||
| self.command.save_new_backup(TEST_DATABASE) | ||
|
|
||
| def test_encrypt(self): | ||
| cmd = ('gpg --import %s' % GPG_PUBLIC_PATH).split() | ||
| subprocess.call(cmd, stdout=DEV_NULL, stderr=DEV_NULL) | ||
| self.command.encrypt = True | ||
| self.command.save_new_backup(TEST_DATABASE) | ||
|
|
||
|
|
||
| @patch('sys.stdout', DEV_NULL) | ||
| class DbbackupCommandCleanupOldBackupsTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbbackupCommand() | ||
| self.command.database = TEST_DATABASE['NAME'] | ||
| self.command.dbcommands = DBCommands(TEST_DATABASE) | ||
| self.command.storage = FakeStorage() | ||
| self.command.clean = True | ||
| self.command.clean_keep = 1 | ||
| self.command.stdout = DEV_NULL | ||
|
|
||
| def test_cleanup_old_backups(self): | ||
| self.command.cleanup_old_backups(TEST_DATABASE) | ||
|
|
||
| def test_cleanup_empty(self): | ||
| self.command.storage.list_files = [] | ||
| self.command.cleanup_old_backups(TEST_DATABASE) | ||
|
|
||
|
|
||
| class DbbackupCommandCompressFileTest(TestCase): | ||
| def setUp(self): | ||
| if six.PY3: | ||
| self.skipTest("Compression isn't implemented in Python3") | ||
| open(TEST_DATABASE['NAME'], 'a+b').close() | ||
| self.command = DbbackupCommand() | ||
| self.command.database = TEST_DATABASE['NAME'] | ||
| self.command.dbcommands = DBCommands(TEST_DATABASE) | ||
| self.command.storage = FakeStorage() | ||
| self.command.stdout = DEV_NULL | ||
|
|
||
| def test_compress_file(self): | ||
| inputfile = open(TEST_DATABASE['NAME']) | ||
| self.command.compress_file(inputfile) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| import subprocess | ||
| from mock import patch | ||
| from django.test import TestCase | ||
| from django.core.management.base import CommandError | ||
| from django.conf import settings | ||
| from django.utils import six | ||
| from dbbackup.management.commands.dbrestore import Command as DbrestoreCommand | ||
| from dbbackup.dbcommands import DBCommands | ||
| from dbbackup.tests.utils import FakeStorage, ENCRYPTED_FILE, TEST_DATABASE | ||
| from dbbackup.tests.utils import GPG_PRIVATE_PATH, DEV_NULL, COMPRESSED_FILE | ||
|
|
||
|
|
||
| @patch('dbbackup.management.commands.dbrestore.input', return_value='y') | ||
| class DbrestoreCommandRestoreBackupTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbrestoreCommand() | ||
| self.command.stdout = DEV_NULL | ||
| self.command.uncompress = False | ||
| self.command.decrypt = False | ||
| self.command.backup_extension = 'bak' | ||
| self.command.filepath = 'foofile' | ||
| self.command.database = TEST_DATABASE | ||
| self.command.dbcommands = DBCommands(TEST_DATABASE) | ||
| self.command.storage = FakeStorage() | ||
|
|
||
| def test_no_filepath(self, *args): | ||
| self.command.storage.list_files = ['foo.bak'] | ||
| self.command.filepath = None | ||
| self.command.restore_backup() | ||
|
|
||
| def test_no_backup_found(self, *args): | ||
| self.command.filepath = None | ||
| with self.assertRaises(CommandError): | ||
| self.command.restore_backup() | ||
|
|
||
| def test_uncompress(self, *args): | ||
| self.command.storage.file_read = COMPRESSED_FILE | ||
| self.command.filepath = COMPRESSED_FILE | ||
| self.command.uncompress = True | ||
| self.command.restore_backup() | ||
|
|
||
| def test_decrypt(self, *args): | ||
| if six.PY3: | ||
| self.skipTest("Decryption isn't implemented in Python3") | ||
| cmd = ('gpg --import %s' % GPG_PRIVATE_PATH).split() | ||
| subprocess.call(cmd, stdout=DEV_NULL, stderr=DEV_NULL) | ||
| self.command.decrypt = True | ||
| self.command.restore_backup() | ||
|
|
||
|
|
||
| class DbrestoreCommandGetDatabaseTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbrestoreCommand() | ||
|
|
||
| def test_give_db_name(self): | ||
| db = self.command._get_database({'database': 'default'}) | ||
| self.assertEqual(db, settings.DATABASES['default']) | ||
|
|
||
| def test_no_given_db(self): | ||
| db = self.command._get_database({}) | ||
| self.assertEqual(db, settings.DATABASES['default']) | ||
|
|
||
| @patch('django.conf.settings.DATABASES', {'db1': {}, 'db2': {}}) | ||
| def test_no_given_db_multidb(self): | ||
| with self.assertRaises(CommandError): | ||
| self.command._get_database({}) | ||
|
|
||
|
|
||
| class DbrestoreCommandGetExtensionTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbrestoreCommand() | ||
|
|
||
| def test_tar(self): | ||
| ext = self.command.get_extension('foo.tar') | ||
| self.assertEqual(ext, '.tar') | ||
|
|
||
| def test_tar_gz(self): | ||
| ext = self.command.get_extension('foo.tar.gz') | ||
| self.assertEqual(ext, '.gz') | ||
|
|
||
| def test_no_extension(self): | ||
| ext = self.command.get_extension('foo') | ||
| self.assertEqual(ext, '') | ||
|
|
||
|
|
||
| class DbrestoreCommandUncompressTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbrestoreCommand() | ||
|
|
||
| def test_uncompress(self): | ||
| inputfile = open(COMPRESSED_FILE, 'rb') | ||
| fd = self.command.uncompress_file(inputfile) | ||
| fd.seek(0) | ||
| self.assertEqual(fd.read(), b'foo\n') | ||
|
|
||
|
|
||
| class DbrestoreCommandDecryptTest(TestCase): | ||
| def setUp(self): | ||
| self.command = DbrestoreCommand() | ||
| cmd = ('gpg --import %s' % GPG_PRIVATE_PATH).split() | ||
| subprocess.call(cmd, stdout=DEV_NULL, stderr=DEV_NULL) | ||
|
|
||
| @patch('dbbackup.management.commands.dbrestore.input', return_value=None) | ||
| def test_decrypt(self, *args): | ||
| if six.PY3: | ||
| self.skipTest("Decryption isn't implemented in Python3") | ||
| inputfile = open(ENCRYPTED_FILE, 'r+b') | ||
| uncryptfile = self.command.unencrypt_file(inputfile) | ||
| uncryptfile.seek(0) | ||
| self.assertEqual('foo\n', uncryptfile.read()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| import os | ||
| from django.conf import settings | ||
| from dbbackup.storage.base import BaseStorage | ||
|
|
||
| BASE_FILE = os.path.join(settings.BASE_DIR, 'tests/test.txt') | ||
| ENCRYPTED_FILE = os.path.join(settings.BASE_DIR, 'tests/test.txt.gpg') | ||
| COMPRESSED_FILE = os.path.join(settings.BASE_DIR, 'tests/test.txt.gz') | ||
| ENCRYPTED_COMPRESSED_FILE = os.path.join(settings.BASE_DIR, 'tests/test.txt.gz.gpg') | ||
| TEST_DATABASE = {'ENGINE': 'django.db.backends.sqlite3', 'NAME': '/tmp/foo.db', 'USER': 'foo', 'PASSWORD': 'bar', 'HOST': 'foo', 'PORT': 122} | ||
|
|
||
| GPG_PRIVATE_PATH = os.path.join(settings.BASE_DIR, 'tests/gpg/secring.gpg') | ||
| GPG_PUBLIC_PATH = os.path.join(settings.BASE_DIR, 'tests/gpg/pubring.gpg') | ||
| DEV_NULL = open(os.devnull, 'w') | ||
|
|
||
|
|
||
| class FakeStorage(BaseStorage): | ||
| name = 'FakeStorage' | ||
| list_files = ['foo', 'bar'] | ||
| deleted_files = [] | ||
| file_read = ENCRYPTED_FILE | ||
|
|
||
| def delete_file(self, filepath): | ||
| self.deleted_files.append(filepath) | ||
|
|
||
| def list_directory(self, raw=False): | ||
| return self.list_files | ||
|
|
||
| def write_file(self, filehandle, filename): | ||
| pass | ||
|
|
||
| def read_file(self, filepath): | ||
| return open(self.file_read, 'rb') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Binary file not shown.
Binary file not shown.