diff --git a/Pipfile.lock b/Pipfile.lock index 8223728..221ffb9 100644 --- a/Pipfile.lock +++ b/Pipfile.lock @@ -107,6 +107,13 @@ ], "version": "==0.4.3" }, + "parameterized": { + "hashes": [ + "sha256:6a94dbea30c6abde99fd4c2f2042c1bf7f980e48908bf92ead62394f93cf57ed", + "sha256:ea0326ba5bbbe7c427329a27b75003410df07d1173ca254976f8f5a64922c322" + ], + "version": "==0.7.1" + }, "ptable": { "hashes": [ "sha256:aa7fc151cb40f2dabcd2275ba6f7fd0ff8577a86be3365cd3fb297cbe09cc292" diff --git a/VERSION b/VERSION index 1b87bcd..314c3d7 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.1.4 \ No newline at end of file +1.1.5 \ No newline at end of file diff --git a/cli/kaos_cli/constants.py b/cli/kaos_cli/constants.py index 26ea474..8b7e380 100644 --- a/cli/kaos_cli/constants.py +++ b/cli/kaos_cli/constants.py @@ -98,3 +98,5 @@ CONTEXTS = 'contexts' ACTIVE = 'active' REMOTE = 'remote' + +TEST_CONFIG_PATH = os.path.join(KAOS_HOME, 'kaos_cli', 'facades', 'tests', 'fixtures', 'config') diff --git a/cli/kaos_cli/facades/backend_facade.py b/cli/kaos_cli/facades/backend_facade.py index cf63526..98c5339 100644 --- a/cli/kaos_cli/facades/backend_facade.py +++ b/cli/kaos_cli/facades/backend_facade.py @@ -12,7 +12,11 @@ from kaos_cli.services.terraform_service import TerraformService from kaos_cli.utils.environment import check_environment from kaos_cli.utils.helpers import build_dir -from kaos_cli.utils.validators import EnvironmentState, is_cloud_provider +from kaos_cli.utils.validators import EnvironmentState + + +def is_cloud_provider(cloud): + return cloud not in (DOCKER, MINIKUBE) class BackendFacade: @@ -193,7 +197,6 @@ def _remove_build_files(self, dir_build): """ Function to remove backend directory """ - self.state_service.provider_delete(dir_build) if not self.state_service.list_providers(): self.state_service.full_delete(dir_build) diff --git a/cli/kaos_cli/facades/tests/test_backend_facade.py b/cli/kaos_cli/facades/tests/test_backend_facade.py new file mode 100644 index 0000000..c281367 --- /dev/null +++ b/cli/kaos_cli/facades/tests/test_backend_facade.py @@ -0,0 +1,380 @@ +import os +import unittest +import mock +from parameterized import parameterized +from unittest import TestCase, skip + +from kaos_cli.facades.backend_facade import BackendFacade +from kaos_cli.services.terraform_service import TerraformService +from kaos_cli.services.state_service import StateService +from kaos_cli.services.terraform_service import Command +from kaos_cli.constants import CONFIG_PATH, ACTIVE, DEFAULT, CONTEXTS, KAOS_STATE_DIR + +common = ('backend', 'url', 'token', 'infrastructure', 'kubeconfig', 'user') + + +def backend_facade_test_inputs(): + + return [ + ('DOCKER',) + common, + ('GCP',) + common, + ('AWS',) + common, + ('Remote1',) + common, + ] + + +class TestBackendFacade(TestCase): + + @staticmethod + def derive_context(cloud): + env_list = ['', 'dev', 'stage', 'prod'] + for env in env_list: + return cloud + '_' + env + + def manipulate_context_list(self, context_list): + self.state_service.set(CONTEXTS, environments=context_list) + + def arrange_test_fixtures(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + # Instantiation + self.state_service = StateService() + self.command = Command() + self.tf_service = TerraformService(cmd=self.command) + + # Arrange + # Initialization + self.state_service.set(ACTIVE, environment=environment) + self.state_service.set(CONTEXTS, environments=environment) + self.state_service.set(DEFAULT, user=user) + self.state_service.set(environment) + self.state_service.set_section(environment, backend, url=url, token=token) + self.state_service.set_section(environment, infra, kubeconfig=kubeconfig) + self.facade = BackendFacade(self.state_service, self.tf_service) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_get_active_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + test_inputs_active_contexts = [None, '', environment] + + for env in test_inputs_active_contexts: + self.state_service.set(ACTIVE, environment=env) + + # Act + active_context = self.facade.get_active_context() + + # Assert + self.assertEqual(active_context, self.facade.state_service.get(ACTIVE, 'environment')) + + @parameterized.expand(backend_facade_test_inputs) + def test_property_active_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + active_context_property = self.facade.active_context + + # Assert + self.assertEqual(active_context_property, environment) + + @parameterized.expand(backend_facade_test_inputs) + def test_property_url(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + url_property = self.facade.url + + # Assert + self.assertEqual(url_property, url) + + @parameterized.expand(backend_facade_test_inputs) + def test_property_user(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + user_property = self.facade.user + + # Assert + self.assertEqual(user_property, user) + + @parameterized.expand(backend_facade_test_inputs) + def test_property_token(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + token_property = self.facade.token + + # Assert + self.assertEqual(token_property, token) + + @parameterized.expand(backend_facade_test_inputs) + def test_property_kubeconfig(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + kubeconfig_property = self.facade.kubeconfig + + # Assert + self.assertEqual(kubeconfig_property, kubeconfig) + + @skip("Skipping test for init method") + @parameterized.expand(backend_facade_test_inputs) + def test_method_init(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + ## TODO: Rectify the following assertions after fixing the dir_build path for init() after merging PR #95 + # self.facade.init(url, token) + + # Assert + self.assertEqual(0, 0) + self.assertEqual(1, 1) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_list(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + list_result = self.facade.list() + context = list_result[0]['context'] + + # Assert + self.assertEqual(context, environment) + + # Arrange (modify the contexts list) + test_inputs = [None, ['DOCKER', 'GCP', 'AWS']] + for test_input in test_inputs: + self.state_service.set(CONTEXTS, environments=test_input) + + # Act + list_result = self.facade.list() + + if len(list_result) > 0: + context = [x['context'] for x in list_result] + else: + context = list_result + test_input = [] + + # Assert + print("context", context) + self.assertEqual(context, test_input) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_get_active_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + active_context = self.facade.get_active_context() + + # Assert + self.assertEqual(active_context, environment) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_get_context_info(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + index = 0 + cloud_env = self.derive_context(environment) + + # Act + context_info = self.facade.get_context_info(cloud_env, index) + provider = context_info['provider'] + + # Assert + self.assertEqual(provider, environment) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_get_context_by_index(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + index = 0 + + cloud_env = self.derive_context(environment) + + context_info = self.facade.get_context_info(cloud_env, index) + + # Act + context = self.facade.get_context_by_index([context_info], index) + + # Assert + self.assertEqual(context, cloud_env) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_jsonify_context_list(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + cloud_env = self.derive_context(environment) + + # Act + context_info = self.facade.jsonify_context_list(cloud_env) + provider = context_info[0]['provider'] + + # Assert + self.assertEqual(provider, environment) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_set_context_by_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + self.facade.state_service.create() + with open(CONFIG_PATH, 'wb') as f: + f.close() + + # Act + is_set = self.facade.set_context_by_context(environment) + + # Assert + self.assertTrue(is_set) + + # Act + is_set = self.facade.set_context_by_context('invalid_environment') + + # Assert + self.assertFalse(is_set) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_set_context_by_index(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + valid_index = 0 + invalid_index = 1 + + # Act + is_set, current_context = self.facade.set_context_by_index(valid_index) + + # Assert + self.assertTrue(is_set) + self.assertEqual(current_context, environment) + + # Act + is_set, current_context = self.facade.set_context_by_index(invalid_index) + + # Assert + self.assertFalse(is_set) + self.assertIsNone(current_context) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_set_context_list(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + self.facade._set_context_list(environment) + list_result = self.facade.list() + contexts = [x['provider'] for x in list_result] + + # Assert + self.assertIn(environment, contexts) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_unset_context_list(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + self.facade._unset_context_list(environment) + list_result = self.facade.list() + contexts = [x['provider'] for x in list_result] + + # Assert + self.assertNotIn(environment, contexts) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_set_active_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + self.facade._set_active_context(environment) + active_context = self.facade.state_service.get(ACTIVE, 'environment') + + # Assert + self.assertEqual(active_context, environment) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_remove_section(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + self.facade._remove_section(environment) + config = self.facade.state_service.config + + # Assert + self.assertNotIn(environment, config) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_remove_section(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + # Act + self.facade._remove_section(environment) + config = self.facade.state_service.config + + # Assert + self.assertNotIn(environment, config) + + @parameterized.expand(backend_facade_test_inputs) + def test_method_deactivate_context(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + self.arrange_test_fixtures(environment, backend, url, token, infra, kubeconfig, user) + + self.facade.state_service.create() + with open(CONFIG_PATH, 'wb') as f: + f.close() + + # Act + self.facade._deactivate_context() + active_context = self.facade.state_service.get(ACTIVE, 'environment') + + # Assert + self.assertIsNone(active_context) + + +class TestBackendFacadeMocked(unittest.TestCase): + + def arrange_test_fixtures(self, environment, backend, url, token, infra, kubeconfig, user): + # Arrange + # Instantiation + self.state_service = StateService() + self.command = Command() + self.tf_service = TerraformService(cmd=self.command) + + # Arrange + # Initialization + self.state_service.set(ACTIVE, environment=environment) + self.state_service.set(CONTEXTS, environments=environment) + self.state_service.set(DEFAULT, user=user) + self.state_service.set(environment) + self.state_service.set_section(environment, backend, url=url, token=token) + self.state_service.set_section(environment, infra, kubeconfig=kubeconfig) + self.facade = BackendFacade(self.state_service, self.tf_service) + + @mock.patch('shutil.rmtree') + def test_method_remove_build_files(self, rm_mock): + # Arrange + environment = "DOCKER" + + ## common = ('backend', 'url', 'token', 'infrastructure', 'kubeconfig', 'user') + self.arrange_test_fixtures(environment, common[0], common[1], common[2], common[3], common[4], common[5]) + test_path = os.path.join(KAOS_STATE_DIR, 'test') + rm_mock.return_value = 'REMOVED' + + # Act + self.facade._remove_build_files(test_path) + + # Assert + rm_mock.assert_called_with(test_path, ignore_errors=True) + self.assertEqual(rm_mock.return_value, 'REMOVED') diff --git a/cli/kaos_cli/services/state_service.py b/cli/kaos_cli/services/state_service.py index 3db960c..3ceae76 100644 --- a/cli/kaos_cli/services/state_service.py +++ b/cli/kaos_cli/services/state_service.py @@ -8,8 +8,8 @@ class StateService: - def __init__(self, config=None): - self.config = config or ConfigObj(CONFIG_PATH) + def __init__(self): + self.config = ConfigObj(CONFIG_PATH) def set(self, section, **kwargs): self.config[section] = kwargs @@ -45,10 +45,6 @@ def create(): def list_providers(): return [f for f in glob.glob(f'{KAOS_TF_PATH}/**/terraform.tfstate', recursive=True)] - @staticmethod - def provider_delete(dir_build): - shutil.rmtree(dir_build, ignore_errors=True) - @staticmethod def full_delete(dir_build): shutil.rmtree(dir_build, ignore_errors=True) diff --git a/cli/kaos_cli/services/terraform_service.py b/cli/kaos_cli/services/terraform_service.py index 45adba5..350ecf2 100644 --- a/cli/kaos_cli/services/terraform_service.py +++ b/cli/kaos_cli/services/terraform_service.py @@ -13,8 +13,9 @@ def append(self, cmd): self.history.append(cmd) def execute(self): - self.run_cmd(";".join(self.history)) + exitcode, out, err = self.run_cmd(";".join(self.history)) self.clear() + return exitcode, out, err def clear(self): self.history = [] @@ -53,7 +54,8 @@ def destroy(self, directory, extra_vars): f"{extra_vars} --auto-approve {directory}") def execute(self): - self.cmd.execute() + exitcode, out, err = self.cmd.execute() + return exitcode, out, err def cd_dir(self, build_dir): self.cmd.append(f'cd {build_dir}') diff --git a/cli/kaos_cli/services/tests/test_state_service.py b/cli/kaos_cli/services/tests/test_state_service.py new file mode 100644 index 0000000..d054d36 --- /dev/null +++ b/cli/kaos_cli/services/tests/test_state_service.py @@ -0,0 +1,280 @@ +import os +import unittest +import mock +from parameterized import parameterized_class +from configobj import ConfigObj + +from unittest import TestCase + +from kaos_cli.services.state_service import StateService +from kaos_cli.constants import CONFIG_PATH, KAOS_STATE_DIR + + +def state_service_test_inputs(): + + """Returns different combinations of test input data/parameters that are used in the unit tests + :parameter: + + :return: list + a list of tuples consisting of different parameter combinations + Ex: [(context1, param1), (context2, param2)..] + + """ + return [ + ( + 'context1', 'param1' + ), + ( + 'context2', 'param2' + ), + ( + '', 'param3' + ), + ( + 'context4', '' + ), + ( + '', '' + ), + ( + {}, 'param6' + ), + ( + 'context7', {} + ), + ( + {}, {} + ), + ( + '', {} + ), + ( + {}, '' + ), + ( + None, 'param11' + ), + ( + 'context11', None + ), + ( + None, None + ), + ( + {}, None + ), + ( + None, {} + ), + ( + '', None + ), + ( + None, '' + ), + ( + 1, 'param18' + ), + ( + 'context', 2 + ), + ( + 3, {} + ), + ( + {}, 4 + ), + ( + 5, '' + ), + ( + '', 6 + ), + ( + 7, None + ), + ( + None, 8 + ), + + ] + + +@parameterized_class(('context', 'param'), state_service_test_inputs()) +class TestStateService(TestCase): + + def instantiate_service(self): + # Arrange + self.state_service = StateService() + + def test_state_service_create(self): + # Arrange + self.instantiate_service() + + # Act + self.state_service.create() + + # Assert + self.assertTrue(os.path.exists(KAOS_STATE_DIR)) + + def test_state_service_is_created(self): + # Arrange + self.instantiate_service() + + # Act + is_created = self.state_service.is_created(KAOS_STATE_DIR) + + # Assert + self.assertEqual(True, is_created) + + def test_state_service_list_providers(self): + # Arrange + self.instantiate_service() + + # Act + provider_list = self.state_service.list_providers() + + # Assert + self.assertEqual([], provider_list) + + def test_state_service_set(self): + # Arrange + self.instantiate_service() + + # Arrange + if not isinstance(self.context, str): + if self.context is None: + self.context = {} + else: + self.context = str(self.context) + + # Act + self.state_service.set('section', context=self.context) + + # Assert + self.assertEqual(self.context, self.state_service.config['section']['context']) + + def test_state_service_get(self): + # Arrange + self.instantiate_service() + + if not isinstance(self.context, str): + if self.context is None: + self.context = {} + else: + self.context = str(self.context) + + self.state_service.set('section', context=self.context) + + # Act + self.state_service.get('section', 'context') + + # Assert + self.assertEqual(self.context, self.state_service.config['section']['context']) + + def test_state_service_set_sections(self): + # Arrange + self.instantiate_service() + self.state_service.set('section', context=self.context) + + if not isinstance(self.param, str): + if self.param is None: + self.param = {} + else: + self.param = str(self.context) + + # Act + self.state_service.set_section('section', 'subsection', param=self.param) + + # Assert + self.assertEqual(self.param, self.state_service.config['section']['subsection']['param']) + + def test_state_service_get_sections(self): + # Arrange + self.instantiate_service() + self.state_service.set('section', context=self.context) + + if not isinstance(self.param, str): + if self.param is None: + self.param = {} + else: + self.param = str(self.context) + + self.state_service.set_section('section', 'subsection', param=self.param) + + # Act + self.state_service.get_section('section', 'subsection', 'param') + + # Assert + self.assertEqual(self.state_service.config['section']['subsection'], + self.state_service.has_section('section', 'subsection')) + + def test_state_service_write(self): + # Arrange + self.instantiate_service() + + # Act + self.state_service.write() + + # Assert + self.assertEqual(self.state_service.config, ConfigObj(CONFIG_PATH)) + + def test_state_service_has_section(self): + # Arrange + self.instantiate_service() + + self.state_service.set('section', context=self.context) + self.state_service.set_section('section', 'subsection', param=self.param) + + # Act + self.state_service.has_section('section', 'subsection') + + # Assert + self.assertEqual(self.state_service.config['section']['subsection'], + self.state_service.has_section('section', 'subsection')) + + def test_state_service_remove(self): + # Arrange + self.instantiate_service() + self.state_service.set('section', context=self.context) + + # Act + self.state_service.remove('section') + + # Assert + self.assertEqual("removed", "removed" if 'section' not in self.state_service.config else "not removed") + + def test_state_service_remove_section(self): + # Arrange + self.instantiate_service() + self.state_service.set('section', context=self.context) + self.state_service.set_section('section', 'subsection', param=self.param) + + # Act + self.state_service.remove_section('section', 'subsection') + + is_removed = 'subsection' not in self.state_service.config['section'] + + self.assertTrue(is_removed) + + +class TestStateServiceMocked(unittest.TestCase): + + def instantiate_service(self): + # Arrange + self.state_service = StateService() + + @mock.patch('shutil.rmtree') + def test_method_remove_build_files(self, rm_mock): + # Arrange + self.instantiate_service() + test_path = os.path.join(KAOS_STATE_DIR, 'test') + rm_mock.return_value = 'REMOVED' + + # Act + self.state_service.full_delete(test_path) + + # Assert + rm_mock.assert_called_with(test_path, ignore_errors=True) + self.assertEqual(rm_mock.return_value, 'REMOVED') + diff --git a/cli/kaos_cli/services/tests/test_tf_service.py b/cli/kaos_cli/services/tests/test_tf_service.py new file mode 100644 index 0000000..872704d --- /dev/null +++ b/cli/kaos_cli/services/tests/test_tf_service.py @@ -0,0 +1,338 @@ +import os +from parameterized import parameterized + +from unittest import TestCase + +from kaos_cli.services.terraform_service import TerraformService +from kaos_cli.services.terraform_service import Command +from kaos_cli.utils.helpers import verbose_run +from kaos_cli.constants import KAOS_STATE_DIR + + +def test_inputs_append_data(): + + """Returns different combinations of test input data/parameters that are used in the unit tests + :parameter: + + :return: list + a list of tuples consisting of different set of parameter combinations + Ex: [(command1), ([command1, command2]), ([command1, command2, command3])..] + + """ + + return [ + ( + [], + ), + ( + ['echo <---first_echo_command--->'], + ), + ( + ['echo <---first_echo_command--->', 'echo <--- executing the second echo command --->'], + ), + ( + ['echo <--- executing the first echo command --->', 'echo <--- executing the second echo command --->', + 'echo <--- executing the third echo command --->'], + ), + ] + + +def test_inputs_execute_data(): + + """Returns different combinations of test input data/parameters that are used in the unit tests + :parameter: + + :return: list + a list of tuples consisting of different parameter combinations + Ex: [(command1), ([command1, command2]), ([command1, command2, command3])..] + + """ + return [ + ( + ['echo <---first_echo_command--->', 'echo <--- executing the second echo command --->'], + ), + ( + ['echo <--- executing the first echo command --->', 'echo <--- executing the second echo command --->', + 'echo <--- executing the third echo command --->'], + ), + ] + + +class TestTerraformService(TestCase): + + @parameterized.expand(test_inputs_append_data) + def test_command_append(self, command_append_data): + + # Arrange + command = Command() + + # Act + for cmd in command_append_data: + command.append(cmd) + + # Assert + self.assertEqual(command.history, command_append_data) + + @parameterized.expand(test_inputs_execute_data) + def test_command_execute(self, command_execute_data): + + # Arrange + command = Command() + + # Act + for cmd in command_execute_data: + command.append(cmd) + + exitcode, out, err = command.execute() + + # Assert + self.assertEqual(exitcode, 0) + + @parameterized.expand(test_inputs_append_data) + def test_command_clear(self, command_append_data): + + # Arrange + command = Command() + + # Act + for cmd in command_append_data: + command.append(cmd) + + command.clear() + + # Assert + self.assertEqual(command.history, []) + + @parameterized.expand(test_inputs_execute_data) + def test_command_verbose_true(self, command_execute_data): + + # Arrange + command = Command() + verbose = True + + # Act + for cmd in command_execute_data: + command.set_verbose(verbose) + + # Assert + self.assertIsInstance(command.run_cmd, type(command.run_cmd)) + self.assertEqual(command.run_cmd.func, verbose_run) + self.assertEqual(command.run_cmd.args[0], True) + + @parameterized.expand(test_inputs_execute_data) + def test_command_verbose_false(self, command_execute_data): + + # Arrange + command = Command() + verbose = False + + # Act + for cmd in command_execute_data: + command.set_verbose(verbose) + + # Assert + self.assertIsInstance(command.run_cmd, type(command.run_cmd)) + self.assertEqual(command.run_cmd.func, verbose_run) + self.assertEqual(command.run_cmd.args[0], False) + + @parameterized.expand(test_inputs_execute_data) + def test_tf_service_set_verbose_True(self, command_execute_data): + + # Arrange + verbose = True + command = Command() + + # Act + for cmd in command_execute_data: + command.append(cmd) + + tf_service = TerraformService(cmd=command) + tf_service.set_verbose(verbose) + + # Assert + self.assertIsInstance(tf_service.set_verbose, type(tf_service.set_verbose)) + self.assertIsInstance(tf_service.cmd.set_verbose, type(tf_service.cmd.set_verbose)) + self.assertEqual(tf_service.cmd.run_cmd.func, verbose_run) + self.assertEqual(tf_service.cmd.run_cmd.args[0], True) + + @parameterized.expand(test_inputs_execute_data) + def test_tf_service_set_verbose_False(self, command_execute_data): + + # Arrange + verbose = False + command = Command() + + # Act + for cmd in command_execute_data: + command.append(cmd) + + tf_service = TerraformService(cmd=command) + tf_service.set_verbose(verbose) + + # Assert + self.assertIsInstance(tf_service.set_verbose, type(tf_service.set_verbose)) + self.assertIsInstance(tf_service.cmd.set_verbose, type(tf_service.cmd.set_verbose)) + self.assertEqual(tf_service.cmd.run_cmd.func, verbose_run) + self.assertEqual(tf_service.cmd.run_cmd.args[0], False) + + def test_tf_service_init(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + dir_path = KAOS_STATE_DIR + derived_path = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform init {derived_path} "] + + # Act + tf_service.init(dir_path) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_new_workspace(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + env = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform workspace new {env} {directory}"] + + # Act + tf_service.new_workspace(directory, env) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_select_workspace(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + env = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform workspace select {env} {directory}"] + + # Act + tf_service.select_workspace(directory, env) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_plan(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + env = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform workspace select {env} {directory}"] + + # Act + tf_service.select_workspace(directory, env) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_apply(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + extra_vars = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform apply --var-file={directory}/terraform.tfvars {extra_vars} " + f"--auto-approve {directory}"] + + # Act + tf_service.apply(directory, extra_vars) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_destroy(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + extra_vars = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform destroy --var-file={directory}/terraform.tfvars {extra_vars} --auto-approve " + f"{directory}"] + + # Act + tf_service.destroy(directory, extra_vars) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + def test_tf_service_destroy(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + extra_vars = 'prod' + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"terraform destroy --var-file={directory}/terraform.tfvars {extra_vars} --auto-approve " + f"{directory}"] + + # Act + tf_service.destroy(directory, extra_vars) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + @parameterized.expand(test_inputs_execute_data) + def test_tf_service_execute(self, command_execute_data): + + # Arrange + command = Command() + + # Act + for cmd in command_execute_data: + command.append(cmd) + tf_service = TerraformService(cmd=command) + exitcode, out, err = tf_service.execute() + + # Assert + self.assertEqual(exitcode, 0) + + def test_tf_service_cd_dir(self): + + # Arrange + command = Command() + tf_service = TerraformService(cmd=command) + + directory = os.path.join(os.getcwd(), KAOS_STATE_DIR) + command_history = [f"cd {directory}"] + + # Act + tf_service.cd_dir(directory) + + # Assert + self.assertEqual(tf_service.cmd.history, command_history) + + + + + + + + + + + + + + + + + diff --git a/cli/kaos_cli/utils/validators.py b/cli/kaos_cli/utils/validators.py index 2b2892e..2249d05 100644 --- a/cli/kaos_cli/utils/validators.py +++ b/cli/kaos_cli/utils/validators.py @@ -14,7 +14,7 @@ class EnvironmentState: """ - This class is used to validate existence of already built environmnets + This class is used to validate existence of already built environments """ def __init__(self): self.cloud = None diff --git a/cli/requirements.txt b/cli/requirements.txt index 343eb35..458b6ef 100644 --- a/cli/requirements.txt +++ b/cli/requirements.txt @@ -8,4 +8,6 @@ tqdm==4.32.2 dataclasses-json==0.2.14 kaos-model==1.0.0 requests_toolbelt==0.9.1 -configobj==5.0.6 \ No newline at end of file +configobj==5.0.6 +parameterized==0.7.1 +