from __future__ import unicode_literals # Copyright (C) 2014 Kristoffer Gronlund # See COPYING for license information. # # unit tests for utils.py import os import socket import re import imp import subprocess import unittest import pytest import logging from unittest import mock from itertools import chain import crmsh.utils from crmsh import utils, config, tmpfiles, constants, parallax logging.basicConfig(level=logging.DEBUG) def setup_function(): utils._ip_for_cloud = None # Mock memoize method and reload the module under test later with imp mock.patch('crmsh.utils.memoize', lambda x: x).start() imp.reload(utils) @mock.patch("crmsh.sh.ShellUtils.get_stdout_stderr") def test_print_cluster_nodes(mock_run): mock_run.return_value = (0, "data", None) utils.print_cluster_nodes() mock_run.assert_called_once_with("crm_node -l") @mock.patch("crmsh.sh.ShellUtils.get_stdout") def test_package_is_installed_local(mock_run): mock_run.return_value = (0, None) res = utils.package_is_installed("crmsh") assert res is True mock_run.assert_called_once_with("rpm -q --quiet crmsh") @mock.patch('crmsh.utils.detect_file') def test_check_file_content_included_target_not_exist(mock_detect): mock_detect.side_effect = [True, False] res = utils.check_file_content_included("file1", "file2") assert res is False mock_detect.assert_has_calls([ mock.call("file1", remote=None), mock.call("file2", remote=None) ]) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') @mock.patch('crmsh.utils.detect_file') def test_check_file_content_included(mock_detect, mock_run): mock_detect.side_effect = [True, True] mock_run.side_effect = ["data data", "data"] res = utils.check_file_content_included("file1", "file2") assert res is True mock_detect.assert_has_calls([ mock.call("file1", remote=None), mock.call("file2", remote=None) ]) mock_run.assert_has_calls([ mock.call("cat file2", host=None), mock.call("cat file1", host=None) ]) @mock.patch('re.search') @mock.patch('crmsh.sh.ShellUtils.get_stdout') def test_get_nodeid_from_name_run_None1(mock_get_stdout, mock_re_search): mock_get_stdout.return_value = (1, None) mock_re_search_inst = mock.Mock() mock_re_search.return_value = mock_re_search_inst res = utils.get_nodeid_from_name("node1") assert res is None mock_get_stdout.assert_called_once_with('crm_node -l') mock_re_search.assert_not_called() @mock.patch('re.search') @mock.patch('crmsh.sh.ShellUtils.get_stdout') def test_get_nodeid_from_name_run_None2(mock_get_stdout, mock_re_search): mock_get_stdout.return_value = (0, "172167901 node1 member\n172168231 node2 member") mock_re_search.return_value = None res = utils.get_nodeid_from_name("node111") assert res is None mock_get_stdout.assert_called_once_with('crm_node -l') mock_re_search.assert_called_once_with(r'^([0-9]+) node111 ', mock_get_stdout.return_value[1], re.M) @mock.patch('re.search') @mock.patch('crmsh.sh.ShellUtils.get_stdout') def test_get_nodeid_from_name(mock_get_stdout, mock_re_search): mock_get_stdout.return_value = (0, "172167901 node1 member\n172168231 node2 member") mock_re_search_inst = mock.Mock() mock_re_search.return_value = mock_re_search_inst mock_re_search_inst.group.return_value = '172168231' res = utils.get_nodeid_from_name("node2") assert res == '172168231' mock_get_stdout.assert_called_once_with('crm_node -l') mock_re_search.assert_called_once_with(r'^([0-9]+) node2 ', mock_get_stdout.return_value[1], re.M) mock_re_search_inst.group.assert_called_once_with(1) @mock.patch('crmsh.sh.LocalShell.get_rc_and_error') def test_check_ssh_passwd_need(mock_run): mock_run.return_value = (1, 'foo') res = utils.check_ssh_passwd_need("bob", "alice", "node1") assert res is True mock_run.assert_called_once_with( "bob", " ssh -o StrictHostKeyChecking=no -o EscapeChar=none -o ConnectTimeout=15 -T -o Batchmode=yes alice@node1 true", ) @mock.patch('crmsh.utils.list_cluster_nodes') def test_cluster_run_cmd_exception(mock_list_nodes): mock_list_nodes.return_value = None with pytest.raises(ValueError) as err: utils.cluster_run_cmd("test") assert str(err.value) == "Failed to get node list from cluster" mock_list_nodes.assert_called_once_with() @mock.patch('crmsh.utils.list_cluster_nodes') def test_list_cluster_nodes_except_me_exception(mock_list_nodes): mock_list_nodes.return_value = None with pytest.raises(ValueError) as err: utils.list_cluster_nodes_except_me() assert str(err.value) == "Failed to get node list from cluster" mock_list_nodes.assert_called_once_with() @mock.patch('crmsh.utils.this_node') @mock.patch('crmsh.utils.list_cluster_nodes') def test_list_cluster_nodes_except_me(mock_list_nodes, mock_this_node): mock_list_nodes.return_value = ["node1", "node2"] mock_this_node.return_value = "node1" res = utils.list_cluster_nodes_except_me() assert res == ["node2"] mock_list_nodes.assert_called_once_with() mock_this_node.assert_called_once_with() def test_to_ascii(): assert utils.to_ascii(None) is None assert utils.to_ascii('test') == 'test' assert utils.to_ascii(b'test') == 'test' # Test not utf-8 characters with mock.patch('traceback.print_exc') as mock_traceback: assert utils.to_ascii(b'te\xe9st') == 'test' mock_traceback.assert_called_once_with() def test_systeminfo(): assert utils.getuser() is not None assert utils.gethomedir() is not None assert utils.get_tempdir() is not None def test_shadowcib(): assert utils.get_cib_in_use() == "" utils.set_cib_in_use("foo") assert utils.get_cib_in_use() == "foo" utils.clear_cib_in_use() assert utils.get_cib_in_use() == "" def test_booleans(): truthy = ['yes', 'Yes', 'True', 'true', 'TRUE', 'YES', 'on', 'On', 'ON'] falsy = ['no', 'false', 'off', 'OFF', 'FALSE', 'nO'] not_truthy = ['', 'not', 'ONN', 'TRUETH', 'yess'] for case in chain(truthy, falsy): assert utils.verify_boolean(case) is True for case in truthy: assert utils.is_boolean_true(case) is True assert utils.is_boolean_false(case) is False assert utils.get_boolean(case) is True for case in falsy: assert utils.is_boolean_true(case) is False assert utils.is_boolean_false(case) is True assert utils.get_boolean(case, dflt=True) is False for case in not_truthy: assert utils.verify_boolean(case) is False assert utils.is_boolean_true(case) is False assert utils.is_boolean_false(case) is False assert utils.get_boolean(case) is False def test_olist(): lst = utils.olist(['B', 'C', 'A']) lst.append('f') lst.append('aA') lst.append('_') assert 'aa' in lst assert 'a' in lst assert list(lst) == ['b', 'c', 'a', 'f', 'aa', '_'] def test_add_sudo(): tmpuser = config.core.user try: config.core.user = 'root' assert utils.add_sudo('ls').startswith('sudo') config.core.user = '' assert utils.add_sudo('ls') == 'ls' finally: config.core.user = tmpuser def test_str2tmp(): txt = "This is a test string" filename = utils.str2tmp(txt) assert os.path.isfile(filename) assert open(filename).read() == txt + "\n" assert utils.file2str(filename) == txt os.unlink(filename) @mock.patch('logging.Logger.error') def test_sanity(mock_error): sane_paths = ['foo/bar', 'foo', '/foo/bar', 'foo0', 'foo_bar', 'foo-bar', '0foo', '.foo', 'foo.bar'] insane_paths = ['#foo', 'foo?', 'foo*', 'foo$', 'foo[bar]', 'foo`', "foo'", 'foo/*'] for p in sane_paths: assert utils.is_path_sane(p) for p in insane_paths: assert not utils.is_path_sane(p) sane_filenames = ['foo', '0foo', '0', '.foo'] insane_filenames = ['foo/bar'] for p in sane_filenames: assert utils.is_filename_sane(p) for p in insane_filenames: assert not utils.is_filename_sane(p) sane_names = ['foo'] insane_names = ["f'o"] for n in sane_names: assert utils.is_name_sane(n) for n in insane_names: assert not utils.is_name_sane(n) def test_nvpairs2dict(): assert utils.nvpairs2dict(['a=b', 'c=d']) == {'a': 'b', 'c': 'd'} assert utils.nvpairs2dict(['a=b=c', 'c=d']) == {'a': 'b=c', 'c': 'd'} assert utils.nvpairs2dict(['a']) == {'a': None} def test_validity(): assert utils.is_id_valid('foo0') assert not utils.is_id_valid('0foo') def test_msec(): assert utils.crm_msec('1ms') == 1 assert utils.crm_msec('1s') == 1000 assert utils.crm_msec('1us') == 0 assert utils.crm_msec('1') == 1000 assert utils.crm_msec('1m') == 60*1000 assert utils.crm_msec('1h') == 60*60*1000 def test_parse_sysconfig(): """ bsc#1129317: Fails on this line FW_SERVICES_ACCEPT_EXT="0/0,tcp,22,,hitcount=3,blockseconds=60,recentname=ssh" """ s = ''' FW_SERVICES_ACCEPT_EXT="0/0,tcp,22,,hitcount=3,blockseconds=60,recentname=ssh" ''' fd, fname = tmpfiles.create() with open(fname, 'w') as f: f.write(s) sc = utils.parse_sysconfig(fname) assert ("FW_SERVICES_ACCEPT_EXT" in sc) def test_sysconfig_set(): s = ''' FW_SERVICES_ACCEPT_EXT="0/0,tcp,22,,hitcount=3,blockseconds=60,recentname=ssh" ''' fd, fname = tmpfiles.create() with open(fname, 'w') as f: f.write(s) utils.sysconfig_set(fname, FW_SERVICES_ACCEPT_EXT="foo=bar", FOO="bar") sc = utils.parse_sysconfig(fname) assert (sc.get("FW_SERVICES_ACCEPT_EXT") == "foo=bar") assert (sc.get("FOO") == "bar") def test_sysconfig_set_bsc1145823(): s = '''# this is test #age=1000 ''' fd, fname = tmpfiles.create() with open(fname, 'w') as f: f.write(s) utils.sysconfig_set(fname, age="100") sc = utils.parse_sysconfig(fname) assert (sc.get("age") == "100") @mock.patch("crmsh.utils.IP.is_ipv6") @mock.patch("socket.socket") @mock.patch("crmsh.utils.closing") def test_check_port_open_false(mock_closing, mock_socket, mock_is_ipv6): mock_is_ipv6.return_value = False sock_inst = mock.Mock() mock_socket.return_value = sock_inst mock_closing.return_value.__enter__.return_value = sock_inst sock_inst.connect_ex.return_value = 1 assert utils.check_port_open("10.10.10.1", 22) is False mock_is_ipv6.assert_called_once_with("10.10.10.1") mock_socket.assert_called_once_with(socket.AF_INET, socket.SOCK_STREAM) mock_closing.assert_called_once_with(sock_inst) sock_inst.connect_ex.assert_called_once_with(("10.10.10.1", 22)) @mock.patch("crmsh.utils.IP.is_ipv6") @mock.patch("socket.socket") @mock.patch("crmsh.utils.closing") def test_check_port_open_true(mock_closing, mock_socket, mock_is_ipv6): mock_is_ipv6.return_value = True sock_inst = mock.Mock() mock_socket.return_value = sock_inst mock_closing.return_value.__enter__.return_value = sock_inst sock_inst.connect_ex.return_value = 0 assert utils.check_port_open("2001:db8:10::7", 22) is True mock_is_ipv6.assert_called_once_with("2001:db8:10::7") mock_socket.assert_called_once_with(socket.AF_INET6, socket.SOCK_STREAM) mock_closing.assert_called_once_with(sock_inst) sock_inst.connect_ex.assert_called_once_with(("2001:db8:10::7", 22)) def test_valid_port(): assert utils.valid_port(1) is False assert utils.valid_port(10000000) is False assert utils.valid_port(1234) is True @mock.patch("crmsh.corosync.get_value") def test_is_qdevice_configured_false(mock_get_value): mock_get_value.return_value = "ip" assert utils.is_qdevice_configured() is False mock_get_value.assert_called_once_with("quorum.device.model") @mock.patch("crmsh.corosync.get_value") def test_is_qdevice_configured_true(mock_get_value): mock_get_value.return_value = "net" assert utils.is_qdevice_configured() is True mock_get_value.assert_called_once_with("quorum.device.model") @mock.patch("crmsh.corosync.get_value") def test_is_qdevice_tls_on_false(mock_get_value): mock_get_value.return_value = "off" assert utils.is_qdevice_tls_on() is False mock_get_value.assert_called_once_with("quorum.device.net.tls") @mock.patch("crmsh.corosync.get_value") def test_is_qdevice_tls_on_true(mock_get_value): mock_get_value.return_value = "on" assert utils.is_qdevice_tls_on() is True mock_get_value.assert_called_once_with("quorum.device.net.tls") @mock.patch("crmsh.sh.ShellUtils.get_stdout") def test_get_nodeinfo_from_cmaptool_return_none(mock_get_stdout): mock_get_stdout.return_value = (1, None) assert bool(utils.get_nodeinfo_from_cmaptool()) is False mock_get_stdout.assert_called_once_with("corosync-cmapctl -b runtime.totem.pg.mrp.srp.members") @mock.patch("re.findall") @mock.patch("re.search") @mock.patch("crmsh.sh.ShellUtils.get_stdout") def test_get_nodeinfo_from_cmaptool(mock_get_stdout, mock_search, mock_findall): mock_get_stdout.return_value = (0, 'runtime.totem.pg.mrp.srp.members.1.ip (str) = r(0) ip(192.168.43.129)\nruntime.totem.pg.mrp.srp.members.2.ip (str) = r(0) ip(192.168.43.128)') match_inst1 = mock.Mock() match_inst2 = mock.Mock() mock_search.side_effect = [match_inst1, match_inst2] match_inst1.group.return_value = '1' match_inst2.group.return_value = '2' mock_findall.side_effect = [["192.168.43.129"], ["192.168.43.128"]] result = utils.get_nodeinfo_from_cmaptool() assert result['1'] == ["192.168.43.129"] assert result['2'] == ["192.168.43.128"] mock_get_stdout.assert_called_once_with("corosync-cmapctl -b runtime.totem.pg.mrp.srp.members") mock_search.assert_has_calls([ mock.call(r'members\.(.*)\.ip', 'runtime.totem.pg.mrp.srp.members.1.ip (str) = r(0) ip(192.168.43.129)'), mock.call(r'members\.(.*)\.ip', 'runtime.totem.pg.mrp.srp.members.2.ip (str) = r(0) ip(192.168.43.128)') ]) match_inst1.group.assert_called_once_with(1) match_inst2.group.assert_called_once_with(1) mock_findall.assert_has_calls([ mock.call(r'[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}', 'runtime.totem.pg.mrp.srp.members.1.ip (str) = r(0) ip(192.168.43.129)'), mock.call(r'[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}', 'runtime.totem.pg.mrp.srp.members.2.ip (str) = r(0) ip(192.168.43.128)') ]) @mock.patch("crmsh.utils.get_nodeinfo_from_cmaptool") @mock.patch("crmsh.service_manager.ServiceManager.service_is_active") def test_valid_nodeid_false_service_not_active(mock_is_active, mock_nodeinfo): mock_is_active.return_value = False assert utils.valid_nodeid("3") is False mock_is_active.assert_called_once_with('corosync.service') mock_nodeinfo.assert_not_called() @mock.patch("crmsh.utils.get_nodeinfo_from_cmaptool") @mock.patch("crmsh.service_manager.ServiceManager.service_is_active") def test_valid_nodeid_false(mock_is_active, mock_nodeinfo): mock_is_active.return_value = True mock_nodeinfo.return_value = {'1': ["10.10.10.1"], "2": ["20.20.20.2"]} assert utils.valid_nodeid("3") is False mock_is_active.assert_called_once_with('corosync.service') mock_nodeinfo.assert_called_once_with() @mock.patch("crmsh.utils.get_nodeinfo_from_cmaptool") @mock.patch("crmsh.service_manager.ServiceManager.service_is_active") def test_valid_nodeid_true(mock_is_active, mock_nodeinfo): mock_is_active.return_value = True mock_nodeinfo.return_value = {'1': ["10.10.10.1"], "2": ["20.20.20.2"]} assert utils.valid_nodeid("2") is True mock_is_active.assert_called_once_with('corosync.service') mock_nodeinfo.assert_called_once_with() @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_aws_false(mock_run): mock_run.side_effect = ["test", "test"] assert utils.detect_aws() is False mock_run.assert_has_calls([ mock.call("dmidecode -s system-version"), mock.call("dmidecode -s system-manufacturer") ]) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_aws_xen(mock_run): mock_run.side_effect = ["4.2.amazon", "Xen"] assert utils.detect_aws() is True mock_run.assert_has_calls([ mock.call("dmidecode -s system-version"), mock.call("dmidecode -s system-manufacturer") ]) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_aws_kvm(mock_run): mock_run.side_effect = ["Not Specified", "Amazon EC2"] assert utils.detect_aws() is True mock_run.assert_has_calls([ mock.call("dmidecode -s system-version"), mock.call("dmidecode -s system-manufacturer") ]) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_azure_false(mock_run): mock_run.side_effect = ["test", "test"] assert utils.detect_azure() is False mock_run.assert_has_calls([ mock.call("dmidecode -s system-manufacturer"), mock.call("dmidecode -s chassis-asset-tag") ]) @mock.patch("crmsh.utils._cloud_metadata_request") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_azure_microsoft_corporation(mock_run, mock_request): mock_run.side_effect = ["microsoft corporation", "test"] mock_request.return_value = "data" assert utils.detect_azure() is True mock_run.assert_has_calls([ mock.call("dmidecode -s system-manufacturer"), mock.call("dmidecode -s chassis-asset-tag") ]) @mock.patch("crmsh.utils._cloud_metadata_request") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_azure_chassis(mock_run, mock_request): mock_run.side_effect = ["test", "7783-7084-3265-9085-8269-3286-77"] mock_request.return_value = "data" assert utils.detect_azure() is True mock_run.assert_has_calls([ mock.call("dmidecode -s system-manufacturer"), mock.call("dmidecode -s chassis-asset-tag") ]) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_gcp_false(mock_run): mock_run.return_value = "test" assert utils.detect_gcp() is False mock_run.assert_called_once_with("dmidecode -s bios-vendor") @mock.patch("crmsh.utils._cloud_metadata_request") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_detect_gcp(mock_run, mock_request): mock_run.return_value = "Google instance" mock_request.return_value = "data" assert utils.detect_gcp() is True mock_run.assert_called_once_with("dmidecode -s bios-vendor") @mock.patch("crmsh.utils.is_program") def test_detect_cloud_no_cmd(mock_is_program): mock_is_program.return_value = False assert utils.detect_cloud() is None mock_is_program.assert_called_once_with("dmidecode") @mock.patch("crmsh.utils.detect_aws") @mock.patch("crmsh.utils.is_program") def test_detect_cloud_aws(mock_is_program, mock_aws): mock_is_program.return_value = True mock_aws.return_value = True assert utils.detect_cloud() == constants.CLOUD_AWS mock_is_program.assert_called_once_with("dmidecode") mock_aws.assert_called_once_with() @mock.patch("crmsh.utils.detect_azure") @mock.patch("crmsh.utils.detect_aws") @mock.patch("crmsh.utils.is_program") def test_detect_cloud_azure(mock_is_program, mock_aws, mock_azure): mock_is_program.return_value = True mock_aws.return_value = False mock_azure.return_value = True assert utils.detect_cloud() == constants.CLOUD_AZURE mock_is_program.assert_called_once_with("dmidecode") mock_aws.assert_called_once_with() mock_azure.assert_called_once_with() @mock.patch("crmsh.utils.detect_gcp") @mock.patch("crmsh.utils.detect_azure") @mock.patch("crmsh.utils.detect_aws") @mock.patch("crmsh.utils.is_program") def test_detect_cloud_gcp(mock_is_program, mock_aws, mock_azure, mock_gcp): mock_is_program.return_value = True mock_aws.return_value = False mock_azure.return_value = False mock_gcp.return_value = True assert utils.detect_cloud() == constants.CLOUD_GCP mock_is_program.assert_called_once_with("dmidecode") mock_aws.assert_called_once_with() mock_azure.assert_called_once_with() mock_gcp.assert_called_once_with() @mock.patch("crmsh.sh.ShellUtils.get_stdout") def test_interface_choice(mock_get_stdout): ip_a_output = """ 1: lo: mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000 link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 inet 127.0.0.1/8 scope host lo valid_lft forever preferred_lft forever inet6 ::1/128 scope host valid_lft forever preferred_lft forever 2: enp1s0: mtu 1500 qdisc pfifo_fast state UP group default qlen 1000 link/ether 52:54:00:9e:1b:4f brd ff:ff:ff:ff:ff:ff inet 192.168.122.241/24 brd 192.168.122.255 scope global enp1s0 valid_lft forever preferred_lft forever inet6 fe80::5054:ff:fe9e:1b4f/64 scope link valid_lft forever preferred_lft forever 3: br-933fa0e1438c: mtu 1500 qdisc noqueue state UP group default link/ether 9e:fe:24:df:59:49 brd ff:ff:ff:ff:ff:ff inet 10.10.10.1/24 brd 10.10.10.255 scope global br-933fa0e1438c valid_lft forever preferred_lft forever inet6 fe80::9cfe:24ff:fedf:5949/64 scope link valid_lft forever preferred_lft forever 4: veth3fff6e9@if7: mtu 1500 qdisc noqueue master docker0 state UP group default link/ether 1e:2c:b3:73:6b:42 brd ff:ff:ff:ff:ff:ff link-netnsid 0 inet6 fe80::1c2c:b3ff:fe73:6b42/64 scope link valid_lft forever preferred_lft forever valid_lft forever preferred_lft forever """ mock_get_stdout.return_value = (0, ip_a_output) assert utils.interface_choice() == ["enp1s0", "br-933fa0e1438c", "veth3fff6e9"] mock_get_stdout.assert_called_once_with("ip a") class TestIP(unittest.TestCase): """ Unitary tests for class utils.IP """ @classmethod def setUpClass(cls): """ Global setUp. """ def setUp(self): """ Test setUp. """ self.ip_inst = utils.IP("10.10.10.1") def tearDown(self): """ Test tearDown. """ @classmethod def tearDownClass(cls): """ Global tearDown. """ @mock.patch('ipaddress.ip_address') def test_ip_address(self, mock_ip_address): mock_ip_address_inst = mock.Mock() mock_ip_address.return_value = mock_ip_address_inst self.ip_inst.ip_address mock_ip_address.assert_called_once_with("10.10.10.1") @mock.patch('crmsh.utils.IP.ip_address', new_callable=mock.PropertyMock) def test_version(self, mock_ip_address): mock_ip_address_inst = mock.Mock(version=4) mock_ip_address.return_value = mock_ip_address_inst res = self.ip_inst.version self.assertEqual(res, mock_ip_address_inst.version) mock_ip_address.assert_called_once_with() @mock.patch('crmsh.utils.IP.ip_address', new_callable=mock.PropertyMock) def test_is_mcast(self, mock_ip_address): mock_ip_address_inst = mock.Mock(is_multicast=False) mock_ip_address.return_value = mock_ip_address_inst res = utils.IP.is_mcast("10.10.10.1") self.assertEqual(res, False) mock_ip_address.assert_called_once_with() @mock.patch('crmsh.utils.IP.version', new_callable=mock.PropertyMock) def test_is_ipv6(self, mock_version): mock_version.return_value = 4 res = utils.IP.is_ipv6("10.10.10.1") self.assertEqual(res, False) mock_version.assert_called_once_with() @mock.patch('crmsh.utils.IP.ip_address', new_callable=mock.PropertyMock) def test_is_loopback(self, mock_ip_address): mock_ip_address_inst = mock.Mock(is_loopback=False) mock_ip_address.return_value = mock_ip_address_inst res = self.ip_inst.is_loopback self.assertEqual(res, mock_ip_address_inst.is_loopback) mock_ip_address.assert_called_once_with() class TestInterface(unittest.TestCase): """ Unitary tests for class utils.Interface """ @classmethod def setUpClass(cls): """ Global setUp. """ def setUp(self): """ Test setUp. """ self.interface = utils.Interface("10.10.10.123/24") def tearDown(self): """ Test tearDown. """ @classmethod def tearDownClass(cls): """ Global tearDown. """ def test_ip_with_mask(self): assert self.interface.ip_with_mask == "10.10.10.123/24" @mock.patch('ipaddress.ip_interface') def test_ip_interface(self, mock_ip_interface): mock_ip_interface_inst = mock.Mock() mock_ip_interface.return_value = mock_ip_interface_inst self.interface.ip_interface mock_ip_interface.assert_called_once_with("10.10.10.123/24") @mock.patch('crmsh.utils.Interface.ip_interface', new_callable=mock.PropertyMock) def test_network(self, mock_ip_interface): mock_ip_interface_inst = mock.Mock() mock_ip_interface.return_value = mock_ip_interface_inst mock_ip_interface_inst.network = mock.Mock(network_address="10.10.10.0") assert self.interface.network == "10.10.10.0" mock_ip_interface.assert_called_once_with() class TestInterfacesInfo(unittest.TestCase): """ Unitary tests for class utils.InterfacesInfo """ network_output_error = """1: lo inet 127.0.0.1/8 scope host lo\ valid_lft forever preferred_lft forever 2: enp1s0 inet 192.168.122.241/24 brd 192.168.122.255 scope global enp1s0 61: tun0 inet 10.163.45.46 peer 10.163.45.45/32 scope global tun0""" @classmethod def setUpClass(cls): """ Global setUp. """ def setUp(self): """ Test setUp. """ self.interfaces_info = utils.InterfacesInfo() self.interfaces_info_with_second_hb = utils.InterfacesInfo(second_heartbeat=True) self.interfaces_info_with_custom_nic = utils.InterfacesInfo(second_heartbeat=True, custom_nic_list=['eth1']) self.interfaces_info_with_wrong_nic = utils.InterfacesInfo(custom_nic_list=['eth7']) self.interfaces_info_fake = utils.InterfacesInfo() self.interfaces_info_fake._nic_info_dict = { "eth0": [mock.Mock(ip="10.10.10.1", network="10.10.10.0"), mock.Mock(ip="10.10.10.2", network="10.10.10.0")], "eth1": [mock.Mock(ip="20.20.20.1", network="20.20.20.0")] } self.interfaces_info_fake._default_nic_list = ["eth7"] def tearDown(self): """ Test tearDown. """ @classmethod def tearDownClass(cls): """ Global tearDown. """ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_get_interfaces_info_no_address(self, mock_run): only_lo = "1: lo inet 127.0.0.1/8 scope host lo\ valid_lft forever preferred_lft forever" mock_run.return_value = (0, only_lo, None) with self.assertRaises(ValueError) as err: self.interfaces_info.get_interfaces_info() self.assertEqual("No address configured", str(err.exception)) mock_run.assert_called_once_with("ip -4 -o addr show") @mock.patch('crmsh.utils.Interface') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_get_interfaces_info_one_addr(self, mock_run, mock_interface): mock_run.return_value = (0, self.network_output_error, None) mock_interface_inst_1 = mock.Mock(is_loopback=True, is_link_local=False) mock_interface_inst_2 = mock.Mock(is_loopback=False, is_link_local=False) mock_interface.side_effect = [mock_interface_inst_1, mock_interface_inst_2] with self.assertRaises(ValueError) as err: self.interfaces_info_with_second_hb.get_interfaces_info() self.assertEqual("Cannot configure second heartbeat, since only one address is available", str(err.exception)) mock_run.assert_called_once_with("ip -4 -o addr show") mock_interface.assert_has_calls([ mock.call("127.0.0.1/8"), mock.call("192.168.122.241/24") ]) def test_nic_list(self): res = self.interfaces_info_fake.nic_list self.assertEqual(res, ["eth0", "eth1"]) def test_interface_list(self): res = self.interfaces_info_fake.interface_list assert len(res) == 3 @mock.patch('crmsh.utils.InterfacesInfo.interface_list', new_callable=mock.PropertyMock) def test_ip_list(self, mock_interface_list): mock_interface_list.return_value = [ mock.Mock(ip="10.10.10.1"), mock.Mock(ip="10.10.10.2") ] res = self.interfaces_info_fake.ip_list self.assertEqual(res, ["10.10.10.1", "10.10.10.2"]) mock_interface_list.assert_called_once_with() @mock.patch('crmsh.utils.InterfacesInfo.ip_list', new_callable=mock.PropertyMock) @mock.patch('crmsh.utils.InterfacesInfo.get_interfaces_info') def test_get_local_ip_list(self, mock_get_info, mock_ip_list): mock_ip_list.return_value = ["10.10.10.1", "10.10.10.2"] res = utils.InterfacesInfo.get_local_ip_list(False) self.assertEqual(res, mock_ip_list.return_value) mock_get_info.assert_called_once_with() mock_ip_list.assert_called_once_with() @mock.patch('crmsh.utils.InterfacesInfo.ip_list', new_callable=mock.PropertyMock) @mock.patch('crmsh.utils.IP.is_ipv6') @mock.patch('crmsh.utils.InterfacesInfo.get_interfaces_info') def test_ip_in_local(self, mock_get_info, mock_is_ipv6, mock_ip_list): mock_is_ipv6.return_value = False mock_ip_list.return_value = ["10.10.10.1", "10.10.10.2"] res = utils.InterfacesInfo.ip_in_local("10.10.10.1") assert res is True mock_get_info.assert_called_once_with() mock_ip_list.assert_called_once_with() mock_is_ipv6.assert_called_once_with("10.10.10.1") @mock.patch('crmsh.utils.InterfacesInfo.interface_list', new_callable=mock.PropertyMock) def test_network_list(self, mock_interface_list): mock_interface_list.return_value = [ mock.Mock(network="10.10.10.0"), mock.Mock(network="20.20.20.0") ] res = self.interfaces_info.network_list self.assertEqual(res, list(set(["10.10.10.0", "20.20.20.0"]))) mock_interface_list.assert_called_once_with() def test_nic_first_ip(self): res = self.interfaces_info_fake._nic_first_ip("eth0") self.assertEqual(res, "10.10.10.1") @mock.patch('crmsh.utils.InterfacesInfo.nic_list', new_callable=mock.PropertyMock) @mock.patch('logging.Logger.warning') @mock.patch('crmsh.utils.InterfacesInfo.get_interfaces_info') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_get_default_nic_list_from_route_no_default(self, mock_run, mock_get_interfaces_info, mock_warn, mock_nic_list): output = """10.10.10.0/24 dev eth1 proto kernel scope link src 10.10.10.51 20.20.20.0/24 dev eth2 proto kernel scope link src 20.20.20.51""" mock_run.return_value = (0, output, None) mock_nic_list.side_effect = [["eth0", "eth1"], ["eth0", "eth1"]] res = self.interfaces_info.get_default_nic_list_from_route() self.assertEqual(res, ["eth0"]) mock_run.assert_called_once_with("ip -o route show") mock_warn.assert_called_once_with("No default route configured. Using the first found nic") mock_nic_list.assert_has_calls([mock.call(), mock.call()]) @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_get_default_nic_list_from_route(self, mock_run): output = """default via 192.168.122.1 dev eth8 proto dhcp 10.10.10.0/24 dev eth1 proto kernel scope link src 10.10.10.51 20.20.20.0/24 dev eth2 proto kernel scope link src 20.20.20.51 192.168.122.0/24 dev eth8 proto kernel scope link src 192.168.122.120""" mock_run.return_value = (0, output, None) res = self.interfaces_info.get_default_nic_list_from_route() self.assertEqual(res, ["eth8"]) mock_run.assert_called_once_with("ip -o route show") @mock.patch('crmsh.utils.InterfacesInfo.nic_list', new_callable=mock.PropertyMock) def test_get_default_ip_list_failed_detect(self, mock_nic_list): mock_nic_list.side_effect = [["eth0", "eth1"], ["eth0", "eth1"]] with self.assertRaises(ValueError) as err: self.interfaces_info_with_wrong_nic.get_default_ip_list() self.assertEqual("Failed to detect IP address for eth7", str(err.exception)) mock_nic_list.assert_has_calls([mock.call(), mock.call()]) @mock.patch('crmsh.utils.InterfacesInfo._nic_first_ip') @mock.patch('crmsh.utils.InterfacesInfo.nic_list', new_callable=mock.PropertyMock) def test_get_default_ip_list(self, mock_nic_list, mock_first_ip): mock_nic_list.side_effect = [["eth0", "eth1"], ["eth0", "eth1"], ["eth0", "eth1"]] mock_first_ip.side_effect = ["10.10.10.1", "20.20.20.1"] res = self.interfaces_info_with_custom_nic.get_default_ip_list() self.assertEqual(res, ["10.10.10.1", "20.20.20.1"]) mock_nic_list.assert_has_calls([mock.call(), mock.call(), mock.call()]) mock_first_ip.assert_has_calls([mock.call("eth1"), mock.call("eth0")]) @mock.patch("crmsh.utils.get_nodeid_from_name") def test_get_iplist_from_name_no_nodeid(mock_get_nodeid): mock_get_nodeid.return_value = None res = utils.get_iplist_from_name("test") assert res == [] mock_get_nodeid.assert_called_once_with("test") @mock.patch("crmsh.utils.get_nodeinfo_from_cmaptool") @mock.patch("crmsh.utils.get_nodeid_from_name") def test_get_iplist_from_name_no_nodeinfo(mock_get_nodeid, mock_get_nodeinfo): mock_get_nodeid.return_value = "1" mock_get_nodeinfo.return_value = None res = utils.get_iplist_from_name("test") assert res == [] mock_get_nodeid.assert_called_once_with("test") mock_get_nodeinfo.assert_called_once_with() @mock.patch("crmsh.utils.get_nodeinfo_from_cmaptool") @mock.patch("crmsh.utils.get_nodeid_from_name") def test_get_iplist_from_name(mock_get_nodeid, mock_get_nodeinfo): mock_get_nodeid.return_value = "1" mock_get_nodeinfo.return_value = {"1": ["10.10.10.1"], "2": ["10.10.10.2"]} res = utils.get_iplist_from_name("test") assert res == ["10.10.10.1"] mock_get_nodeid.assert_called_once_with("test") mock_get_nodeinfo.assert_called_once_with() @mock.patch("crmsh.sh.ShellUtils.get_stdout_stderr") def test_ping_node(mock_run): mock_run.return_value = (1, None, "error data") with pytest.raises(ValueError) as err: utils.ping_node("node_unreachable") assert str(err.value) == 'host "node_unreachable" is unreachable: error data' mock_run.assert_called_once_with("ping -c 1 node_unreachable") def test_calculate_quorate_status(): assert utils.calculate_quorate_status(3, 2) is True assert utils.calculate_quorate_status(3, 1) is False @mock.patch("crmsh.sh.ClusterShell.get_stdout_or_raise_error") def test_get_quorum_votes_dict(mock_run): mock_run.return_value = """ Votequorum information ---------------------- Expected votes: 1 Highest expected: 1 Total votes: 1 Quorum: 1 Flags: Quorate """ res = utils.get_quorum_votes_dict() assert res == {'Expected': '1', 'Total': '1'} mock_run.assert_called_once_with("corosync-quorumtool -s", None, success_exit_status={0, 2}) def test_re_split_string(): assert utils.re_split_string('[; ]', "/dev/sda1; /dev/sdb1 ; ") == ["/dev/sda1", "/dev/sdb1"] assert utils.re_split_string('[; ]', "/dev/sda1 ") == ["/dev/sda1"] @mock.patch('crmsh.utils.get_dev_info') def test_has_dev_partitioned(mock_get_dev_info): mock_get_dev_info.return_value = """ disk part """ res = utils.has_dev_partitioned("/dev/sda1") assert res is True mock_get_dev_info.assert_called_once_with("/dev/sda1", "NAME", peer=None) @mock.patch('crmsh.utils.get_dev_uuid') def test_compare_uuid_with_peer_dev_cannot_find_local(mock_get_dev_uuid): mock_get_dev_uuid.return_value = "" with pytest.raises(ValueError) as err: utils.compare_uuid_with_peer_dev(["/dev/sdb1"], "node2") assert str(err.value) == "Cannot find UUID for /dev/sdb1 on local" mock_get_dev_uuid.assert_called_once_with("/dev/sdb1") @mock.patch('crmsh.utils.get_dev_uuid') def test_compare_uuid_with_peer_dev_cannot_find_peer(mock_get_dev_uuid): mock_get_dev_uuid.side_effect = ["1234", ""] with pytest.raises(ValueError) as err: utils.compare_uuid_with_peer_dev(["/dev/sdb1"], "node2") assert str(err.value) == "Cannot find UUID for /dev/sdb1 on node2" mock_get_dev_uuid.assert_has_calls([ mock.call("/dev/sdb1"), mock.call("/dev/sdb1", "node2") ]) @mock.patch('crmsh.utils.get_dev_uuid') def test_compare_uuid_with_peer_dev(mock_get_dev_uuid): mock_get_dev_uuid.side_effect = ["1234", "5678"] with pytest.raises(ValueError) as err: utils.compare_uuid_with_peer_dev(["/dev/sdb1"], "node2") assert str(err.value) == "UUID of /dev/sdb1 not same with peer node2" mock_get_dev_uuid.assert_has_calls([ mock.call("/dev/sdb1"), mock.call("/dev/sdb1", "node2") ]) @mock.patch('crmsh.utils.get_dev_info') def test_is_dev_used_for_lvm(mock_dev_info): mock_dev_info.return_value = "lvm" res = utils.is_dev_used_for_lvm("/dev/sda1") assert res is True mock_dev_info.assert_called_once_with("/dev/sda1", "TYPE", peer=None) @mock.patch('crmsh.utils.get_dev_info') def test_is_dev_a_plain_raw_disk_or_partition(mock_dev_info): mock_dev_info.return_value = "raid1\nlvm" res = utils.is_dev_a_plain_raw_disk_or_partition("/dev/md127") assert res is False mock_dev_info.assert_called_once_with("/dev/md127", "TYPE", peer=None) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_get_dev_info(mock_run): mock_run.return_value = "data" res = utils.get_dev_info("/dev/sda1", "TYPE") assert res == "data" mock_run.assert_called_once_with("lsblk -fno TYPE /dev/sda1", None) @mock.patch('crmsh.utils.get_dev_info') def test_get_dev_fs_type(mock_get_info): mock_get_info.return_value = "data" res = utils.get_dev_fs_type("/dev/sda1") assert res == "data" mock_get_info.assert_called_once_with("/dev/sda1", "FSTYPE", peer=None) @mock.patch('crmsh.utils.get_dev_info') def test_get_dev_uuid(mock_get_info): mock_get_info.return_value = "uuid" res = utils.get_dev_uuid("/dev/sda1") assert res == "uuid" mock_get_info.assert_called_once_with("/dev/sda1", "UUID", peer=None) @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_get_pe_number_except(mock_run): mock_run.return_value = "data" with pytest.raises(ValueError) as err: utils.get_pe_number("vg1") assert str(err.value) == "Cannot find PE on VG(vg1)" mock_run.assert_called_once_with("vgdisplay vg1") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_get_pe_number(mock_run): mock_run.return_value = """ PE Size 4.00 MiB Total PE 1534 Alloc PE / Size 1534 / 5.99 GiB """ res = utils.get_pe_number("vg1") assert res == 1534 mock_run.assert_called_once_with("vgdisplay vg1") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_get_all_vg_name(mock_run): mock_run.return_value = """ --- Volume group --- VG Name ocfs2-vg System ID """ res = utils.get_all_vg_name() assert res == ["ocfs2-vg"] mock_run.assert_called_once_with("vgdisplay") @mock.patch('crmsh.utils.randomword') def test_gen_unused_id(mock_rand): mock_rand.return_value = "1234xxxx" res = utils.gen_unused_id(["test-id"], "test-id") assert res == "test-id-1234xxxx" mock_rand.assert_called_once_with(6) @mock.patch('random.choice') def test_randomword(mock_rand): import string mock_rand.side_effect = ['z', 'f', 'k', 'e', 'c', 'd'] res = utils.randomword() assert res == "zfkecd" mock_rand.assert_has_calls([mock.call(string.ascii_lowercase) for x in range(6)]) @mock.patch('crmsh.cibconfig.cib_factory') def test_all_exist_id(mock_cib): mock_cib.refresh = mock.Mock() mock_cib.id_list = mock.Mock() mock_cib.id_list.return_value = ['1', '2'] res = utils.all_exist_id() assert res == ['1', '2'] mock_cib.id_list.assert_called_once_with() mock_cib.refresh.assert_called_once_with() @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_has_mount_point_used(mock_run): mock_run.return_value = """ /dev/vda2 on /usr/local type btrfs (rw,relatime,space_cache,subvolid=259,subvol=/@/usr/local) /dev/vda2 on /opt type btrfs (rw,relatime,space_cache,subvolid=263,subvol=/@/opt) /dev/vda2 on /var/lib/docker/btrfs type btrfs (rw,relatime,space_cache,subvolid=258,subvol=/@/var) """ res = utils.has_mount_point_used("/opt") assert res is True mock_run.assert_called_once_with("mount") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_has_disk_mounted(mock_run): mock_run.return_value = """ /dev/vda2 on /usr/local type btrfs (rw,relatime,space_cache,subvolid=259,subvol=/@/usr/local) /dev/vda2 on /opt type btrfs (rw,relatime,space_cache,subvolid=263,subvol=/@/opt) /dev/vda2 on /var/lib/docker/btrfs type btrfs (rw,relatime,space_cache,subvolid=258,subvol=/@/var) """ res = utils.has_disk_mounted("/dev/vda2") assert res is True mock_run.assert_called_once_with("mount") @mock.patch('crmsh.sbd.SBDManager.is_using_diskless_sbd') @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_has_stonith_running(mock_run, mock_diskless): mock_run.return_value = """ stonith-sbd 1 fence device found """ mock_diskless.return_value = True res = utils.has_stonith_running() assert res is True mock_run.assert_called_once_with("stonith_admin -L") mock_diskless.assert_called_once_with() @mock.patch('crmsh.utils.S_ISBLK') @mock.patch('os.stat') def test_is_block_device_error(mock_stat, mock_isblk): mock_stat_inst = mock.Mock(st_mode=12345) mock_stat.return_value = mock_stat_inst mock_isblk.side_effect = OSError res = utils.is_block_device("/dev/sda1") assert res is False mock_stat.assert_called_once_with("/dev/sda1") mock_isblk.assert_called_once_with(12345) @mock.patch('crmsh.utils.S_ISBLK') @mock.patch('os.stat') def test_is_block_device(mock_stat, mock_isblk): mock_stat_inst = mock.Mock(st_mode=12345) mock_stat.return_value = mock_stat_inst mock_isblk.return_value = True res = utils.is_block_device("/dev/sda1") assert res is True mock_stat.assert_called_once_with("/dev/sda1") mock_isblk.assert_called_once_with(12345) @mock.patch('crmsh.utils.ping_node') @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_check_all_nodes_reachable(mock_run, mock_ping): mock_run.return_value = "1084783297 15sp2-1 member" utils.check_all_nodes_reachable() mock_run.assert_called_once_with("crm_node -l") mock_ping.assert_called_once_with("15sp2-1") @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_detect_virt(mock_run): mock_run.return_value = (0, None, None) assert utils.detect_virt() is True mock_run.assert_called_once_with("systemd-detect-virt") @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_is_standby(mock_run): mock_run.return_value = """ Node List: * Node 15sp2-1: standby """ assert utils.is_standby("15sp2-1") is True mock_run.assert_called_once_with("crm_mon -1") @mock.patch('crmsh.sh.cluster_shell') def test_get_dlm_option_dict(mock_run): mock_run_inst = mock.Mock() mock_run.return_value = mock_run_inst mock_run_inst.get_stdout_or_raise_error.return_value = """ key1=value1 key2=value2 """ res_dict = utils.get_dlm_option_dict() assert res_dict == { "key1": "value1", "key2": "value2" } mock_run_inst.get_stdout_or_raise_error.assert_called_once_with("dlm_tool dump_config", None) @mock.patch('crmsh.utils.get_dlm_option_dict') def test_set_dlm_option_exception(mock_get_dict): mock_get_dict.return_value = { "key1": "value1", "key2": "value2" } with pytest.raises(ValueError) as err: utils.set_dlm_option(name="xin") assert str(err.value) == '"name" is not dlm config option' @mock.patch('crmsh.sh.cluster_shell') @mock.patch('crmsh.utils.get_dlm_option_dict') def test_set_dlm_option(mock_get_dict, mock_run): mock_run_inst = mock.Mock() mock_run.return_value = mock_run_inst mock_get_dict.return_value = { "key1": "value1", "key2": "value2" } utils.set_dlm_option(key2="test") mock_run_inst.get_stdout_or_raise_error.assert_called_once_with('dlm_tool set_config "key2=test"', None) @mock.patch('crmsh.utils.has_resource_configured') def test_is_dlm_configured(mock_configured): mock_configured.return_value = True assert utils.is_dlm_configured() is True mock_configured.assert_called_once_with(constants.DLM_CONTROLD_RA, peer=None) @mock.patch('crmsh.sh.cluster_shell') def test_is_quorate_exception(mock_run): mock_run_inst = mock.Mock() mock_run.return_value = mock_run_inst mock_run_inst.get_stdout_or_raise_error.return_value = "data" with pytest.raises(ValueError) as err: utils.is_quorate() assert str(err.value) == "Failed to get quorate status from corosync-quorumtool" mock_run_inst.get_stdout_or_raise_error.assert_called_once_with("corosync-quorumtool -s", None, success_exit_status={0, 2}) @mock.patch('crmsh.sh.cluster_shell') def test_is_quorate(mock_run): mock_run_inst = mock.Mock() mock_run.return_value = mock_run_inst mock_run_inst.get_stdout_or_raise_error.return_value = """ Ring ID: 1084783297/440 Quorate: Yes """ assert utils.is_quorate() is True mock_run_inst.get_stdout_or_raise_error.assert_called_once_with("corosync-quorumtool -s", None, success_exit_status={0, 2}) @mock.patch('crmsh.utils.etree.fromstring') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_list_cluster_nodes_none(mock_run, mock_etree): mock_run.return_value = (0, "data", None) mock_etree.return_value = None res = utils.list_cluster_nodes() assert res is None mock_run.assert_called_once_with(constants.CIB_QUERY, no_reg=False) mock_etree.assert_called_once_with("data") @mock.patch('crmsh.utils.etree.fromstring') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_list_cluster_nodes_none_no_reg(mock_run, mock_etree): mock_run.return_value = (0, "data", None) mock_etree.return_value = None res = utils.list_cluster_nodes(no_reg=True) assert res is None mock_run.assert_called_once_with(constants.CIB_QUERY, no_reg=True) mock_etree.assert_called_once_with("data") @mock.patch('os.path.isfile') @mock.patch('os.getenv') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_list_cluster_nodes_cib_not_exist(mock_run, mock_env, mock_isfile): mock_run.return_value = (1, None, None) mock_env.return_value = constants.CIB_RAW_FILE mock_isfile.return_value = False res = utils.list_cluster_nodes() assert res is None mock_run.assert_called_once_with(constants.CIB_QUERY, no_reg=False) mock_env.assert_called_once_with("CIB_file", constants.CIB_RAW_FILE) mock_isfile.assert_called_once_with(constants.CIB_RAW_FILE) @mock.patch('crmsh.xmlutil.file2cib_elem') @mock.patch('os.path.isfile') @mock.patch('os.getenv') @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_list_cluster_nodes(mock_run, mock_env, mock_isfile, mock_file2elem): mock_run.return_value = (1, None, None) mock_env.return_value = constants.CIB_RAW_FILE mock_isfile.return_value = True mock_cib_inst = mock.Mock() mock_file2elem.return_value = mock_cib_inst mock_node_inst1 = mock.Mock() mock_node_inst2 = mock.Mock() mock_node_inst1.get.side_effect = ["node1", "remote"] mock_node_inst2.get.side_effect = ["node2", "member"] mock_cib_inst.xpath.side_effect = [[mock_node_inst1, mock_node_inst2], "data"] res = utils.list_cluster_nodes() assert res == ["node2"] mock_run.assert_called_once_with(constants.CIB_QUERY, no_reg=False) mock_env.assert_called_once_with("CIB_file", constants.CIB_RAW_FILE) mock_isfile.assert_called_once_with(constants.CIB_RAW_FILE) mock_file2elem.assert_called_once_with(constants.CIB_RAW_FILE) mock_cib_inst.xpath.assert_has_calls([ mock.call(constants.XML_NODE_PATH), mock.call("//primitive[@id='node1']/instance_attributes/nvpair[@name='server']") ]) @mock.patch('os.getenv') @mock.patch('crmsh.sh.cluster_shell') def test_get_property(mock_run, mock_env): mock_run_inst = mock.Mock() mock_run.return_value = mock_run_inst mock_run_inst.get_rc_stdout_stderr_without_input.return_value = (0, "data", "") mock_env.return_value = "cib.xml" assert utils.get_property("no-quorum-policy") == "data" mock_run_inst.get_rc_stdout_stderr_without_input.assert_called_once_with(None, "CIB_file=cib.xml sudo --preserve-env=CIB_file crm configure get_property no-quorum-policy") @mock.patch('logging.Logger.warning') @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') @mock.patch('crmsh.utils.get_property') def test_set_property(mock_get, mock_run, mock_warn): mock_get.return_value = "start" utils.set_property("no-quorum-policy", "stop") mock_run.assert_called_once_with("crm configure property no-quorum-policy=stop") mock_warn.assert_called_once_with('"no-quorum-policy" in crm_config is set to stop, it was start') @mock.patch('crmsh.utils.get_property') def test_set_property_the_same(mock_get): mock_get.return_value = "value1" utils.set_property("no-quorum-policy", "value1") mock_get.assert_called_once_with("no-quorum-policy", "crm_config") @mock.patch('crmsh.utils.crm_msec') @mock.patch('crmsh.utils.get_property') def test_set_property_conditional(mock_get, mock_msec): mock_get.return_value = "10s" mock_msec.side_effect = ["1000", "1000"] utils.set_property("timeout", "10", conditional=True) mock_get.assert_called_once_with("timeout", "crm_config") mock_msec.assert_has_calls([mock.call("10s"), mock.call("10")]) @mock.patch('crmsh.utils.is_dlm_configured') def test_check_no_quorum_policy_with_dlm_return(mock_dlm): mock_dlm.return_value = False utils.check_no_quorum_policy_with_dlm() mock_dlm.assert_called_once_with() @mock.patch('logging.Logger.warning') @mock.patch('crmsh.utils.get_property') @mock.patch('crmsh.utils.is_dlm_configured') def test_check_no_quorum_policy_with_dlm(mock_dlm, mock_get_property, mock_warn): mock_dlm.return_value = True mock_get_property.return_value = "stop" utils.check_no_quorum_policy_with_dlm() mock_dlm.assert_called_once_with() mock_get_property.assert_called_once_with("no-quorum-policy") mock_warn.assert_called_once_with('The DLM cluster best practice suggests to set the cluster property "no-quorum-policy=freeze"') @mock.patch('crmsh.utils.is_qdevice_configured') @mock.patch('crmsh.utils.list_cluster_nodes') def test_is_2node_cluster_without_qdevice(mock_list, mock_is_qdevice): mock_list.return_value = ["node1", "node2"] mock_is_qdevice.return_value = False res = utils.is_2node_cluster_without_qdevice() assert res is True mock_list.assert_called_once_with() mock_is_qdevice.assert_called_once_with() def test_get_systemd_timeout_start_in_sec(): res = utils.get_systemd_timeout_start_in_sec("1min 31s") assert res == 91 @mock.patch('crmsh.utils.is_larger_than_min_version') @mock.patch('crmsh.cibconfig.cib_factory') def test_is_ocf_1_1_cib_schema_detected(mock_cib, mock_larger): config.core.OCF_1_1_SUPPORT = True mock_cib.get_schema = mock.Mock() mock_cib.get_schema.return_value = "pacemaker-3.5" mock_larger.return_value = True assert utils.is_ocf_1_1_cib_schema_detected() is True mock_cib.get_schema.assert_called_once_with() mock_larger.assert_called_once_with("pacemaker-3.5", constants.SCHEMA_MIN_VER_SUPPORT_OCF_1_1) @mock.patch('logging.Logger.warning') @mock.patch('crmsh.utils.is_ocf_1_1_cib_schema_detected') def test_handle_role_for_ocf_1_1(mock_support, mock_warn): mock_support.return_value = False assert utils.handle_role_for_ocf_1_1("Promoted") == "Master" mock_support.assert_called_once_with() mock_warn.assert_called_once_with('Convert "%s" to "%s" since the current schema version is old and not upgraded yet. Please consider "%s"', "Promoted", "Master", constants.CIB_UPGRADE) @mock.patch('logging.Logger.info') @mock.patch('crmsh.utils.is_ocf_1_1_cib_schema_detected') def test_handle_role_for_ocf_1_1_convert_new(mock_support, mock_info): config.core.OCF_1_1_SUPPORT = True mock_support.return_value = True assert utils.handle_role_for_ocf_1_1("Master") == "Promoted" mock_support.assert_called_once_with() mock_info.assert_called_once_with('Convert deprecated "%s" to "%s"', "Master", "Promoted") @mock.patch('crmsh.utils.is_ocf_1_1_cib_schema_detected') def test_handle_role_for_ocf_1_1_return(mock_support): mock_support.return_value = True assert utils.handle_role_for_ocf_1_1("Promoted") == "Promoted" mock_support.assert_called_once_with() def test_handle_role_for_ocf_1_1_return_not_role(): assert utils.handle_role_for_ocf_1_1("test", name='other') == "test" def test_compatible_role(): assert utils.compatible_role("Slave", "Unpromoted") is True @mock.patch('logging.Logger.warning') @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error') def test_fetch_cluster_node_list_from_node(mock_run, mock_warn): mock_run.return_value = """ 1 node1 2 node2 lost 3 node3 member """ assert utils.fetch_cluster_node_list_from_node("node1") == ["node3"] mock_run.assert_called_once_with("crm_node -l", "node1") mock_warn.assert_has_calls([ mock.call("The node '%s' has no known name and/or state information", "1"), mock.call("The node '%s'(state '%s') is not a current member", "node2", "lost") ]) @mock.patch('crmsh.utils.list_cluster_nodes_except_me') def test_cluster_copy_file_return(mock_list_nodes): mock_list_nodes.return_value = [] assert utils.cluster_copy_file("/file1") == True @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr') def test_has_sudo_access(mock_run): mock_run.return_value = (0, None, None) assert utils.has_sudo_access() is True mock_run.assert_called_once_with("sudo -S -k -n id -u") @mock.patch('grp.getgrgid') @mock.patch('os.getgroups') def test_in_haclient(mock_group, mock_getgrgid): mock_group.return_value = [90, 100] mock_getgrgid_inst1 = mock.Mock(gr_name=constants.HA_GROUP) mock_getgrgid_inst2 = mock.Mock(gr_name="other") mock_getgrgid.side_effect = [mock_getgrgid_inst1, mock_getgrgid_inst2] assert utils.in_haclient() is True mock_group.assert_called_once_with() @mock.patch('crmsh.utils.in_haclient') @mock.patch('crmsh.userdir.getuser') def test_check_user_access_root(mock_user, mock_in): mock_user.return_value = 'root' utils.check_user_access('cluster') mock_in.assert_not_called() @mock.patch('crmsh.utils.has_sudo_access') @mock.patch('crmsh.utils.in_haclient') @mock.patch('crmsh.userdir.getuser') def test_check_user_access_haclient(mock_user, mock_in, mock_sudo): mock_user.return_value = 'user' mock_in.return_value = True utils.check_user_access('ra') mock_sudo.assert_not_called() @mock.patch('logging.Logger.error') @mock.patch('crmsh.utils.has_sudo_access') @mock.patch('crmsh.utils.in_haclient') @mock.patch('crmsh.userdir.getuser') def test_check_user_access_need_sudo(mock_user, mock_in, mock_sudo, mock_error): mock_user.return_value = 'user' mock_in.return_value = False mock_sudo.return_value = True with pytest.raises(utils.TerminateSubCommand) as err: utils.check_user_access('ra') mock_error.assert_called_once_with('Please run this command starting with "sudo"') @mock.patch('logging.Logger.error') @mock.patch('crmsh.utils.has_sudo_access') @mock.patch('crmsh.utils.in_haclient') @mock.patch('crmsh.userdir.getuser') def test_check_user_access_acl(mock_user, mock_in, mock_sudo, mock_error): mock_user.return_value = 'user' mock_in.return_value = False mock_sudo.return_value = False with pytest.raises(utils.TerminateSubCommand) as err: utils.check_user_access('ra') mock_error.assert_called_once_with('This command needs higher privilege.\nOption 1) Please consider to add "user" as sudoer. For example:\n sudo bash -c \'echo "user ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/user\'\nOption 2) Add "user" to the haclient group. For example:\n sudo usermod -g haclient user') @mock.patch('logging.Logger.error') @mock.patch('crmsh.utils.has_sudo_access') @mock.patch('crmsh.utils.in_haclient') @mock.patch('crmsh.userdir.getuser') def test_check_user_access_cluster(mock_user, mock_in, mock_sudo, mock_error): mock_user.return_value = 'user' mock_in.return_value = False mock_sudo.return_value = False with pytest.raises(utils.TerminateSubCommand) as err: utils.check_user_access('cluster') mock_error.assert_called_once_with('Please run this command starting with "sudo".\nCurrently, this command needs to use sudo to escalate itself as root.\nPlease consider to add "user" as sudoer. For example:\n sudo bash -c \'echo "user ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/user\'')