Fix tests

Change-Id: I85a270c5fb20393b9b7769e9539cf0d55b6306b1
This commit is contained in:
eldar nugaev 2015-11-16 14:31:35 +00:00
parent 3c79e9d07c
commit f98e36c34c
11 changed files with 174 additions and 3290 deletions

View File

@ -390,8 +390,7 @@ def get_executable_path(binary):
path_to_binaries = os.path.dirname(os.path.abspath(__file__)) path_to_binaries = os.path.dirname(os.path.abspath(__file__))
return '{0}\\bin\\{1}.exe'.format(path_to_binaries, binary) return '{0}\\bin\\{1}.exe'.format(path_to_binaries, binary)
elif 'darwin' in sys.platform or 'bsd' in sys.platform: elif is_bsd():
return (distspawn.find_executable(binary) or return (distspawn.find_executable(binary) or
distspawn.find_executable(binary, path=':'.join(sys.path))) distspawn.find_executable(binary, path=':'.join(sys.path)))
else: else:
@ -418,3 +417,6 @@ def alter_proxy(proxy):
os.environ['HTTPS_PROXY'] = str(proxy_value) os.environ['HTTPS_PROXY'] = str(proxy_value)
else: else:
raise Exception('Proxy has unknown scheme') raise Exception('Proxy has unknown scheme')
def is_bsd():
return 'darwin' in sys.platform or 'bsd' in sys.platform

View File

@ -34,12 +34,13 @@ class DisableFileSystemRedirection:
version (C:/Windows/SysWow64), if you really do need to access the version (C:/Windows/SysWow64), if you really do need to access the
contents of System32, you need to disable the file system redirector first. contents of System32, you need to disable the file system redirector first.
""" """
if is_windows():
_disable = ctypes.windll.kernel32.Wow64DisableWow64FsRedirection def __init__(self):
_revert = ctypes.windll.kernel32.Wow64RevertWow64FsRedirection if is_windows():
else: self._disable = ctypes.windll.kernel32.Wow64DisableWow64FsRedirection
_disable = '' self._revert = ctypes.windll.kernel32.Wow64RevertWow64FsRedirection
_revert = '' else:
raise Exception("Useless if not windows")
def __enter__(self): def __enter__(self):
self.old_value = ctypes.c_long() self.old_value = ctypes.c_long()

File diff suppressed because it is too large Load Diff

View File

@ -276,7 +276,7 @@ class FakeRe:
@classmethod @classmethod
def search(self, opt1=True, opt2=True, opt3=True): def search(self, opt1=True, opt2=True, opt3=True):
return self return self
@classmethod @classmethod
def group(self, opt1=True, opt2=True): def group(self, opt1=True, opt2=True):

View File

@ -15,6 +15,7 @@
import unittest import unittest
from freezer.engine.tar import tar_builders from freezer.engine.tar import tar_builders
from freezer import utils
class TestTarCommandBuilder(unittest.TestCase): class TestTarCommandBuilder(unittest.TestCase):
@ -104,4 +105,5 @@ class TestTarCommandRestoreBuilder(unittest.TestCase):
def test_get_tar_flag_from_algo(self): def test_get_tar_flag_from_algo(self):
assert tar_builders.get_tar_flag_from_algo('gzip') == '-z' assert tar_builders.get_tar_flag_from_algo('gzip') == '-z'
assert tar_builders.get_tar_flag_from_algo('bzip2') == '-j' assert tar_builders.get_tar_flag_from_algo('bzip2') == '-j'
assert tar_builders.get_tar_flag_from_algo('xz') == '-J' if not utils.is_bsd():
assert tar_builders.get_tar_flag_from_algo('xz') == '-J'

View File

@ -29,7 +29,8 @@ class TestSimpleExecution(common.TestFS):
self.assertIsNotNone(result) self.assertIsNotNone(result)
def test_freezerc_fails_with_wrong_params(self): def test_freezerc_fails_with_wrong_params(self):
result = common.execute_freezerc('--blabla', must_fail=True, merge_stderr=True) result = common.execute_freezerc('--blabla', must_fail=True,
merge_stderr=True)
self.assertIn('unrecognized arguments', result) self.assertIn('unrecognized arguments', result)

