diff --git a/freezer/lvm.py b/freezer/lvm.py index b8558791..626f3f7a 100644 --- a/freezer/lvm.py +++ b/freezer/lvm.py @@ -82,9 +82,10 @@ def lvm_snap_remove(backup_opt_dict): shell=True, executable=backup_opt_dict.bash_path) (umount_out, mount_err) = umount_proc.communicate() if re.search(r'\S+', umount_out): - logging.critical('[*] Error: impossible to umount {0} {1}\ - '.format(mount_point, mount_err)) - raise Exception + err = '[*] Error: impossible to umount {0} {1}'.format( + mount_point, mount_err) + logging.critical(err) + raise Exception(err) else: # Change working directory to be able to unmount os.chdir(backup_opt_dict.workdir) @@ -101,10 +102,10 @@ def lvm_snap_remove(backup_opt_dict): logging.info('[*] {0}'.format(lvm_rm_out)) return True else: - logging.critical( - '[*] Error: lvm_snap_rm {0}'.format(lvm_rm_err)) - raise Exception - raise Exception + err = '[*] Error: lvm_snap_rm {0}'.format(lvm_rm_err) + logging.critical(err) + raise Exception(err) + raise Exception('[*] Error: no lvm snap removed') def lvm_snap(backup_opt_dict): @@ -159,9 +160,9 @@ def lvm_snap(backup_opt_dict): executable=backup_opt_dict.bash_path) (lvm_out, lvm_err) = lvm_process.communicate() if lvm_err is False: - logging.critical('[*] lvm snapshot creation error: {0}\ - '.format(lvm_err)) - raise Exception + err = '[*] lvm snapshot creation error: {0}'.format(lvm_err) + logging.critical(err) + raise Exception(err) else: logging.warning('[*] {0}'.format(lvm_out)) @@ -197,12 +198,13 @@ def lvm_snap(backup_opt_dict): '.format(abs_snap_name, backup_opt_dict.lvm_dirmount)) return True if mount_err: - logging.critical('[*] lvm snapshot mounting error: {0}'.format( - mount_err)) - raise Exception + err = '[*] lvm snapshot mounting error: {0}'.format(mount_err) + logging.critical(err) + raise Exception(err) else: - logging.warning('[*] Volume {0} succesfully mounted on {1}\ - '.format(abs_snap_name, backup_opt_dict.lvm_dirmount)) + logging.warning( + '[*] Volume {0} succesfully mounted on {1}'.format( + abs_snap_name, backup_opt_dict.lvm_dirmount)) return True diff --git a/freezer/utils.py b/freezer/utils.py index e9b2d433..c9e5c8ae 100644 --- a/freezer/utils.py +++ b/freezer/utils.py @@ -123,19 +123,20 @@ def validate_all_args(required_list): try: for element in required_list: - if element is False or not element: + if not element: return False except Exception as error: - logging.critical("[*] Error: {0} please provide ALL of the following \ - arguments: {1}".format(error, ' '.join(required_list))) - raise Exception + err = "[*] Error: validate_all_args: {0} {1}".format( + required_list, error) + logging.exception(err) + raise Exception(err) return True def validate_any_args(required_list): ''' - Ensure ANY of the elements of required_list are True. raise ValueError + Ensure ANY of the elements of required_list are True. raise Exception Exception otherwise ''' @@ -143,10 +144,11 @@ def validate_any_args(required_list): for element in required_list: if element: return True - except Exception: - logging.critical("[*] Error: please provide ANY of the following \ - arguments: {0}".format(' '.join(required_list))) - raise Exception + except Exception as error: + err = "[*] Error: validate_any_args: {0} {1}".format( + required_list, error) + logging.exception(err) + raise Exception(err) return False @@ -178,9 +180,10 @@ def create_dir(directory): logging.warning('[*] Directory {0} found!'.format( os.path.expanduser(directory))) except Exception as error: - logging.warning('[*] Error while creating directory {0}: {1}\ - '.format(os.path.expanduser(directory, error))) - raise Exception + err = '[*] Error while creating directory {0}: {1}\ + '.format(os.path.expanduser(directory), error) + logging.exception(err) + raise Exception(err) def get_match_backup(backup_opt_dict): @@ -194,9 +197,10 @@ def get_match_backup(backup_opt_dict): if not backup_opt_dict.backup_name or not backup_opt_dict.container \ or not backup_opt_dict.remote_obj_list: - logging.critical("[*] Error: please provide a valid Swift container,\ - backup name and the container contents") - raise Exception + err = "[*] Error: please provide a valid Swift container,\ + backup name and the container contents" + logging.exception(err) + raise Exception(err) backup_name = backup_opt_dict.backup_name.lower() if backup_opt_dict.remote_obj_list: @@ -252,9 +256,10 @@ def get_rel_oldest_backup(backup_opt_dict): ''' if not backup_opt_dict.backup_name: - logging.critical("[*] Error: please provide a valid backup name in \ - backup_opt_dict.backup_name") - raise Exception + err = "[*] Error: please provide a valid backup name in \ + backup_opt_dict.backup_name" + logging.exception(err) + raise Exception(err) backup_opt_dict.remote_rel_oldest = u'' backup_name = backup_opt_dict.backup_name @@ -286,10 +291,10 @@ def get_abs_oldest_backup(backup_opt_dict): The absolute oldest backup correspond the oldest available level 0 backup. ''' if not backup_opt_dict.backup_name: - - logging.critical("[*] Error: please provide a valid backup name in \ - backup_opt_dict.backup_name") - raise Exception + err = "[*] Error: please provide a valid backup name in \ + backup_opt_dict.backup_name" + logging.exception(err) + raise Exception(err) backup_opt_dict.remote_abs_oldest = u'' if len(backup_opt_dict.remote_match_backup) == 0: @@ -298,9 +303,9 @@ def get_abs_oldest_backup(backup_opt_dict): backup_timestamp = 0 hostname = backup_opt_dict.hostname for remote_obj in backup_opt_dict.remote_match_backup: - object_name = remote_obj.get('name', None) + object_name = remote_obj.get('name', '') obj_name_match = re.search(r'{0}_({1})_(\d+)_(\d+?)$'.format( - hostname, backup_opt_dict.backup_name), remote_obj, re.I) + hostname, backup_opt_dict.backup_name), object_name.lower(), re.I) if not obj_name_match: continue remote_obj_timestamp = int(obj_name_match.group(2)) @@ -332,7 +337,6 @@ def eval_restart_backup(backup_opt_dict): current_timestamp = backup_opt_dict.time_stamp backup_name = backup_opt_dict.backup_name hostname = backup_opt_dict.hostname - first_backup_ts = 0 # Get relative oldest backup by calling get_rel_oldes_backup() backup_opt_dict = get_rel_oldest_backup(backup_opt_dict) if not backup_opt_dict.remote_rel_oldest: @@ -344,9 +348,10 @@ def eval_restart_backup(backup_opt_dict): obj_name_match = re.search(r'{0}_({1})_(\d+)_(\d+?)$'.format( hostname, backup_name), backup_opt_dict.remote_rel_oldest, re.I) if not obj_name_match: - logging.info('[*] No backup match available for backup {0} \ - and host {1}'.format(backup_name, hostname)) - return Exception + err = ('[*] No backup match available for backup {0} ' + 'and host {1}'.format(backup_name, hostname)) + logging.info(err) + return Exception(err) first_backup_ts = int(obj_name_match.group(2)) if (current_timestamp - first_backup_ts) > max_time: @@ -445,9 +450,9 @@ def get_vol_fs_type(backup_opt_dict): vol_name = backup_opt_dict.lvm_srcvol if os.path.exists(vol_name) is False: - logging.critical('[*] Provided volume name not found: {0} \ - '.format(vol_name)) - raise Exception + err = '[*] Provided volume name not found: {0} '.format(vol_name) + logging.exception(err) + raise Exception(err) file_cmd = '{0} -0 -bLs --no-pad --no-buffer --preserve-date \ {1}'.format(backup_opt_dict.file_path, vol_name) @@ -458,17 +463,15 @@ def get_vol_fs_type(backup_opt_dict): (file_out, file_err) = file_process.communicate() file_match = re.search(r'(\S+?) filesystem data', file_out, re.I) if file_match is None: - logging.critical('[*] File system type not guessable: {0}\ - '.format(file_err)) - raise Exception + err = '[*] File system type not guessable: {0}'.format(file_err) + logging.exception(err) + raise (err) else: filesys_type = file_match.group(1) logging.info('[*] File system {0} found for volume {1}'.format( filesys_type, vol_name)) return filesys_type.lower().strip() - raise Exception - def check_backup_existance(backup_opt_dict): ''' @@ -512,9 +515,10 @@ def add_host_name_ts_level(backup_opt_dict, time_stamp=int(time.time())): ''' if backup_opt_dict.backup_name is False: - logging.critical('[*] Error: Please specify the backup name with\ - --backup-name option') - raise Exception + err = ('[*] Error: Please specify the backup name with ' + '--backup-name option') + logging.exception(err) + raise Exception(err) backup_name = u'{0}_{1}_{2}_{3}'.format( backup_opt_dict.hostname, diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/commons.py b/tests/commons.py index 54fa3702..bb8cd8c6 100644 --- a/tests/commons.py +++ b/tests/commons.py @@ -21,6 +21,72 @@ os.environ['OS_USERNAME'] = 'testusername' os.environ['OS_TENANT_NAME'] = 'testtenantename' +class FakeLogging: + + def __init__(self): + return None + + def __call__(self, *args, **kwargs): + return True + + @classmethod + def logging(cls): + return True + + @classmethod + def info(cls): + return True + + +class Fakeget_newest_backup: + + def __init__(self, opt1=True): + return None + + def __call__(self, *args, **kwargs): + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_newest_backup'] = False + return backup_opt + + +class Fakeget_rel_oldest_backup: + + def __init__(self, opt1=True): + return None + + def __call__(self, *args, **kwargs): + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_rel_oldest'] = False + return backup_opt + + +class Fakeget_rel_oldest_backup2: + + def __init__(self, opt1=True): + return None + + def __call__(self, *args, **kwargs): + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_rel_oldest'] = True + return backup_opt + + +class FakeDistutils: + + def __init__(self): + return None + + class spawn: + def __init__(self, *args, **kwargs): + return None + + def __call__(self, *args, **kwargs): + return self + + def find_executable(self, *args, **kwargs): + return True + + class FakeArgparse: def __init__(self): @@ -45,6 +111,7 @@ class FakeArgparse: @classmethod def parse_args(self): + self.hostname = None return self @@ -171,23 +238,42 @@ class FakeSwiftClient: def put_object(self, opt1=True, opt2=True, opt3=True, opt4=True, opt5=True, headers=True, content_length=True, content_type=True): return True + def head_object(self, opt1=True, opt2=True): + return True + class FakeRe: def __init__(self): return None - class search: - def __init__(self, opt1=True, opt2=True, opt3=True): - self.opt1 = opt1 - self.opt2 = opt2 - self.opt3 = opt3 - return None + @classmethod + def search(self, opt1=True, opt2=True, opt3=True): + return self - def group(self, opt1=True, opt2=True): - self.opt1 = opt1 - self.opt2 = opt2 + @classmethod + def group(self, opt1=True, opt2=True): + if opt1 == 1: return 'testgroup' + else: + return '10' + + +class FakeRe2: + + def __init__(self): + return None + + def __call__(self, *args, **kwargs): + return None + + @classmethod + def search(cls, opt1=True, opt2=True, opt3=True): + return None + + @classmethod + def group(cls, opt1=True, opt2=True): + return None class BackupOpt1: @@ -196,7 +282,7 @@ class BackupOpt1: fakeclient = FakeSwiftClient() fakeconnector = fakeclient.client() fakeswclient = fakeconnector.Connection() - self.mysql_conf_file = '/dev/null' + self.mysql_conf_file = '/tmp/freezer-test-conf-file' self.lvm_auto_snap = '/dev/null' self.lvm_volgroup = 'testgroup' self.lvm_srcvol = 'testvol' @@ -212,7 +298,7 @@ class BackupOpt1: self.umount_path = 'true' self.backup_name = 'test-backup-name' self.hostname = 'test-hostname' - self.curr_backup_level = '0' + self.curr_backup_level = 0 self.src_file = '/tmp' self.tar_path= 'true' self.dereference_symlink = 'true' @@ -224,13 +310,29 @@ class BackupOpt1: self.max_backup_level = '0' self.remove_older_than = '0' self.max_seg_size = '0' - self.time_stamp = '0' + self.time_stamp = 123456789 self.container_segments = 'test-container-segements' self.container = 'test-container' - self.restart_always_backup = '0' self.workdir = '/tmp' self.upload = 'true' self.sw_connector = fakeswclient + self.max_backup_level = '20' + self.encrypt_pass_file = '/dev/random' + self.always_backup_level = '20' + self.remove_older_than = '20' + self.restart_always_backup = 100000 + self.container_segments = 'testcontainerseg' + self.remote_match_backup = [ + 'test-hostname_test-backup-name_1234567_0', + 'test-hostname_test-backup-name_1234567_1', + 'test-hostname_test-backup-name_1234567_2'] + self.remote_obj_list = [ + {'name' : 'test-hostname_test-backup-name_1234567_0'}, + {'name' : 'test-hostname_test-backup-name_1234567_1'}, + {'name' : 'test-hostname_test-backup-name_1234567_2'}, + {'fakename' : 'test-hostname_test-backup-name_1234567_2'}, + {'name' : 'test-hostname-test-backup-name-asdfa-asdfasdf'}] + self.remote_objects = [] class FakeMySQLdb: @@ -262,6 +364,16 @@ class FakeMySQLdb: return True +class FakeMySQLdb2: + + def __init__(self): + return None + + @classmethod + def connect(self, host=True, user=True, passwd=True): + raise Exception + + class FakeMongoDB: def __init__(self, opt1=True): @@ -279,18 +391,38 @@ class FakeMongoDB: return {'me': 'testnode', 'primary': 'testnode'} -class Os: - def __init__(self): +class FakeMongoDB2: + + def __init__(self, opt1=True): return None - def expanduser(self, directory=True): - return True + def __call__(self, opt1=True): + return self + + class admin: + def __init__(self): + return None + + @classmethod + def command(cls, opt1=True): + return {'me': 'testnode', 'primary': 'testanothernode'} + + +class Os: + def __init__(self, directory=True): + return None + + def expanduser(self, directory=True, opt2=True): + return 'testdir' def makedirs(self, directory=True): - return True + return 'testdir' def isdir(self, directory=True): - return True + return 'testdir' def exists(self, directory=True): - return True + return 'testdir' + + def makedirs2(self, directory=True): + raise Exception diff --git a/tests/test_arguments.py b/tests/test_arguments.py index 54cbc758..2d76b16c 100644 --- a/tests/test_arguments.py +++ b/tests/test_arguments.py @@ -3,7 +3,9 @@ from freezer.arguments import backup_arguments import argparse from commons import * - +import sys +import pytest +import distutils.spawn as distspawn import __builtin__ @@ -13,8 +15,18 @@ class TestArguments: fakeargparse = FakeArgparse() fakeargparse = fakeargparse.ArgumentParser() + fakedistutils = FakeDistutils() + fakedistutilsspawn = fakedistutils.spawn() monkeypatch.setattr( argparse, 'ArgumentParser', fakeargparse) - assert backup_arguments() is not Exception or not False + platform = sys.platform + assert backup_arguments() is not False + + sys.__dict__['platform'] = 'darwin' + pytest.raises(Exception, backup_arguments) + monkeypatch.setattr( + distspawn, 'find_executable', fakedistutilsspawn.find_executable) + assert backup_arguments() is not False + sys.__dict__['platform'] = platform \ No newline at end of file diff --git a/tests/test_backup.py b/tests/test_backup.py index 40e71ce1..b42af266 100644 --- a/tests/test_backup.py +++ b/tests/test_backup.py @@ -10,6 +10,7 @@ import os import MySQLdb import pymongo import re +import pytest from commons import * import __builtin__ @@ -28,7 +29,6 @@ class TestBackUP: fakelvm = Lvm() fakesubprocess = FakeSubProcess() fakesubprocesspopen = fakesubprocess.Popen() - fakeopen = FakeOpen() fakemultiprocessing = FakeMultiProcessing() fakemultiprocessingqueue = fakemultiprocessing.Queue() fakemultiprocessingpipe = fakemultiprocessing.Pipe() @@ -42,7 +42,7 @@ class TestBackUP: multiprocessing, 'Process', fakemultiprocessing.Process) monkeypatch.setattr( multiprocessing, '__init__', fakemultiprocessinginit) - monkeypatch.setattr(__builtin__, 'open', fakeopen.open) + #monkeypatch.setattr(__builtin__, 'open', fakeopen.open) monkeypatch.setattr( subprocess.Popen, 'communicate', fakesubprocesspopen.communicate) monkeypatch.setattr( @@ -56,9 +56,21 @@ class TestBackUP: monkeypatch.setattr(os.path, 'exists', expanduser.exists) monkeypatch.setattr(swiftclient, 'client', fakeswiftclient.client) - assert backup_mode_mysql( - backup_opt, int(time.time()), test_meta) is None + mysql_conf_file = backup_opt.mysql_conf_file + backup_opt.__dict__['mysql_conf_file'] = None + pytest.raises(Exception, backup_mode_mysql, backup_opt, 123456789, test_meta) + # Generate mysql conf test file + backup_opt.__dict__['mysql_conf_file'] = mysql_conf_file + with open(backup_opt.mysql_conf_file, 'w') as mysql_conf_fd: + mysql_conf_fd.write('host=abcd\nuser=abcd\npassword=abcd\n') + assert backup_mode_mysql( + backup_opt, 123456789, test_meta) is None + #os.unlink(backup_opt.mysql_conf_file) + + fakemysql2 = FakeMySQLdb2() + monkeypatch.setattr(MySQLdb, 'connect', fakemysql2.connect) + pytest.raises(Exception, backup_mode_mysql, backup_opt, 123456789, test_meta) def test_backup_mode_fs(self, monkeypatch): @@ -89,7 +101,14 @@ class TestBackUP: monkeypatch.setattr(os.path, 'exists', expanduser.exists) assert backup_mode_fs( - backup_opt, int(time.time()), test_meta) is None + backup_opt, 123456789, test_meta) is None + + backup_opt.__dict__['no_incremental'] = False + with open( + '/tmp/tar_metadata_test-hostname_test-backup-name_123456789_0', 'w') as fd: + fd.write('testcontent\n') + assert backup_mode_fs( + backup_opt, 123456789, test_meta) is None def test_backup_mode_mongo(self, monkeypatch): @@ -102,7 +121,7 @@ class TestBackUP: fakeos = Os() fakere = FakeRe() fakeswiftclient = FakeSwiftClient() - fakeopen = FakeOpen() + #fakeopen = FakeOpen() fakelvm = Lvm() fakemultiprocessing = FakeMultiProcessing() fakemultiprocessingqueue = fakemultiprocessing.Queue() @@ -122,7 +141,12 @@ class TestBackUP: monkeypatch.setattr(os.path, 'exists', fakeos.exists) monkeypatch.setattr(re, 'search', fakere.search) monkeypatch.setattr(swiftclient, 'client', fakeswiftclient.client) - monkeypatch.setattr(__builtin__, 'open', fakeopen.open) + #monkeypatch.setattr(__builtin__, 'open', fakeopen.open) assert backup_mode_mongo( - backup_opt, int(time.time()), test_meta) is None + backup_opt, 123456789, test_meta) is None + + fakemongo2 = FakeMongoDB2() + monkeypatch.setattr(pymongo, 'MongoClient', fakemongo2) + assert backup_mode_mongo( + backup_opt, 123456789, test_meta) is True diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 00000000..d31c7e00 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python + +from freezer.utils import ( + gen_manifest_meta, validate_all_args, validate_any_args, + sort_backup_list, create_dir, get_match_backup, + get_newest_backup, get_rel_oldest_backup, get_abs_oldest_backup, + eval_restart_backup, start_time, elapsed_time, set_backup_level, + get_vol_fs_type, check_backup_existance, add_host_name_ts_level, + get_mount_from_path) + +from freezer import utils +import pytest +import argparse +import os +import datetime +import re +from commons import * + + +class TestUtils: + + def test_gen_manifest_meta(self): + + backup_opt = BackupOpt1() + manifest_meta = {} + + gen_manifest_meta( + backup_opt, manifest_meta, meta_data_backup_file='testfile') + + manifest_meta['x-object-meta-tar-meta-obj-name'] = 'testtar' + gen_manifest_meta( + backup_opt, manifest_meta, meta_data_backup_file='testfile') + del manifest_meta['x-object-meta-tar-meta-obj-name'] + + manifest_meta['x-object-meta-tar-prev-meta-obj-name'] = 'testtar' + gen_manifest_meta( + backup_opt, manifest_meta, meta_data_backup_file='testfile') + del manifest_meta['x-object-meta-tar-prev-meta-obj-name'] + + backup_opt.__dict__['encrypt_pass_file'] = False + gen_manifest_meta( + backup_opt, manifest_meta, meta_data_backup_file='testfile') + + def test_validate_all_args(self): + + elements1 = ['test1', 'test2', 'test3'] + elements2 = ['test1', '', False, None] + elements3 = None + + assert validate_all_args(elements1) is True + assert validate_all_args(elements2) is False + pytest.raises(Exception, validate_all_args, elements3) + + def test_validate_any_args(self): + + elements1 = ['test1', 'test2', 'test3'] + elements2 = [None, None, False, None] + elements3 = None + + assert validate_any_args(elements1) is True + assert validate_any_args(elements2) is False + pytest.raises(Exception, validate_any_args, elements3) + + def test_sort_backup_list(self): + + backup_opt = BackupOpt1() + + assert type(sort_backup_list(backup_opt)) is list + + def test_create_dir(self, monkeypatch): + + dir1 = '/tmp' + dir2 = '/tmp/testnoexistent1234' + dir3 = '~' + fakeos = Os() + + assert create_dir(dir1) is None + assert create_dir(dir2) is None + os.rmdir(dir2) + assert create_dir(dir3) is None + monkeypatch.setattr(os, 'makedirs', fakeos.makedirs2) + pytest.raises(Exception, create_dir, dir2) + + def test_get_match_backup(self): + + backup_opt = BackupOpt1() + + backup_opt = get_match_backup(backup_opt) + assert len(backup_opt.remote_match_backup) > 0 + + backup_opt.__dict__['backup_name'] = '' + pytest.raises(Exception, get_match_backup, backup_opt) + + def test_get_newest_backup(self, monkeypatch): + + backup_opt = BackupOpt1() + backup_opt = get_newest_backup(backup_opt) + assert len(backup_opt.remote_newest_backup) > 0 + + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_match_backup'] = '' + backup_opt = get_newest_backup(backup_opt) + assert backup_opt.remote_match_backup is not True + + backup_opt = BackupOpt1() + fakere2 = FakeRe2() + monkeypatch.setattr(re, 'search', fakere2.search) + backup_opt = get_newest_backup(backup_opt) + assert backup_opt.remote_match_backup is not True + + def test_get_rel_oldest_backup(self): + + backup_opt = BackupOpt1() + backup_opt = get_rel_oldest_backup(backup_opt) + assert len(backup_opt.remote_rel_oldest) > 0 + + backup_opt.__dict__['backup_name'] = '' + pytest.raises(Exception, get_rel_oldest_backup, backup_opt) + + def test_get_abs_oldest_backup(self): + + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_match_backup'] = [] + backup_opt = get_abs_oldest_backup(backup_opt) + assert len(backup_opt.remote_abs_oldest) == 0 + + backup_opt = BackupOpt1() + backup_opt.__dict__['remote_match_backup'] = backup_opt.remote_obj_list + backup_opt = get_abs_oldest_backup(backup_opt) + assert len(backup_opt.remote_abs_oldest) > 0 + + backup_opt = BackupOpt1() + backup_opt.__dict__['backup_name'] = '' + pytest.raises(Exception, get_abs_oldest_backup, backup_opt) + + def test_eval_restart_backup(self, monkeypatch): + + backup_opt = BackupOpt1() + assert eval_restart_backup(backup_opt) is False + + backup_opt.__dict__['restart_always_backup'] = None + assert eval_restart_backup(backup_opt) is False + + backup_opt = BackupOpt1() + fake_get_rel_oldest_backup = Fakeget_rel_oldest_backup() + monkeypatch.setattr(utils, 'get_rel_oldest_backup', fake_get_rel_oldest_backup) + assert eval_restart_backup(backup_opt) is False + + backup_opt = BackupOpt1() + fake_get_rel_oldest_backup2 = Fakeget_rel_oldest_backup2() + monkeypatch.setattr(utils, 'get_rel_oldest_backup', fake_get_rel_oldest_backup2) + fakere2 = FakeRe2() + monkeypatch.setattr(re, 'search', fakere2.search) + assert eval_restart_backup(backup_opt) is not None + #pytest.raises(Exception, eval_restart_backup, backup_opt) + + + def test_start_time(self): + + (time_stamp, day_time) = start_time() + assert type(day_time) is datetime.datetime + assert type(time_stamp) is int + + def test_elapsed_time(self): + + (time_stamp, day_time) = start_time() + assert elapsed_time(day_time) is None + + def test_set_backup_level(self): + + manifest_meta = dict() + backup_opt = BackupOpt1() + manifest_meta['x-object-meta-backup-name'] = True + manifest_meta['x-object-meta-backup-current-level'] = 1 + manifest_meta['x-object-meta-always-backup-level'] = 3 + manifest_meta['x-object-meta-restart-always-backup'] = 3 + + (backup_opt, manifest_meta_dict) = set_backup_level( + backup_opt, manifest_meta) + assert manifest_meta['x-object-meta-backup-current-level'] is not False + + backup_opt = BackupOpt1() + manifest_meta['x-object-meta-maximum-backup-level'] = 2 + (backup_opt, manifest_meta_dict) = set_backup_level( + backup_opt, manifest_meta) + assert manifest_meta['x-object-meta-backup-current-level'] is not False + + backup_opt = BackupOpt1() + backup_opt.__dict__['curr_backup_level'] = 1 + (backup_opt, manifest_meta_dict) = set_backup_level( + backup_opt, manifest_meta) + assert manifest_meta['x-object-meta-backup-current-level'] is not False + + manifest_meta = dict() + backup_opt = BackupOpt1() + manifest_meta['x-object-meta-backup-name'] = False + manifest_meta['x-object-meta-maximum-backup-level'] = 0 + manifest_meta['x-object-meta-backup-current-level'] = 1 + (backup_opt, manifest_meta) = set_backup_level( + backup_opt, manifest_meta) + assert manifest_meta['x-object-meta-backup-current-level'] == '0' + + def test_get_vol_fs_type(self, monkeypatch): + + backup_opt = BackupOpt1() + pytest.raises(Exception, get_vol_fs_type, backup_opt) + + fakeos = Os() + monkeypatch.setattr(os.path, 'exists', fakeos.exists) + #fakesubprocess = FakeSubProcess() + pytest.raises(Exception, get_vol_fs_type, backup_opt) + + fakere = FakeRe() + monkeypatch.setattr(re, 'search', fakere.search) + assert type(get_vol_fs_type(backup_opt)) is str + + def test_check_backup_existance(self, monkeypatch): + + backup_opt = BackupOpt1() + backup_opt.__dict__['backup_name'] = None + assert type(check_backup_existance(backup_opt)) is dict + + fakeswiftclient = FakeSwiftClient() + backup_opt = BackupOpt1() + assert check_backup_existance(backup_opt) is True + + fake_get_newest_backup = Fakeget_newest_backup() + monkeypatch.setattr(utils, 'get_newest_backup', fake_get_newest_backup) + assert type(check_backup_existance(backup_opt)) is dict + + def test_add_host_name_ts_level(self): + + backup_opt = BackupOpt1() + backup_opt.__dict__['backup_name'] = False + pytest.raises(Exception, add_host_name_ts_level, backup_opt) + + backup_opt = BackupOpt1() + assert type(add_host_name_ts_level(backup_opt)) is unicode + + def test_get_mount_from_path(self): + + dir1 = '/tmp' + dir2 = '/tmp/nonexistentpathasdf' + assert type(get_mount_from_path(dir1)) is str + pytest.raises(Exception, get_mount_from_path, dir2) diff --git a/tox.ini b/tox.ini index f343bd2f..cf102ef0 100644 --- a/tox.ini +++ b/tox.ini @@ -13,7 +13,7 @@ deps = install_command = pip install -U {opts} {packages} setenv = VIRTUAL_ENV={envdir} -commands = py.test -v -n 2 +commands = python runtests.py -v -n 2 --cov-report term-missing --cov freezer [pytest] python_files = test_*.py