| #!/usr/bin/env python3 |
| """ |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you under the Apache License, Version 2.0 (the |
| "License"); you may not use this file except in compliance |
| with the License. You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, software |
| distributed under the License is distributed on an "AS IS" BASIS, |
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| See the License for the specific language governing permissions and |
| limitations under the License. |
| """ |
| |
| import subprocess |
| from mock.mock import MagicMock |
| from unittest import TestCase |
| from mock.mock import patch |
| import sys |
| |
| from ambari_commons import OSCheck |
| from only_for_platform import ( |
| get_platform, |
| not_for_platform, |
| only_for_platform, |
| os_distro_value, |
| PLATFORM_WINDOWS, |
| ) |
| from mock.mock import MagicMock, patch, ANY, Mock |
| |
| with patch.object( |
| OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value) |
| ): |
| setup_agent = __import__("setupAgent") |
| |
| |
| class TestSetupAgent(TestCase): |
| @patch("sys.exit") |
| @patch("socket.socket") |
| def test_checkServerReachability(self, socket_mock, exit_mock): |
| ret = setup_agent.checkServerReachability("localhost", 8080) |
| self.assertTrue(socket_mock.called) |
| |
| s = socket_mock.return_value |
| s.connect = MagicMock() |
| |
| def side_effect(): |
| raise Exception(1, "socket is closed") |
| |
| s.connect.side_effect = side_effect |
| ret = setup_agent.checkServerReachability("localhost", 8080) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| self.assertTrue("log" in ret) |
| pass |
| |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "execOsCommand") |
| def test_configureAgent(self, execOsCommand_mock): |
| # Test if expected_hostname is passed |
| hostname = "test.hst" |
| setup_agent.configureAgent(hostname, "root") |
| cmdStr = str(execOsCommand_mock.call_args_list[0][0]) |
| self.assertTrue(hostname in cmdStr) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(setup_agent, "execOsCommand") |
| @patch("os.environ") |
| @patch("subprocess.Popen") |
| @patch("time.sleep") |
| def test_runAgent(self, sleep_mock, popen_mock, environ_mock, execOsCommand_mock): |
| expected_hostname = "test.hst" |
| passphrase = "passphrase" |
| agent_status = MagicMock() |
| agent_status.returncode = 0 |
| popen_mock.return_value = agent_status |
| execOsCommand_mock.return_value = {"log": "log", "exitstatus": 0} |
| # Test if expected_hostname is passed |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", False) |
| cmdStr = str(popen_mock.call_args_list[0][0]) |
| self.assertTrue(expected_hostname in cmdStr) |
| self.assertFalse("-v" in cmdStr) |
| self.assertEqual(ret["exitstatus"], 0) |
| self.assertTrue(sleep_mock.called) |
| self.assertEqual(execOsCommand_mock.call_count, 1) |
| |
| execOsCommand_mock.reset_mock() |
| popen_mock.reset_mock() |
| sleep_mock.reset_mock() |
| |
| # Test if verbose=True |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", True) |
| self.assertTrue(expected_hostname in cmdStr) |
| cmdStr = str(popen_mock.call_args_list[0][0]) |
| self.assertTrue("-v" in cmdStr) |
| self.assertEqual(ret["exitstatus"], 0) |
| self.assertTrue(sleep_mock.called) |
| self.assertEqual(execOsCommand_mock.call_count, 1) |
| |
| execOsCommand_mock.reset_mock() |
| popen_mock.reset_mock() |
| sleep_mock.reset_mock() |
| |
| # Key 'log' not found |
| execOsCommand_mock.return_value = {"log": "log", "exitstatus": 1} |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", False) |
| cmdStr = str(popen_mock.call_args_list[0][0]) |
| self.assertTrue(expected_hostname in cmdStr) |
| self.assertEqual(ret["exitstatus"], 0) |
| self.assertEqual(execOsCommand_mock.call_count, 3) |
| |
| execOsCommand_mock.reset_mock() |
| popen_mock.reset_mock() |
| |
| # Retcode id not 0 |
| agent_status.returncode = 2 |
| execOsCommand_mock.return_value = {"log": "log", "exitstatus": 2} |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", False) |
| cmdStr = str(popen_mock.call_args_list[0][0]) |
| self.assertTrue(expected_hostname in cmdStr) |
| self.assertEqual(ret["exitstatus"], 2) |
| execOsCommand_mock.reset_mock() |
| pass |
| |
| @only_for_platform(PLATFORM_WINDOWS) |
| @patch.object(setup_agent, "run_os_command") |
| @patch("os.environ") |
| @patch("time.sleep") |
| def test_runAgent(self, sleep_mock, environ_mock, run_os_command_mock): |
| expected_hostname = "test.hst" |
| passphrase = "passphrase" |
| run_os_command_mock.return_value = (0, "log", "") |
| # Test if expected_hostname is passed |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", False) |
| self.assertEqual(run_os_command_mock.call_count, 1) |
| cmdStr = str(run_os_command_mock.call_args_list[0][0]) |
| self.assertEqual(cmdStr, str((["cmd", "/c", "ambari-agent.cmd", "restart"],))) |
| self.assertFalse("-v" in cmdStr) |
| self.assertEqual(ret["exitstatus"], 0) |
| self.assertFalse(sleep_mock.called) |
| |
| run_os_command_mock.reset_mock() |
| sleep_mock.reset_mock() |
| |
| # Retry command |
| run_os_command_mock.side_effect = [(2, "log", "err"), (0, "log", "")] |
| ret = setup_agent.runAgent(passphrase, expected_hostname, "root", False) |
| self.assertEqual(run_os_command_mock.call_count, 2) |
| self.assertTrue("Retrying" in ret["log"][1]) |
| self.assertEqual(ret["exitstatus"], 0) |
| self.assertTrue(sleep_mock.called) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_initial_on_suse( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| getAvailableAgentPackageVersions_method, |
| ): |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": "1.1.1", |
| } |
| is_suse_family_method.return_value = True |
| is_ubuntu_family_method.return_value = False |
| |
| projectVersion = "1.1.1" |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_initial_on_ubuntu( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| getAvailableAgentPackageVersions_method, |
| ): |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": "1.1.1", |
| } |
| is_suse_family_method.return_value = False |
| is_ubuntu_family_method.return_value = True |
| |
| projectVersion = "1.1.1" |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @only_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_initial_on_suse( |
| self, findNearestAgentPackageVersion_method, getAvailableAgentPackageVersions_method |
| ): |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": "1.1.1", |
| } |
| |
| projectVersion = "1.1.1" |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_nearest_on_suse( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| ): |
| is_suse_family_method.return_value = True |
| is_ubuntu_family_method.return_value = False |
| |
| projectVersion = "" |
| nearest_version = projectVersion + "1.1.1" |
| findNearestAgentPackageVersion_method.return_value = { |
| "exitstatus": 0, |
| "log": [nearest_version, ""], |
| } |
| |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_nearest_on_ubuntu( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| ): |
| is_suse_family_method.return_value = False |
| is_ubuntu_family_method.return_value = True |
| |
| projectVersion = "" |
| nearest_version = projectVersion + "1.1.1" |
| findNearestAgentPackageVersion_method.return_value = { |
| "exitstatus": 0, |
| "log": [nearest_version, ""], |
| } |
| |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @only_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| def test_returned_optimal_version_is_nearest_on_windows( |
| self, findNearestAgentPackageVersion_method, getAvailableAgentPackageVersions_method |
| ): |
| projectVersion = "" |
| nearest_version = projectVersion + "1.1.1" |
| findNearestAgentPackageVersion_method.return_value = { |
| "exitstatus": 0, |
| "log": [nearest_version, ""], |
| } |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": nearest_version, |
| } |
| |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["exitstatus"] == 1) |
| pass |
| |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_initial( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| getAvailableAgentPackageVersions_method, |
| ): |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": "1.1.1", |
| } |
| is_suse_family_method.return_value = False |
| is_ubuntu_family_method.return_value = False |
| |
| projectVersion = "1.1.1" |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertTrue(result_version["log"] == projectVersion) |
| pass |
| |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "getAvailableAgentPackageVersions") |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "findNearestAgentPackageVersion") |
| def test_returned_optimal_version_is_default( |
| self, |
| findNearestAgentPackageVersion_method, |
| is_ubuntu_family_method, |
| is_suse_family_method, |
| getAvailableAgentPackageVersions_method, |
| ): |
| getAvailableAgentPackageVersions_method.return_value = { |
| "exitstatus": 0, |
| "log": "1.1.1", |
| } |
| is_suse_family_method.return_value = False |
| is_ubuntu_family_method.return_value = False |
| findNearestAgentPackageVersion_method.return_value = { |
| "exitstatus": 0, |
| "log": ["1.1.1.1", ""], |
| } |
| |
| projectVersion = "1.1.2" |
| result_version = setup_agent.getOptimalVersion(projectVersion) |
| |
| self.assertTrue(findNearestAgentPackageVersion_method.called) |
| self.assertEqual(result_version["exitstatus"], 1) |
| pass |
| |
| @not_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(subprocess, "Popen") |
| def test_execOsCommand(self, Popen_mock): |
| self.assertIsNone(setup_agent.execOsCommand("hostname -f")) |
| |
| @only_for_platform(PLATFORM_WINDOWS) |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(subprocess, "Popen") |
| def test_execOsCommand(self, Popen_mock): |
| p = MagicMock() |
| p.communicate.return_value = ("", "") |
| p.returncode = 0 |
| Popen_mock.return_value = p |
| retval = setup_agent.execOsCommand("hostname -f") |
| self.assertEqual(retval["exitstatus"], 0) |
| pass |
| |
| @patch("os.path.exists") |
| @patch.object(setup_agent, "get_ambari_repo_file_full_name") |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "checkVerbose") |
| @patch.object(setup_agent, "isAgentPackageAlreadyInstalled") |
| @patch.object(setup_agent, "runAgent") |
| @patch.object(setup_agent, "configureAgent") |
| @patch.object(setup_agent, "installAgent") |
| @patch("ambari_commons.OSCheck.is_suse_family") |
| @patch("ambari_commons.OSCheck.is_ubuntu_family") |
| @patch.object(setup_agent, "getOptimalVersion") |
| @patch.object(setup_agent, "checkServerReachability") |
| @patch("sys.exit") |
| @patch("os.path.dirname") |
| @patch("os.path.realpath") |
| def test_setup_agent_main( |
| self, |
| dirname_mock, |
| realpath_mock, |
| exit_mock, |
| checkServerReachability_mock, |
| getOptimalVersion_mock, |
| is_ubuntu_family_mock, |
| is_suse_family_mock, |
| installAgent_mock, |
| configureAgent_mock, |
| runAgent_mock, |
| isAgentPackageAlreadyInstalled_mock, |
| checkVerbose_mock, |
| get_ambari_repo_file_full_name_mock, |
| os_path_exists_mock, |
| ): |
| checkServerReachability_mock.return_value = {"log": "log", "exitstatus": 0} |
| installAgent_mock.return_value = {"log": "log", "exitstatus": 0} |
| configureAgent_mock.return_value = {"log": "log", "exitstatus": 0} |
| runAgent_mock.return_value = {"log": "log", "exitstatus": 0} |
| getOptimalVersion_mock.return_value = {"log": "1.1.2, 1.1.3, ", "exitstatus": 1} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertTrue(checkVerbose_mock.called) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| self.assertTrue(getOptimalVersion_mock.called) |
| |
| # A version of ambari-agent was installed or available in the repo file |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| |
| getOptimalVersion_mock.return_value = {"log": "1.1.1", "exitstatus": 0} |
| isAgentPackageAlreadyInstalled_mock.return_value = False |
| is_suse_family_mock.return_value = True |
| is_ubuntu_family_mock.return_value = False |
| get_ambari_repo_file_full_name_mock.return_value = "Ambari_Repo_File" |
| |
| def os_path_exists_side_effect(*args, **kwargs): |
| # Ambari repo file exists |
| if args[0] == get_ambari_repo_file_full_name_mock(): |
| return True |
| else: |
| return False |
| |
| os_path_exists_mock.side_effect = os_path_exists_side_effect |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertTrue(isAgentPackageAlreadyInstalled_mock.called) |
| self.assertTrue(installAgent_mock.called) |
| self.assertFalse(is_suse_family_mock.called) |
| self.assertFalse(is_ubuntu_family_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| isAgentPackageAlreadyInstalled_mock.reset_mock() |
| is_suse_family_mock.reset_mock() |
| is_ubuntu_family_mock.reset_mock() |
| installAgent_mock.reset_mock() |
| |
| getOptimalVersion_mock.return_value = {"log": "", "exitstatus": 0} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertFalse(isAgentPackageAlreadyInstalled_mock.called) |
| self.assertFalse(is_suse_family_mock.called) |
| self.assertFalse(is_ubuntu_family_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| isAgentPackageAlreadyInstalled_mock.reset_mock() |
| is_suse_family_mock.reset_mock() |
| is_ubuntu_family_mock.reset_mock() |
| installAgent_mock.reset_mock() |
| |
| is_suse_family_mock.return_value = False |
| is_ubuntu_family_mock.return_value = False |
| getOptimalVersion_mock.return_value = {"log": "1.1.1", "exitstatus": 0} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertTrue(isAgentPackageAlreadyInstalled_mock.called) |
| self.assertTrue(installAgent_mock.called) |
| self.assertFalse(is_suse_family_mock.called) |
| self.assertFalse(is_ubuntu_family_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| isAgentPackageAlreadyInstalled_mock.reset_mock() |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| isAgentPackageAlreadyInstalled_mock.reset_mock() |
| is_suse_family_mock.reset_mock() |
| is_ubuntu_family_mock.reset_mock() |
| installAgent_mock.reset_mock() |
| |
| ret = setup_agent.main( |
| ( |
| "setupAgent.py", |
| "agents_host", |
| "password", |
| "server_hostname", |
| "{ambariVersion}", |
| "8080", |
| ) |
| ) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "null", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| is_suse_family_mock.return_value = False |
| is_ubuntu_family_mock.return_value = False |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "null", "null") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 0) |
| |
| exit_mock.reset_mock() |
| # if "yum -y install --nogpgcheck ambari-agent" return not 0 result |
| installAgent_mock.return_value = {"log": "log", "exitstatus": 1} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| |
| installAgent_mock.reset_mock() |
| exit_mock.reset_mock() |
| # if suse |
| is_suse_family_mock.return_value = True |
| is_ubuntu_family_mock.return_value = False |
| # if "zypper install -y ambari-agent" return not 0 result |
| installAgent_mock.return_value = {"log": "log", "exitstatus": 1} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| |
| exit_mock.reset_mock() |
| # if ubuntu |
| is_suse_family_mock.return_value = False |
| is_ubuntu_family_mock.return_value = True |
| |
| installAgent_mock.return_value = {"log": "log", "exitstatus": 1} |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| |
| # A version of ambari-agent was installed and ambari repo file was not found |
| exit_mock.reset_mock() |
| installAgent_mock.reset_mock() |
| getOptimalVersion_mock.reset_mock() |
| os_path_exists_mock.side_effect = None |
| os_path_exists_mock.return_value = False # ambari repo file was not found |
| getOptimalVersion_mock.return_value = {"log": "1.1.1", "exitstatus": 0} |
| isAgentPackageAlreadyInstalled_mock.return_value = False |
| is_suse_family_mock.return_value = True |
| is_ubuntu_family_mock.return_value = False |
| get_ambari_repo_file_full_name_mock.return_value = "Ambari_Repo_File" |
| |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertTrue(isAgentPackageAlreadyInstalled_mock.called) |
| self.assertFalse( |
| installAgent_mock.called |
| ) # ambari repo file is missing, so installAgent() will not be called |
| self.assertFalse(is_suse_family_mock.called) |
| self.assertFalse(is_ubuntu_family_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 2) # ambari repo file not found error code |
| self.assertTrue( |
| get_ambari_repo_file_full_name_mock() in ret["log"] |
| ) # ambari repo file not found message |
| |
| # ambari-agent was not installed and ambari repo file was not found |
| exit_mock.reset_mock() |
| os_path_exists_mock.side_effect = None |
| os_path_exists_mock.return_value = False # ambari repo file not found |
| getOptimalVersion_mock.reset_mock() |
| isAgentPackageAlreadyInstalled_mock.reset_mock() |
| is_suse_family_mock.reset_mock() |
| is_ubuntu_family_mock.reset_mock() |
| installAgent_mock.reset_mock() |
| |
| getOptimalVersion_mock.return_value = { |
| "log": "", |
| "exitstatus": 1, |
| } # ambari-agent not installed |
| ret = setup_agent.main( |
| ("setupAgent.py", "agents_host", "password", "server_hostname", "1.1.1", "8080") |
| ) |
| self.assertFalse(exit_mock.called) |
| self.assertTrue(getOptimalVersion_mock.called) |
| self.assertFalse(isAgentPackageAlreadyInstalled_mock.called) |
| self.assertFalse(is_suse_family_mock.called) |
| self.assertFalse(is_ubuntu_family_mock.called) |
| self.assertTrue("exitstatus" in ret) |
| self.assertEqual(ret["exitstatus"], 1) |
| self.assertTrue( |
| get_ambari_repo_file_full_name_mock() in ret["log"] |
| ) # ambari repo file not found message |
| pass |
| |
| @patch.object(setup_agent, "execOsCommand") |
| def test_findNearestAgentPackageVersion(self, execOsCommand_mock): |
| setup_agent.findNearestAgentPackageVersion("1.1.1") |
| self.assertTrue(execOsCommand_mock.called) |
| execOsCommand_mock.reset_mock() |
| setup_agent.findNearestAgentPackageVersion("") |
| self.assertTrue(execOsCommand_mock.called) |
| pass |
| |
| @patch.object(setup_agent, "execOsCommand") |
| def test_isAgentPackageAlreadyInstalled(self, execOsCommand_mock): |
| execOsCommand_mock.return_value = {"exitstatus": 0, "log": "1.1.1"} |
| self.assertTrue(setup_agent.isAgentPackageAlreadyInstalled("1.1.1")) |
| self.assertTrue(execOsCommand_mock.called) |
| execOsCommand_mock.reset_mock() |
| execOsCommand_mock.return_value = {"exitstatus": 1, "log": "1.1.1"} |
| self.assertFalse(setup_agent.isAgentPackageAlreadyInstalled("1.1.1")) |
| self.assertTrue(execOsCommand_mock.called) |
| pass |
| |
| @patch.object(setup_agent, "execOsCommand") |
| def test_getAvailableAgentPackageVersions(self, execOsCommand_mock): |
| setup_agent.getAvailableAgentPackageVersions() |
| self.assertTrue(execOsCommand_mock.called) |
| pass |
| |
| @patch.object(OSCheck, "os_distribution", new=MagicMock(return_value=os_distro_value)) |
| @patch.object(setup_agent, "execOsCommand") |
| def test_installAgent(self, execOsCommand_mock): |
| setup_agent.installAgent("1.1.1") |
| self.assertTrue(execOsCommand_mock.called) |
| pass |