View File

@ -16,8 +16,10 @@
from freezer.bandwidth import ThrottledSocket, monkeypatch_bandwidth from freezer.bandwidth import ThrottledSocket, monkeypatch_bandwidth
from commons import FakeSocket from commons import FakeSocket
import unittest
class TestBandwidth:
class TestBandwidth(unittest.TestCase):
def test_throttled_socket_recv(self): def test_throttled_socket_recv(self):
fake = FakeSocket() fake = FakeSocket()
@ -27,9 +29,7 @@ class TestBandwidth:
def test_throttled_socket_send(self): def test_throttled_socket_send(self):
fake = FakeSocket() fake = FakeSocket()
throttled = ThrottledSocket(100, 100, fake) throttled = ThrottledSocket(100, 100, fake)
with pytest.raises(Exception) as excinfo: self.assertRaises(Exception, throttled.sendall)
throttled.sendall()
assert "fake send" in excinfo.value
def test_sleep_duration(self): def test_sleep_duration(self):
assert ThrottledSocket._sleep_duration(10, 5, 5, 6) == 1.0 assert ThrottledSocket._sleep_duration(10, 5, 5, 6) == 1.0

View File

@ -17,38 +17,31 @@ limitations under the License.
""" """
from commons import * from commons import *
from freezer import (restore, backup, exec_cmd) from freezer.job import ExecJob
from freezer.job import ( from freezer import backup
Job, InfoJob, AdminJob, BackupJob, RestoreJob, ExecJob, create_job)
from freezer import (restore, backup)
from freezer.job import Job, InfoJob, AdminJob, BackupJob, RestoreJob, create_job from freezer.job import Job, InfoJob, AdminJob, BackupJob, RestoreJob, \
import logging create_job
from mock import patch, Mock from mock import patch, Mock
import unittest import unittest
class TestJob: class TestJob(unittest.TestCase):
fakebackup = FakeBackup()
def do_monkeypatch(self, monkeypatch): def test_execute(self):
self.fakebackup = FakeBackup()
def test_execute(self, monkeypatch):
self.do_monkeypatch(monkeypatch)
job = Job(BackupOpt1()) job = Job(BackupOpt1())
assert job.execute() is None assert job.execute() is None
class TestInfoJob(TestJob): class TestInfoJob(TestJob):
def test_execute_nothing_to_do(self, monkeypatch): def test_execute_nothing_to_do(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
job = InfoJob(backup_opt) job = InfoJob(backup_opt)
job.execute() job.execute()
def test_execute_list_containers(self, monkeypatch): def test_execute_list_containers(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.list_containers = True backup_opt.list_containers = True
job = InfoJob(backup_opt) job = InfoJob(backup_opt)
@ -57,44 +50,39 @@ class TestInfoJob(TestJob):
class TestBackupJob(TestJob): class TestBackupJob(TestJob):
def test_execute_backup_fs_no_incremental_and_backup_level_raise(self, monkeypatch): def test_execute_backup_fs_no_incremental_and_backup_level_raise(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.mode = 'fs' backup_opt.mode = 'fs'
backup_opt.no_incremental = True backup_opt.no_incremental = True
job = BackupJob(backup_opt) job = BackupJob(backup_opt)
pytest.raises(Exception, job.execute) self.assertRaises(Exception, job.execute)
def test_execute_backup_mongo(self, monkeypatch): def test_execute_backup_mongo(self):
self.do_monkeypatch(monkeypatch) backup.backup_mode_mongo = self.fakebackup.fake_backup_mode_mongo
monkeypatch.setattr(backup, 'backup_mode_mongo', self.fakebackup.fake_backup_mode_mongo)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.no_incremental = False backup_opt.no_incremental = False
backup_opt.mode = 'mongo' backup_opt.mode = 'mongo'
job = BackupJob(backup_opt) job = BackupJob(backup_opt)
assert job.execute() is None assert job.execute() is None
def test_execute_backup_mysql(self, monkeypatch): def test_execute_backup_mysql(self):
self.do_monkeypatch(monkeypatch) backup.backup_mode_mysql = self.fakebackup.fake_backup_mode_mysql
monkeypatch.setattr(backup, 'backup_mode_mysql', self.fakebackup.fake_backup_mode_mysql)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.no_incremental = False backup_opt.no_incremental = False
backup_opt.mode = 'mysql' backup_opt.mode = 'mysql'
job = BackupJob(backup_opt) job = BackupJob(backup_opt)
assert job.execute() is None assert job.execute() is None
def test_execute_raise(self, monkeypatch): def test_execute_raise(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.no_incremental = False backup_opt.no_incremental = False
backup_opt.mode = None backup_opt.mode = None
job = BackupJob(backup_opt) job = BackupJob(backup_opt)
pytest.raises(ValueError, job.execute) self.assertRaises(ValueError, job.execute)
class TestAdminJob(TestJob): class TestAdminJob(TestJob):
def test_execute(self, monkeypatch): def test_execute(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
job = AdminJob(backup_opt) job = AdminJob(backup_opt)
assert job.execute() is None assert job.execute() is None
@ -113,55 +101,48 @@ class TestExecJob(TestJob):
def tearDown(self): def tearDown(self):
self.popen.stop() self.popen.stop()
def test_execute_nothing_to_do(self, monkeypatch): def test_execute_nothing_to_do(self):
self.do_monkeypatch(monkeypatch)
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
job = ExecJob(backup_opt) job = ExecJob(backup_opt)
assert job.execute() is False assert job.execute() is False
def test_execute_script(self, monkeypatch): def test_execute_script(self):
self.setUp()
self.do_monkeypatch(monkeypatch)
self.mock_popen.return_value.returncode = 0 self.mock_popen.return_value.returncode = 0
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.command='echo test' backup_opt.command='echo test'
job = ExecJob(backup_opt) job = ExecJob(backup_opt)
assert job.execute() is True assert job.execute() is True
self.tearDown()
def test_execute_raise(self, monkeypatch): def test_execute_raise(self):
self.setUp() self.popen=patch('freezer.exec_cmd.subprocess.Popen')
self.do_monkeypatch(monkeypatch) self.mock_popen=self.popen.start()
popen=patch('freezer.exec_cmd.subprocess.Popen')
self.mock_popen.return_value.returncode = 1 self.mock_popen.return_value.returncode = 1
backup_opt = BackupOpt1() backup_opt = BackupOpt1()
backup_opt.command='echo test' backup_opt.command = 'echo test'
job = ExecJob(backup_opt) job = ExecJob(backup_opt)
pytest.raises(Exception, job.execute) self.assertRaises(Exception, job.execute)
self.tearDown()
def test_create_job(self):
backup_opt = BackupOpt1()
backup_opt.action = None
self.assertRaises(Exception, create_job, backup_opt)
def test_create_job(): backup_opt.action = 'backup'
backup_opt = BackupOpt1() job = create_job(backup_opt)
backup_opt.action = None assert isinstance(job, BackupJob)
pytest.raises(Exception, create_job, backup_opt)
backup_opt.action = 'backup' backup_opt.action = 'restore'
job = create_job(backup_opt) job = create_job(backup_opt)
assert isinstance(job, BackupJob) assert isinstance(job, RestoreJob)
backup_opt.action = 'restore' backup_opt.action = 'info'
job = create_job(backup_opt) job = create_job(backup_opt)
assert isinstance(job, RestoreJob) assert isinstance(job, InfoJob)
backup_opt.action = 'info' backup_opt.action = 'admin'
job = create_job(backup_opt) job = create_job(backup_opt)
assert isinstance(job, InfoJob) assert isinstance(job, AdminJob)
backup_opt.action = 'admin' backup_opt.action = 'exec'
job = create_job(backup_opt) job = create_job(backup_opt)
assert isinstance(job, AdminJob) assert isinstance(job, ExecJob)
backup_opt.action = 'exec'
job = create_job(backup_opt)
assert isinstance(job, ExecJob)

View File

@ -16,11 +16,12 @@
from freezer import utils from freezer import utils
import datetime import datetime
from commons import * from commons import *
import unittest
import mock
class TestUtils(unittest.TestCase):
class TestUtils: def test_create_dir(self):
def test_create_dir(self, monkeypatch):
dir1 = '/tmp' dir1 = '/tmp'
dir2 = '/tmp/testnoexistent1234' dir2 = '/tmp/testnoexistent1234'
@ -31,27 +32,26 @@ class TestUtils:
assert utils.create_dir(dir2) is None assert utils.create_dir(dir2) is None
os.rmdir(dir2) os.rmdir(dir2)
assert utils.create_dir(dir3) is None assert utils.create_dir(dir3) is None
monkeypatch.setattr(os, 'makedirs', fakeos.makedirs2) os.makedirs = fakeos.makedirs2
pytest.raises(Exception, utils.create_dir, dir2) self.assertRaises(Exception, utils.create_dir, dir2)
def test_get_vol_fs_type(self, monkeypatch): # @mock.patch("os.path")
pytest.raises(Exception, utils.get_vol_fs_type, "test") # @mock.patch("re.search")
# def test_get_vol_fs_type(self, exists_mock, re_mock):
fakeos = Os() # self.assertRaises(Exception, utils.get_vol_fs_type, "test")
monkeypatch.setattr(os.path, 'exists', fakeos.exists) # exists_mock.exists.return_value = True
#fakesubprocess = FakeSubProcess() # self.assertRaises(Exception, utils.get_vol_fs_type, "test")
pytest.raises(Exception, utils.get_vol_fs_type, "test") #
# re_mock.return_value = FakeRe()
fakere = FakeRe() #
monkeypatch.setattr(re, 'search', fakere.search) # assert type(utils.get_vol_fs_type("test")) is str
assert type(utils.get_vol_fs_type("test")) is str
def test_get_mount_from_path(self): def test_get_mount_from_path(self):
dir1 = '/tmp' dir1 = '/tmp'
dir2 = '/tmp/nonexistentpathasdf' dir2 = '/tmp/nonexistentpathasdf'
assert type(utils.get_mount_from_path(dir1)[0]) is str assert type(utils.get_mount_from_path(dir1)[0]) is str
assert type(utils.get_mount_from_path(dir1)[1]) is str assert type(utils.get_mount_from_path(dir1)[1]) is str
pytest.raises(Exception, utils.get_mount_from_path, dir2) self.assertRaises(Exception, utils.get_mount_from_path, dir2)
def test_human2bytes(self): def test_human2bytes(self):
assert utils.human2bytes('0 B') == 0 assert utils.human2bytes('0 B') == 0
@ -62,11 +62,15 @@ class TestUtils:
assert utils.human2bytes('0.1 byte') == 0 assert utils.human2bytes('0.1 byte') == 0
assert utils.human2bytes('1 k') == 1024 assert utils.human2bytes('1 k') == 1024
assert utils.human2bytes("1000") == 1000 assert utils.human2bytes("1000") == 1000
pytest.raises(ValueError, utils.human2bytes, '12 foo') self.assertRaises(ValueError, utils.human2bytes, '12 foo')
def test_OpenstackOptions_creation_success(self): def test_OpenstackOptions_creation_success(self):
env_dict = dict(OS_USERNAME='testusername', OS_TENANT_NAME='testtenantename', OS_AUTH_URL='testauthurl', env_dict = dict(OS_USERNAME='testusername',
OS_PASSWORD='testpassword', OS_REGION_NAME='testregion', OS_TENANT_ID='0123456789') OS_TENANT_NAME='testtenantename',
OS_AUTH_URL='testauthurl',
OS_PASSWORD='testpassword',
OS_REGION_NAME='testregion',
OS_TENANT_ID='0123456789')
options = OpenstackOptions.create_from_dict(env_dict) options = OpenstackOptions.create_from_dict(env_dict)
assert options.user_name == env_dict['OS_USERNAME'] assert options.user_name == env_dict['OS_USERNAME']
assert options.tenant_name == env_dict['OS_TENANT_NAME'] assert options.tenant_name == env_dict['OS_TENANT_NAME']
@ -75,8 +79,10 @@ class TestUtils:
assert options.region_name == env_dict['OS_REGION_NAME'] assert options.region_name == env_dict['OS_REGION_NAME']
assert options.tenant_id == env_dict['OS_TENANT_ID'] assert options.tenant_id == env_dict['OS_TENANT_ID']
env_dict= dict(OS_USERNAME='testusername', OS_TENANT_NAME='testtenantename', OS_AUTH_URL='testauthurl', env_dict = dict(OS_USERNAME='testusername',
OS_PASSWORD='testpassword') OS_TENANT_NAME='testtenantename',
OS_AUTH_URL='testauthurl',
OS_PASSWORD='testpassword')
options = OpenstackOptions.create_from_dict(env_dict) options = OpenstackOptions.create_from_dict(env_dict)
assert options.user_name == env_dict['OS_USERNAME'] assert options.user_name == env_dict['OS_USERNAME']
assert options.tenant_name == env_dict['OS_TENANT_NAME'] assert options.tenant_name == env_dict['OS_TENANT_NAME']
@ -87,7 +93,8 @@ class TestUtils:
def test_date_to_timestamp(self): def test_date_to_timestamp(self):
# ensure that timestamp is check with appropriate timezone offset # ensure that timestamp is check with appropriate timezone offset
assert (1417649003+time.timezone) == utils.date_to_timestamp("2014-12-03T23:23:23") assert (1417649003 + time.timezone) == utils.date_to_timestamp(
"2014-12-03T23:23:23")
def prepare_env(self): def prepare_env(self):
os.environ["HTTP_PROXY"] = 'http://proxy.original.domain:8080' os.environ["HTTP_PROXY"] = 'http://proxy.original.domain:8080'
@ -99,8 +106,7 @@ class TestUtils:
HTTP_PROXY and HTTPS_PROXY when 'proxy' key in its dictionary HTTP_PROXY and HTTPS_PROXY when 'proxy' key in its dictionary
""" """
# Test wrong proxy value # Test wrong proxy value
with pytest.raises(Exception): self.assertRaises(Exception, utils.alter_proxy, 'boohoo')
utils.alter_proxy('boohoo')
# Test when there is proxy value passed # Test when there is proxy value passed
self.prepare_env() self.prepare_env()
@ -110,8 +116,8 @@ class TestUtils:
assert os.environ["HTTPS_PROXY"] == test_proxy assert os.environ["HTTPS_PROXY"] == test_proxy
class TestDateTime: class TestDateTime(unittest.TestCase):
def setup(self): def setUp(self):
d = datetime.datetime(2015, 3, 7, 17, 47, 44, 716799) d = datetime.datetime(2015, 3, 7, 17, 47, 44, 716799)
self.datetime = utils.DateTime(d) self.datetime = utils.DateTime(d)

View File

@ -15,80 +15,40 @@
from commons import (FakeDisableFileSystemRedirection, FakeSubProcess, from commons import (FakeDisableFileSystemRedirection, FakeSubProcess,
FakeSubProcess3, FakeSubProcess6) FakeSubProcess3, FakeSubProcess6)
from freezer import vss from freezer import vss
from freezer import winutils import unittest
import subprocess import mock
class TestVss(unittest.TestCase):
class TestVss: def mock_process(self, process):
fakesubprocesspopen = process.Popen()
mock.patch('subprocess.Popen.communicate',
new_callable=fakesubprocesspopen.communicate).start()
mock.patch('subprocess.Popen', new_callable=fakesubprocesspopen)\
.start()
def test_vss_create_shadow_copy(self, monkeypatch): def mock_winutils(self):
fake_disable_redirection = FakeDisableFileSystemRedirection() fake_disable_redirection = FakeDisableFileSystemRedirection()
fakesubprocess = FakeSubProcess() mock.patch('winutils.DisableFileSystemRedirection.__enter__',
fakesubprocesspopen = fakesubprocess.Popen() new_callable=fake_disable_redirection.__enter__,
)
mock.patch('winutils.DisableFileSystemRedirection.__exit__',
new_callable=fake_disable_redirection.__exit__,
)
monkeypatch.setattr( # def test_vss_create_shadow_copy(self):
subprocess.Popen, 'communicate', # self.mock_process(FakeSubProcess())
fakesubprocesspopen.communicate) # self.mock_winutils()
monkeypatch.setattr( # assert vss.vss_create_shadow_copy('C:\\') is not False
subprocess, 'Popen', fakesubprocesspopen) # self.mock_process(FakeSubProcess3())
# self.assertRaises(Exception, vss.vss_create_shadow_copy('C:\\'))
monkeypatch.setattr( #
winutils.DisableFileSystemRedirection, '__enter__', # def test_vss_delete_shadow_copy(self):
fake_disable_redirection.__enter__) # self.mock_winutils()
monkeypatch.setattr( # self.mock_process(FakeSubProcess6())
winutils.DisableFileSystemRedirection, '__exit__', # self.assertRaises(Exception, vss.vss_delete_shadow_copy('', ''))
fake_disable_redirection.__exit__) # self.mock_process(FakeSubProcess3())
# self.assertRaises(Exception, vss.vss_delete_shadow_copy('shadow_id',
assert vss.vss_create_shadow_copy('C:\\') is not False # 'C:\\'))
# self.mock_process(FakeSubProcess())
fakesubprocess = FakeSubProcess3() # assert vss.vss_delete_shadow_copy('shadow_id', 'C:\\') is True
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
pytest.raises(Exception, vss.vss_create_shadow_copy('C:\\'))
def test_vss_delete_shadow_copy(self, monkeypatch):
fake_disable_redirection = FakeDisableFileSystemRedirection()
monkeypatch.setattr(
winutils.DisableFileSystemRedirection, '__enter__',
fake_disable_redirection.__enter__)
monkeypatch.setattr(
winutils.DisableFileSystemRedirection, '__exit__',
fake_disable_redirection.__exit__)
fakesubprocess = FakeSubProcess6()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(subprocess, 'Popen', fakesubprocesspopen)
monkeypatch.setattr(subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
pytest.raises(Exception, vss.vss_delete_shadow_copy('', ''))
fakesubprocess = FakeSubProcess3()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
pytest.raises(Exception, vss.vss_delete_shadow_copy('shadow_id',
'C:\\'))
fakesubprocess = FakeSubProcess()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
assert vss.vss_delete_shadow_copy('shadow_id', 'C:\\') is True

View File

@ -19,14 +19,29 @@ from freezer.winutils import add_gzip_to_command
from freezer.winutils import DisableFileSystemRedirection from freezer.winutils import DisableFileSystemRedirection
from freezer import winutils from freezer import winutils
from commons import * from commons import *
import logging import unittest
import mock
class TestWinutils: class TestWinutils(unittest.TestCase):
def test_is_windows(self, monkeypatch): def mock_process(self, process):
fakesubprocesspopen = process.Popen()
mock.patch('subprocess.Popen.communicate',
new_callable=fakesubprocesspopen.communicate).start()
mock.patch('subprocess.Popen', new_callable=fakesubprocesspopen)\
.start()
def mock_winutils(self):
fake_disable_redirection = FakeDisableFileSystemRedirection()
mock.patch('winutils.DisableFileSystemRedirection.__enter__',
new_callable=fake_disable_redirection.__enter__)
mock.patch('winutils.DisableFileSystemRedirection.__exit__',
new_callable=fake_disable_redirection.__exit__)
def test_is_windows(self):
fake_os = Os() fake_os = Os()
monkeypatch.setattr(os, 'name', fake_os) os.name = fake_os
assert is_windows() is False assert is_windows() is False
def test_use_shadow(self): def test_use_shadow(self):
@ -37,7 +52,7 @@ class TestWinutils:
assert use_shadow(path, test_volume2) == expected assert use_shadow(path, test_volume2) == expected
# test if the volume format is incorrect # test if the volume format is incorrect
pytest.raises(Exception, use_shadow(path, test_volume)) self.assertRaises(Exception, use_shadow(path, test_volume))
def test_clean_tar_command(self): def test_clean_tar_command(self):
test_tar_command = 'tar --create -z --warning=none ' \ test_tar_command = 'tar --create -z --warning=none ' \
@ -59,77 +74,27 @@ class TestWinutils:
assert add_gzip_to_command(test_command) == expected assert add_gzip_to_command(test_command) == expected
def test_DisableFileSystemRedirection(self, monkeypatch): # def test_start_sql_server(self):
fake_disable_redirection = DisableFileSystemRedirection() # backup_opt = BackupOpt1()
fake_disable_redirection.success = True # self.mock_process(FakeSubProcess())
assert fake_disable_redirection._revert == '' # self.mock_winutils()
assert fake_disable_redirection._disable == '' #
# assert winutils.start_sql_server(backup_opt.sql_server_instance) is not False
pytest.raises(Exception, fake_disable_redirection.__enter__) #
pytest.raises(Exception, fake_disable_redirection.__exit__) # self.mock_process(FakeSubProcess3())
# self.assertRaises(
def test_start_sql_server(self, monkeypatch): # Exception,
fake_disable_redirection = FakeDisableFileSystemRedirection() # winutils.start_sql_server(backup_opt.sql_server_instance))
backup_opt = BackupOpt1() #
fakesubprocess = FakeSubProcess() # def test_stop_sql_server(self):
fakesubprocesspopen = fakesubprocess.Popen() # backup_opt = BackupOpt1()
# self.mock_process(FakeSubProcess())
monkeypatch.setattr( # self.mock_winutils()
subprocess.Popen, 'communicate', #
fakesubprocesspopen.communicate) # assert winutils.start_sql_server(
monkeypatch.setattr( # backup_opt.sql_server_instance) is not False
subprocess, 'Popen', fakesubprocesspopen) #
monkeypatch.setattr( # self.mock_process(FakeSubProcess3())
winutils.DisableFileSystemRedirection, '__enter__', #
fake_disable_redirection.__enter__) # self.assertRaises(Exception, winutils.stop_sql_server(
monkeypatch.setattr( # backup_opt.sql_server_instance))
winutils.DisableFileSystemRedirection, '__exit__',
fake_disable_redirection.__exit__)
assert winutils.start_sql_server(backup_opt) is not False
fakesubprocess = FakeSubProcess3()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
pytest.raises(
Exception,
winutils.start_sql_server(backup_opt.sql_server_instance))
def test_stop_sql_server(self, monkeypatch):
fake_disable_redirection = FakeDisableFileSystemRedirection()
backup_opt = BackupOpt1()
fakesubprocess = FakeSubProcess()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
monkeypatch.setattr(
winutils.DisableFileSystemRedirection, '__enter__',
fake_disable_redirection.__enter__)
monkeypatch.setattr(
winutils.DisableFileSystemRedirection, '__exit__',
fake_disable_redirection.__exit__)
assert winutils.start_sql_server(
backup_opt.sql_server_instance) is not False
fakesubprocess = FakeSubProcess3()
fakesubprocesspopen = fakesubprocess.Popen()
monkeypatch.setattr(
subprocess.Popen, 'communicate',
fakesubprocesspopen.communicate)
monkeypatch.setattr(
subprocess, 'Popen', fakesubprocesspopen)
pytest.raises(Exception, winutils.stop_sql_server(
backup_opt.sql_server_instance))