Skip to content

Commit

Permalink
Refactor local test fixtures to use patchers
Browse files Browse the repository at this point in the history
  • Loading branch information
caleb531 committed Jun 12, 2015
1 parent d438fe4 commit 34f8284
Show file tree
Hide file tree
Showing 2 changed files with 79 additions and 58 deletions.
53 changes: 39 additions & 14 deletions tests/fixtures/local.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
#!/usr/bin/env python3

import glob
import src.local as swb
import fixtures.shared as shared
from unittest.mock import Mock
import os
import src.local
from unittest.mock import Mock, patch


mock_backups = [
Expand All @@ -15,17 +14,43 @@
]


def before_all():
shared.before_all(swb)
swb.glob.iglob = Mock(return_value=mock_backups)
swb.os.stat = lambda path: Mock(st_mtime=mock_backups.index(path))
swb.input = Mock()
original_stat = os.stat


def before_each():
shared.before_each(swb)
def mock_stat(path):
if path in mock_backups:
return Mock(st_mtime=mock_backups.index(path))
else:
return original_stat(path)


def after_each():
shared.after_each(swb)
swb.input.reset_mock()
patch_makedirs = patch('src.local.os.makedirs').start()
patch_remove = patch('src.local.os.remove')
patch_rmdir = patch('src.local.os.rmdir')
patch_stat = patch('src.local.os.stat', new=mock_stat)
patch_iglob = patch('src.local.glob.iglob', return_value=mock_backups)
src.local.input = input
patch_input = patch('src.local.input')
patch_popen = patch('src.local.subprocess.Popen',
return_value=Mock(returncode=0))


def set_up():
patch_makedirs.start()
patch_remove.start()
patch_rmdir.start()
patch_stat.start()
patch_iglob.start()
patch_input.start()
patch_popen.start()


def tear_down():
patch_makedirs.stop()
patch_remove.stop()
patch_rmdir.stop()
patch_stat.stop()
patch_iglob.stop()
patch_input.stop()
src.local.subprocess.Popen.reset_mock()
patch_popen.stop()
84 changes: 40 additions & 44 deletions tests/test_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,15 @@
import nose.tools as nose
import src.local as swb
from unittest.mock import ANY, mock_open, NonCallableMagicMock, patch
from fixtures.local import before_all, before_each, after_each, mock_backups
from fixtures.local import set_up, tear_down, mock_backups


TEST_CONFIG_PATH = 'tests/files/config.ini'
TEST_BACKUP_PATH = '~/Backups/mysite.sql.bz2'
TEST_BACKUP_PATH_TIMESTAMPED = '~/Backups/%Y/%m/%d/mysite.sql.bz2'


