AMBARI-5486. Add tests for OSCheck module (aonishuk)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/0276ecde Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/0276ecde Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/0276ecde Branch: refs/heads/trunk Commit: 0276ecdea1bd76a11300c8a589bf2f7331100c32 Parents: 3f139ac Author: Andrew Onishuk <[email protected]> Authored: Thu Apr 24 16:33:20 2014 +0300 Committer: Andrew Onishuk <[email protected]> Committed: Thu Apr 24 16:33:20 2014 +0300 ---------------------------------------------------------------------- ambari-server/src/test/python/TestOSCheck.py | 241 ++++++++++++++++++++++ 1 file changed, 241 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/0276ecde/ambari-server/src/test/python/TestOSCheck.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/python/TestOSCheck.py b/ambari-server/src/test/python/TestOSCheck.py new file mode 100644 index 0000000..a71aea4 --- /dev/null +++ b/ambari-server/src/test/python/TestOSCheck.py @@ -0,0 +1,241 @@ +# !/usr/bin/env python + +''' +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 platform +import datetime +import os +import errno +import tempfile +import sys +from unittest import TestCase +from mock.mock import patch + +from common_functions import OSCheck +import os_check_type + +with patch("platform.linux_distribution", return_value=('Suse', '11', 'Final')): + # We have to use this import HACK because the filename contains a dash + ambari_server = __import__('ambari-server') + + +class TestOSCheck(TestCase): + @patch("platform.linux_distribution") + @patch("os.path.exists") + def test_get_os_type(self, mock_exists, mock_linux_distribution): + + # 1 - Any system + mock_exists.return_value = False + mock_linux_distribution.return_value = ('my_os', '', '') + result = OSCheck.get_os_type() + self.assertEquals(result, 'my_os') + + # 2 - Negative case + mock_linux_distribution.return_value = ('', 'aaaa', 'bbbbb') + try: + result = OSCheck.get_os_type() + self.fail("Should throw exception in OSCheck.get_os_type()") + except Exception as e: + # Expected + self.assertEquals("Cannot detect os type. Exiting...", str(e)) + pass + + # 3 - path exist: '/etc/oracle-release' + mock_exists.return_value = True + mock_linux_distribution.return_value = ('some_os', '', '') + result = OSCheck.get_os_type() + self.assertEquals(result, 'oraclelinux') + + # 4 - Common system + mock_exists.return_value = False + mock_linux_distribution.return_value = ('CenToS', '', '') + result = OSCheck.get_os_type() + self.assertEquals(result, 'centos') + + + @patch("platform.linux_distribution") + @patch("os.path.exists") + def test_get_os_family(self, mock_exists, mock_linux_distribution): + + # 1 - Any system + mock_exists.return_value = False + mock_linux_distribution.return_value = ('MY_os', '', '') + result = OSCheck.get_os_family() + self.assertEquals(result, 'my_os') + + # 2 - Redhat + mock_exists.return_value = False + mock_linux_distribution.return_value = ('Centos Linux', '', '') + result = OSCheck.get_os_family() + self.assertEquals(result, 'redhat') + + # 3 - Debian + mock_exists.return_value = False + mock_linux_distribution.return_value = ('Ubuntu', '', '') + result = OSCheck.get_os_family() + self.assertEquals(result, 'debian') + + # 4 - Suse + mock_exists.return_value = False + mock_linux_distribution.return_value = ( + 'suse linux enterprise server', '', '') + result = OSCheck.get_os_family() + self.assertEquals(result, 'suse') + + mock_exists.return_value = False + mock_linux_distribution.return_value = ('SLED', '', '') + result = OSCheck.get_os_family() + self.assertEquals(result, 'suse') + + # 5 - Negative case + mock_linux_distribution.return_value = ('', '111', '2222') + try: + result = OSCheck.get_os_family() + self.fail("Should throw exception in OSCheck.get_os_family()") + except Exception as e: + # Expected + self.assertEquals("Cannot detect os type. Exiting...", str(e)) + pass + + + @patch("platform.linux_distribution") + def test_get_os_version(self, mock_linux_distribution): + + # 1 - Any system + mock_linux_distribution.return_value = ('', '123.45', '') + result = OSCheck.get_os_version() + self.assertEquals(result, '123.45') + + # 2 - Negative case + mock_linux_distribution.return_value = ('ssss', '', 'ddddd') + try: + result = OSCheck.get_os_version() + self.fail("Should throw exception in OSCheck.get_os_version()") + except Exception as e: + # Expected + self.assertEquals("Cannot detect os version. Exiting...", str(e)) + pass + + + @patch("platform.linux_distribution") + def test_get_os_major_version(self, mock_linux_distribution): + + # 1 + mock_linux_distribution.return_value = ('', '123.45.67', '') + result = OSCheck.get_os_major_version() + self.assertEquals(result, '123') + + # 2 + mock_linux_distribution.return_value = ('Suse', '11', '') + result = OSCheck.get_os_major_version() + self.assertEquals(result, '11') + + + @patch("platform.linux_distribution") + def test_get_os_release_name(self, mock_linux_distribution): + + # 1 - Any system + mock_linux_distribution.return_value = ('', '', 'MY_NEW_RELEASE') + result = OSCheck.get_os_release_name() + self.assertEquals(result, 'my_new_release') + + # 2 - Negative case + mock_linux_distribution.return_value = ('aaaa', 'bbbb', '') + try: + result = OSCheck.get_os_release_name() + self.fail("Should throw exception in OSCheck.get_os_release_name()") + except Exception as e: + # Expected + self.assertEquals("Cannot detect os release name. Exiting...", str(e)) + pass + + + @patch.object(ambari_server, "get_conf_dir") + def test_update_ambari_properties_os(self, get_conf_dir_mock): + + properties = ["server.jdbc.user.name=ambari-server\n", + "server.jdbc.database=ambari\n", + "ambari-server.user=root\n", + "server.jdbc.user.name=ambari-server\n", + "jdk.name=jdk-6u31-linux-x64.bin\n", + "jce.name=jce_policy-6.zip\n", + "server.os_type=old_sys_os6\n", + "java.home=/usr/jdk64/jdk1.6.0_31\n"] + + ambari_server.OS_FAMILY = "family_of_trolls" + ambari_server.OS_VERSION = "666" + + get_conf_dir_mock.return_value = '/etc/ambari-server/conf' + + (tf1, fn1) = tempfile.mkstemp() + (tf2, fn2) = tempfile.mkstemp() + ambari_server.AMBARI_PROPERTIES_RPMSAVE_FILE = fn1 + ambari_server.AMBARI_PROPERTIES_FILE = fn2 + + with open(ambari_server.AMBARI_PROPERTIES_RPMSAVE_FILE, 'w') as f: + for line in properties: + f.write(line) + + #Call tested method + ambari_server.update_ambari_properties() + + with open(ambari_server.AMBARI_PROPERTIES_FILE, 'r') as f: + ambari_properties_content = f.readlines() + + count = 0 + for line in ambari_properties_content: + if ( not line.startswith('#') ): + count += 1 + if (line == "server.os_type=old_sys_os6\n"): + self.fail("line=" + line) + else: + pass + + self.assertEquals(count, 8) + # Command should not fail if *.rpmsave file is missing + result = ambari_server.update_ambari_properties() + self.assertEquals(result, 0) + + @patch("platform.linux_distribution") + def test_os_type_check(self, mock_linux_distribution): + + # 1 - server and agent os compatible + mock_linux_distribution.return_value = ('aaa', '11', 'bb') + base_args = ["os_check_type.py", "aaa11"] + sys.argv = list(base_args) + + try: + os_check_type.main() + except SystemExit as e: + # exit_code=0 + self.assertEquals("0", str(e)) + + # 2 - server and agent os is not compatible + mock_linux_distribution.return_value = ('ddd', '33', 'bb') + base_args = ["os_check_type.py", "zzz_x77"] + sys.argv = list(base_args) + + try: + os_check_type.main() + self.fail("Must fail because os's not compatible.") + except Exception as e: + self.assertEquals( + "Local OS is not compatible with cluster primary OS. Please perform manual bootstrap on this host.", + str(e)) + pass
