diff --git a/tests/test_ova_downloader.py b/tests/test_ova_downloader.py new file mode 100644 index 0000000..6fc735c --- /dev/null +++ b/tests/test_ova_downloader.py @@ -0,0 +1,70 @@ +import jenkins +import unittest +import requests +from ova_downloader import get_last_build_template_url, download_ova +from unittest.mock import patch +import os + + +class MockRequests: + headers = {'content-length': '9'} + + @staticmethod + def close(): + return + + @staticmethod + def iter_content(block_size): + return + + +class LastBuildUrl(unittest.TestCase): + @patch.object(jenkins.Jenkins, 'get_job_info') + def test_get_build(self, mocked_get_job_info): + result = {'lastSuccessfulBuild': {'url': ''}} + mocked_get_job_info.return_value = result + self.assertTrue(get_last_build_template_url) + + @patch.object(jenkins.Jenkins, 'get_job_info') + def test_key_error_get_build(self, mocked_get_job_info): + result = {'lastSuccessfulBuild': {}} + mocked_get_job_info.return_value = result + with self.assertRaises(KeyError): + get_last_build_template_url() + + +class DownloadOva(unittest.TestCase): + + @patch('ova_downloader.tqdm') + @patch('ova_downloader.get_last_build_template_url') + @patch('ova_downloader.requests.get') + def test_fail_to_download_ova(self, mocked_get, mock_last_build, mock_tqdm): + mock_last_build.return_value = '' + mocked_get.get.return_value = MockRequests + cont_length = int(mocked_get.get.return_value.headers.get('content-length')) + mock_tqdm.return_value = [bytes(str(i), 'utf8') for i in range(cont_length)] + with self.assertRaises(EnvironmentError): + download_ova(dir_path=os.getcwd()) + + @patch('ova_downloader.get_last_build_template_url') + @patch('ova_downloader.requests.get') + def test_env_error(self, mocked_get, mock_last_build): + mock_last_build.return_value = '' + mocked_get.get.return_value = MockRequests + with self.assertRaises(EnvironmentError): + download_ova(dir_path='{}/folder_not_exist'.format(os.getcwd())) + + @patch('ova_downloader.tqdm') + @patch('ova_downloader.get_last_build_template_url') + @patch('ova_downloader.requests.get') + def test_download_ova_passed(self, mocked_get, mock_last_build, mock_tqdm): + mock_last_build.return_value = '' + mocked_get.return_value = MockRequests + cont_length = int(mocked_get.return_value.headers.get('content-length')) + mock_tqdm.return_value = (bytes(str(i), 'utf8') for i in range(cont_length)) + self.assertIsNone(download_ova(dir_path=os.getcwd())) + + +if __name__ == '__main__': + unittest.main() + diff --git a/tests/test_vaclient.py b/tests/test_vaclient.py new file mode 100644 index 0000000..e31605b --- /dev/null +++ b/tests/test_vaclient.py @@ -0,0 +1,79 @@ +import unittest +from unittest import mock +import paramiko +from unittest.mock import patch + +from vaclient import VAClient + + +class FakeTransport: + def __init__(self): + self.active = True + self.o_session = True + + def is_active(self): + return self.active + + def open_session(self): + return self + + def recv(self, lol): + return 'hello world' + + +class FakeChannel: + + def __init__(self, closed=True): + self.closed = closed + + @property + def channel(self): + return self + + def recv_exit_status(self): + return 0 + + def readlines(self): + return 'channel stdout' + + +class FakeSSHClient: + def __init__(self): + self.transport = FakeTransport + # self.raise_exception = raise_exception + # self.excaption_name = excaption_name + + def connect(self, *args): + pass + + def set_missing_host_key_policy(self, policy): + pass + + def exec_command(self, cmd, get_pty=True): + return (FakeChannel(), + FakeChannel(), + FakeChannel()) + + def get_transport(self): + return FakeTransport() + + def close(self): + pass + + def __call__(self, *args, **kwargs): + pass + + +class TestPar(unittest.TestCase): + def test_connect_is_created(self): + with mock.patch.object(paramiko, + 'SSHClient', + mock.Mock(return_value=FakeSSHClient())): + + con = VAClient(ip='192.168.242.22') + self.assertTrue(con.connect()) + + +if __name__ == '__main__': + unittest.main() +