@nose.nottest
def get_test_config():
def get_config():
config = configparser.RawConfigParser()
config.read(TEST_CONFIG_PATH)
return config
Expand All @@ -28,10 +27,10 @@ def test_config_parser():
nose.assert_is_instance(config, configparser.RawConfigParser)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_create_remote_backup():
'''should create remote backup via SSH'''
config = get_test_config()
config = get_config()
swb.back_up(config)
swb.subprocess.Popen.assert_any_call([
'ssh', '-p 2222', 'myname@mysite.com', 'python3', '-', 'back-up',
Expand All @@ -41,126 +40,126 @@ def test_create_remote_backup():
stdin=ANY, stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_download_remote_backup():
'''should download remote backup via SCP'''
config = get_test_config()
config = get_config()
swb.back_up(config)
swb.subprocess.Popen.assert_any_call([
'scp', '-P 2222', 'myname@mysite.com:~/\'backups/mysite.sql.bz2\'',
os.path.expanduser(TEST_BACKUP_PATH)],
stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_create_dir_structure():
'''should create intermediate directories'''
config = get_test_config()
config = get_config()
swb.back_up(config)
swb.os.makedirs.assert_called_with(
os.path.expanduser(os.path.dirname(TEST_BACKUP_PATH)))


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.os.makedirs', side_effect=OSError)
def test_create_dir_structure_silent_fail(makedirs):
'''should fail silently if intermediate directories already exist'''
config = get_test_config()
config = get_config()
swb.back_up(config)
makedirs.assert_called_with(os.path.expanduser(
os.path.dirname(TEST_BACKUP_PATH)))


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_purge_remote_backup():
'''should purge remote backup after download'''
config = get_test_config()
config = get_config()
swb.back_up(config)
swb.subprocess.Popen.assert_any_call([
'ssh', '-p 2222', 'myname@mysite.com', 'python3', '-', 'purge-backup',
'~/\'backups/mysite.sql.bz2\''],
stdin=ANY, stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_purge_oldest_backups():
'''should purge oldest local backups after download'''
config = get_test_config()
config = get_config()
config.set('paths', 'local_backup', TEST_BACKUP_PATH_TIMESTAMPED)
swb.back_up(config)
for path in mock_backups[:-3]:
swb.os.remove.assert_any_call(path)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_null_max_local_backups():
'''should keep all backups if max_local_backups is not set'''
config = get_test_config()
config = get_config()
config.set('paths', 'local_backup', TEST_BACKUP_PATH_TIMESTAMPED)
config.remove_option('backup', 'max_local_backups')
swb.back_up(config)
nose.assert_equal(swb.os.remove.call_count, 0)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_purge_empty_dirs():
'''should purge empty timestamped directories'''
config = get_test_config()
config = get_config()
config.set('paths', 'local_backup', TEST_BACKUP_PATH_TIMESTAMPED)
swb.back_up(config)
for path in mock_backups[:-3]:
swb.os.rmdir.assert_any_call(path)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.os.rmdir', side_effect=OSError)
def test_keep_nonempty_dirs(rmdir):
'''should not purge nonempty timestamped directories'''
config = get_test_config()
config = get_config()
config.set('paths', 'local_backup', TEST_BACKUP_PATH_TIMESTAMPED)
swb.back_up(config)
for path in mock_backups[:-3]:
rmdir.assert_any_call(path)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.back_up')
def test_main_back_up(back_up):
'''should call back_up() when config path is passed to main()'''
config = get_test_config()
config = get_config()
args = [swb.__file__, TEST_CONFIG_PATH]
with patch('src.local.sys.argv', args, create=True):
swb.main()
back_up.assert_called_with(config, stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.shlex')
def test_missing_shlex_quote(shlex):
'''should use pipes.quote() if shlex.quote() is missing (<3.3)'''
del shlex.quote
config = get_test_config()
config = get_config()
swb.back_up(config)
swb.subprocess.Popen.assert_any_call(
# Only check if path is quoted (ignore preceding arguments)
([ANY] * 6) + ['~/\'backups/mysite.sql.bz2\''],
stdin=ANY, stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.sys.exit')
def test_ssh_error(exit):
'''should exit if SSH process returns non-zero exit code'''
config = get_test_config()
config = get_config()
swb.subprocess.Popen.return_value.returncode = 3
swb.back_up(config)
exit.assert_called_with(3)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_quiet_mode():
'''should silence SSH output in quiet mode'''
config = get_test_config()
config = get_config()
args = [swb.__file__, '-q', TEST_CONFIG_PATH]
with patch('src.local.sys.argv', args, create=True):
file_obj = mock_open()
Expand All @@ -172,11 +171,11 @@ def test_quiet_mode():
stderr=devnull)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.restore')
def test_main_restore(restore):
'''should call restore() when config path is passed to main()'''
config = get_test_config()
config = get_config()
args = [swb.__file__, TEST_CONFIG_PATH, '-r', TEST_BACKUP_PATH]
with patch('src.local.sys.argv', args, create=True):
swb.main()
Expand All @@ -185,11 +184,11 @@ def test_main_restore(restore):
stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.restore')
def test_force_mode(restore):
'''should bypass restore confirmation in force mode'''
config = get_test_config()
config = get_config()
args = [swb.__file__, '-f', TEST_CONFIG_PATH, '-r', TEST_BACKUP_PATH]
with patch('src.local.sys.argv', args, create=True):
swb.main()
Expand All @@ -198,11 +197,11 @@ def test_force_mode(restore):
stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
@patch('src.local.input')
def test_restore_confirm_cancel(input):
'''should exit script when user cancels restore confirmation'''
config = get_test_config()
config = get_config()
args = [swb.__file__, TEST_CONFIG_PATH, '-r', TEST_BACKUP_PATH]
with patch('src.local.sys.argv', args, create=True):
responses = ['n', 'N', ' n ', '']
Expand All @@ -212,31 +211,28 @@ def test_restore_confirm_cancel(input):
swb.main()


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_upload_local_backup():
'''should upload local backup to remote for restoration'''
config = get_test_config()
config = get_config()
swb.restore(config, TEST_BACKUP_PATH, stdout=None, stderr=None)
swb.subprocess.Popen.assert_any_call([
'scp', '-P 2222', TEST_BACKUP_PATH,
'myname@mysite.com:~/\'backups/mysite.sql.bz2\''],
stdout=None, stderr=None)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_process_wait_back_up():
'''should wait for each process to finish when backing up'''
config = get_test_config()
config = get_config()
swb.back_up(config)
nose.assert_equal(swb.subprocess.Popen.return_value.wait.call_count, 3)


@nose.with_setup(before_each, after_each)
@nose.with_setup(set_up, tear_down)
def test_process_wait_restore():
'''should wait for each process to finish when restoring'''
config = get_test_config()
config = get_config()
swb.restore(config, TEST_BACKUP_PATH, stdout=None, stderr=None)
nose.assert_equal(swb.subprocess.Popen.return_value.wait.call_count, 2)


before_all()

0 comments on commit 34f8284

Please sign in to comment.