diff --git a/tests/unit/host_helpers/__init__.py b/tests/unit/host_helpers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/host_helpers/test_apparmor.py b/tests/unit/host_helpers/test_apparmor.py new file mode 100644 index 000000000..6c8cfb445 --- /dev/null +++ b/tests/unit/host_helpers/test_apparmor.py @@ -0,0 +1,37 @@ + +from hotsos.core.host_helpers import apparmor as host_apparmor + +from .. import utils + + +class TestApparmorHelper(utils.BaseTestCase): + """ Unit tests for apparmor helper """ + def test_aa_status_profiles(self): + helper = host_apparmor.ApparmorHelper() + profiles = helper.profiles + num_profiles = 2 + self.assertEqual(profiles['enforce']['count'], 253) + self.assertEqual(len(profiles['enforce']['profiles']), 253) + self.assertEqual(profiles['enforce']['profiles'][-1], 'virt-aa-helper') + self.assertEqual(helper.profiles_enforce, + profiles['enforce']['profiles']) + self.assertEqual(profiles['complain']['count'], 0) + self.assertEqual(len(profiles['complain']['profiles']), 0) + self.assertEqual(helper.profiles_complain, []) + if 'kill' in profiles: + num_profiles += 1 + self.assertEqual(profiles['kill']['count'], 0) + self.assertEqual(len(profiles['kill']['profiles']), 0) + self.assertEqual(helper.profiles_kill, []) + if 'unconfined' in profiles: + num_profiles += 1 + self.assertEqual(profiles['unconfined']['count'], 0) + self.assertEqual(len(profiles['unconfined']['profiles']), 0) + self.assertEqual(helper.profiles_unconfined, []) + self.assertEqual(len(profiles), num_profiles) + + def test_aa_profile_factory(self): + profile = getattr(host_apparmor.AAProfileFactory(), + 'virt-aa-helper') + self.assertEqual(profile.name, 'virt-aa-helper') + self.assertEqual(profile.mode, 'enforce') diff --git a/tests/unit/host_helpers/test_cli.py b/tests/unit/host_helpers/test_cli.py new file mode 100644 index 000000000..9dbf4ae58 --- /dev/null +++ b/tests/unit/host_helpers/test_cli.py @@ -0,0 +1,153 @@ +import os +import subprocess +from unittest import mock + +from hotsos.core.config import HotSOSConfig +from hotsos.core.host_helpers import cli as host_cli + +from .. import utils + + +class TestCLIHelper(utils.BaseTestCase): + """ + NOTE: remember that a data_root is configured so helpers will always + use fake_data_root if possible. If you write a test that wants to + test a scenario where no data root is set (i.e. no sosreport) you need + to unset it as part of the test. + """ + + def test_journalctl(self): + HotSOSConfig.use_all_logs = False + HotSOSConfig.max_logrotate_depth = 7 + self.assertEqual(host_cli.JournalctlBase().since_date, + "2022-02-09") + HotSOSConfig.use_all_logs = True + self.assertEqual(host_cli.JournalctlBase().since_date, + "2022-02-03") + HotSOSConfig.max_logrotate_depth = 1000 + self.assertEqual(host_cli.JournalctlBase().since_date, + "2019-05-17") + + def test_ns_ip_addr(self): + ns = "qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5" + out = host_cli.CLIHelper().ns_ip_addr(namespace=ns) + self.assertIsInstance(out, list) + self.assertEqual(len(out), 18) + + def test_udevadm_info_dev(self): + out = host_cli.CLIHelper().udevadm_info_dev(device='/dev/vdb') + self.assertEqual(out, []) + + @mock.patch.object(host_cli, 'subprocess') + def test_ps(self, mock_subprocess): + path = os.path.join(HotSOSConfig.data_root, "ps") + with open(path, 'r', encoding='utf-8') as fd: + out = fd.readlines() + + self.assertEqual(host_cli.CLIHelper().ps(), out) + self.assertFalse(mock_subprocess.called) + + def test_get_date_local(self): + HotSOSConfig.data_root = '/' + self.assertEqual(type(host_cli.CLIHelper().date()), str) + + def test_get_date(self): + self.assertEqual(host_cli.CLIHelper().date(), '1644509957') + + @utils.create_data_root({'sos_commands/date/date': + 'Thu Mar 25 10:55:05 2021'}) + def test_get_date_no_tz(self): + self.assertEqual(host_cli.CLIHelper().date(), '1616669705') + + @utils.create_data_root({'sos_commands/date/date': + 'Thu Mar 25 10:55:05 -03 2021'}) + def test_get_date_w_numeric_tz(self): + self.assertEqual(host_cli.CLIHelper().date(), '1616680505') + + @utils.create_data_root({'sos_commands/date/date': + 'Thu Mar 25 10:55:05 UTC 2021'}) + def test_get_date_w_tz(self): + self.assertEqual(host_cli.CLIHelper().date(), '1616669705') + + @utils.create_data_root({'sos_commands/date/date': + 'Thu Mar 25 10:55:05 123UTC 2021'}) + def test_get_date_w_invalid_tz(self): + with self.assertLogs(logger='hotsos', level='ERROR') as log: + self.assertEqual(host_cli.CLIHelper().date(), "") + # If invalid date, log.error() will have been called + self.assertEqual(len(log.output), 1) + self.assertIn('has invalid date string', log.output[0]) + + def test_ovs_ofctl_bin_w_errors(self): + + def fake_run(cmd, *_args, **_kwargs): + if 'OpenFlow13' in cmd: + m = mock.MagicMock() + m.returncode = 0 + m.stdout = 'testdata'.encode(encoding='utf_8', errors='strict') + m.stderr = '' + return m + + raise subprocess.CalledProcessError(1, 'ofctl') + + HotSOSConfig.data_root = '/' + with mock.patch.object(host_cli.subprocess, 'run') as \ + mock_run: + mock_run.side_effect = fake_run + + # Test errors with eventual success + helper = host_cli.CLIHelper() + self.assertEqual(helper.ovs_ofctl(command='show', args='br-int'), + ['testdata']) + + mock_run.side_effect = \ + subprocess.CalledProcessError(1, 'ofctl') + + # Ensure that if all fails the result is always iterable + helper = host_cli.CLIHelper() + self.assertEqual(helper.ovs_ofctl(command='show', args='br-int'), + []) + + @mock.patch.object(host_cli.CLIHelper, 'command_catalog', + {'sleep': [host_cli.BinCmd('time sleep 2')]}) + def test_cli_timeout(self): + cli = host_cli.CLIHelper() + orig_cfg = HotSOSConfig.CONFIG + try: + # ensure bin command executed + HotSOSConfig.data_root = '/' + HotSOSConfig.command_timeout = 1 + out = cli.sleep() + # a returned [] implies an exception was raised and caught + self.assertEqual(out, []) + finally: + # restore + HotSOSConfig.set(**orig_cfg) + + @mock.patch.object(host_cli.CLIHelper, 'command_catalog', + {'sleep': [host_cli.BinCmd('time sleep 1')]}) + def test_cli_no_timeout(self): + cli = host_cli.CLIHelper() + orig_cfg = HotSOSConfig.CONFIG + try: + # ensure bin command executed + HotSOSConfig.data_root = '/' + out = cli.sleep() + self.assertEqual(len(out), 2) + finally: + # restore + HotSOSConfig.set(**orig_cfg) + + def test_clitempfile(self): + with host_cli.CLIHelperFile() as cli: + self.assertEqual(os.path.basename(cli.date()), 'date') + + with host_cli.CLIHelperFile() as cli: + orig_cfg = HotSOSConfig.CONFIG + try: + # ensure bin command executed + HotSOSConfig.data_root = '/' + self.assertEqual(cli.date(), cli.output_file) + finally: + # restore + HotSOSConfig.set(**orig_cfg) diff --git a/tests/unit/host_helpers/test_config.py b/tests/unit/host_helpers/test_config.py new file mode 100644 index 000000000..f00a75c5d --- /dev/null +++ b/tests/unit/host_helpers/test_config.py @@ -0,0 +1,47 @@ +import os + +from hotsos.core.config import HotSOSConfig +from hotsos.core.host_helpers import config as host_config + +from .. import utils + +DUMMY_CONFIG = """ +[a-section] +a-key = 1023 +b-key = 10-23 +c-key = 2-8,10-31 +""" + + +class TestConfigHelper(utils.BaseTestCase): + """ Unit tests for config helper """ + @utils.create_data_root({'test.conf': DUMMY_CONFIG}) + def test_iniconfig_base(self): + conf = os.path.join(HotSOSConfig.data_root, 'test.conf') + cfg = host_config.IniConfigBase(conf) + self.assertTrue(cfg.exists) + self.assertEqual(cfg.get('a-key'), '1023') + self.assertEqual(cfg.get('a-key', section='a-section'), '1023') + self.assertIsNone(cfg.get('a-key', section='missing-section')) + self.assertEqual(cfg.get('a-key', expand_to_list=True), [1023]) + + expanded = cfg.get('b-key', expand_to_list=True) + self.assertEqual(expanded, list(range(10, 24))) + self.assertEqual(cfg.squash_int_range(expanded), '10-23') + + expanded = cfg.get('c-key', expand_to_list=True) + self.assertEqual(expanded, list(range(2, 9)) + list(range(10, 32))) + self.assertEqual(cfg.squash_int_range(expanded), '2-8,10-31') + + @utils.create_data_root({'test.conf': DUMMY_CONFIG}) + def test_squash_int_range(self): + self.assertEqual(host_config.ConfigBase.squash_int_range([]), '') + expanded = list(range(2, 9)) + self.assertEqual(host_config.ConfigBase.squash_int_range(expanded), + '2-8') + expanded = list(range(2, 9)) + list(range(10, 32)) + self.assertEqual(host_config.ConfigBase.squash_int_range(expanded), + '2-8,10-31') + expanded = list(range(2, 9)) + [10] + list(range(12, 32)) + self.assertEqual(host_config.ConfigBase.squash_int_range(expanded), + '2-8,10,12-31') diff --git a/tests/unit/host_helpers/test_filestat.py b/tests/unit/host_helpers/test_filestat.py new file mode 100644 index 000000000..e17cd511f --- /dev/null +++ b/tests/unit/host_helpers/test_filestat.py @@ -0,0 +1,18 @@ +import os + +from hotsos.core.config import HotSOSConfig +from hotsos.core.host_helpers import filestat as host_file + +from .. import utils + + +class TestFileStatHelper(utils.BaseTestCase): + """ Unit tests for filesstat helper """ + @utils.create_data_root({'foo': 'bar'}) + def test_filestat_factory(self): + fpath = os.path.join(HotSOSConfig.data_root, 'foo') + fileobj = host_file.FileFactory().foo + self.assertEqual(fileobj.mtime, os.path.getmtime(fpath)) + + fileobj = host_file.FileFactory().noexist + self.assertEqual(fileobj.mtime, 0) diff --git a/tests/unit/host_helpers/test_network.py b/tests/unit/host_helpers/test_network.py new file mode 100644 index 000000000..46ceff6fc --- /dev/null +++ b/tests/unit/host_helpers/test_network.py @@ -0,0 +1,169 @@ +from unittest import mock + +from hotsos.core.host_helpers import cli as host_cli +from hotsos.core.host_helpers import network as host_network + +from .. import utils + + +class TestHostNetworkingHelper(utils.BaseTestCase): + """ Unit tests for networking helper. """ + def test_get_host_interfaces(self): + expected = ['lo', 'ens3', 'ens4', 'ens5', 'ens6', 'ens7', 'ens8', + 'ens9', 'br-ens3', 'ovs-system', 'br-tun', 'br-int', + 'br-ex', 'br-data', 'lxdbr0', 'veth1883dceb@if16', + 'veth5cc250bc@if18', 'veth396824c3@if20', + 'vethe7aaf6c3@if22', 'veth59e22e6f@if24', + 'veth8aa19e05@if26', 'veth0d284c32@if28', + 'vxlan_sys_4789', 'tap0e778df8-ca'] + helper = host_network.HostNetworkingHelper() + ifaces = helper.host_interfaces + names = [iface.name for iface in ifaces] + self.assertEqual(names, expected) + + def test_get_host_interfaces_w_ns(self): + expected = ['lo', 'ens3', 'ens4', 'ens5', 'ens6', 'ens7', 'ens8', + 'ens9', 'br-ens3', 'ovs-system', 'br-tun', 'br-int', + 'br-ex', 'br-data', 'lxdbr0', 'veth1883dceb@if16', + 'veth5cc250bc@if18', 'veth396824c3@if20', + 'vethe7aaf6c3@if22', 'veth59e22e6f@if24', + 'veth8aa19e05@if26', 'veth0d284c32@if28', + 'vxlan_sys_4789', 'tap0e778df8-ca', 'lo', + 'fpr-984c22fd-6@if2', 'fg-c8dcce74-c4', 'lo', + 'rfp-984c22fd-6@if2', 'qr-3a70b31c-3f', 'lo', + 'ha-550dc175-c0', 'qg-14f81a43-69', 'sg-189f4c40-9d'] + helper = host_network.HostNetworkingHelper() + ifaces = helper.host_interfaces_all + names = [iface.name for iface in ifaces] + self.assertEqual(names, expected) + + def test_get_interface_with_addr_not_exists(self): + helper = host_network.HostNetworkingHelper() + iface = helper.get_interface_with_addr('1.2.3.4') + self.assertIsNone(iface) + + def test_get_interface_with_addr_exists(self): + expected = {'br-ens3': { + 'addresses': ['10.0.0.128'], + 'hwaddr': '22:c2:7b:1c:12:1b', + 'mtu': 1500, + 'state': 'UP', + 'speed': 'unknown'}} + helper = host_network.HostNetworkingHelper() + iface = helper.get_interface_with_addr('10.0.0.128') + self.assertEqual(iface.to_dict(), expected) + + @mock.patch.object(host_network, 'CLIHelper') + def test_get_interface_with_speed_exists(self, mock_cli): + cli = host_cli.CLIHelper() + orig_ip_addr = cli.ip_addr() + orig_ip_link = cli.ip_link() + mock_cli.return_value = mock.MagicMock() + mock_cli.return_value.ethtool.return_value = ['Speed: 100000Mb/s\n'] + mock_cli.return_value.ip_addr.return_value = orig_ip_addr + mock_cli.return_value.ip_link.return_value = orig_ip_link + expected = {'br-ens3': {'addresses': ['10.0.0.128'], + 'hwaddr': '22:c2:7b:1c:12:1b', + 'mtu': 1500, + 'state': 'UP', + 'speed': '100000Mb/s'}} + helper = host_network.HostNetworkingHelper() + iface = helper.get_interface_with_addr('10.0.0.128') + self.assertEqual(iface.to_dict(), expected) + + def test_get_interface_stats(self): + expected = {'rx': {'dropped': 0, + 'errors': 0, + 'overrun': 0, + 'packets': 1628707}, + 'tx': {'dropped': 0, + 'errors': 0, + 'packets': 1520974}} + helper = host_network.HostNetworkingHelper() + iface = helper.get_interface_with_addr('10.0.0.128') + self.assertEqual(iface.stats, expected) + + def test_get_interfaces_cached(self): + helper = host_network.HostNetworkingHelper() + _ = helper.host_interfaces_all + ifaces = helper.cache.get('interfaces') # pylint: disable=E1111 + expected = ['lo', + 'ens3', + 'ens4', + 'ens5', + 'ens6', + 'ens7', + 'ens8', + 'ens9', + 'br-ens3', + 'ovs-system', + 'br-tun', + 'br-int', + 'br-ex', + 'br-data', + 'lxdbr0', + 'veth1883dceb@if16', + 'veth5cc250bc@if18', + 'veth396824c3@if20', + 'vethe7aaf6c3@if22', + 'veth59e22e6f@if24', + 'veth8aa19e05@if26', + 'veth0d284c32@if28', + 'vxlan_sys_4789', + 'tap0e778df8-ca'] + self.assertEqual([i['name'] for i in ifaces], expected) + ns_ifaces = helper.cache.get('ns-interfaces') # pylint: disable=E1111 + expected = [('lo', + 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), + ('fpr-984c22fd-6@if2', + 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), + ('fg-c8dcce74-c4', + 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), + ('lo', 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('rfp-984c22fd-6@if2', + 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('qr-3a70b31c-3f', + 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('lo', 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('ha-550dc175-c0', + 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('qg-14f81a43-69', + 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), + ('sg-189f4c40-9d', + 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5')] + self.assertEqual([(i['name'], i['namespace']) for i in ns_ifaces], + expected) + + addr = '10.0.0.128' + iface = helper.get_interface_with_addr(addr) + # do this to cache stats + _ = iface.stats + helper = host_network.HostNetworkingHelper() + data = helper.cache_load('interfaces') + iface_found = False + for _iface in data: + if _iface['name'] == iface.name: + iface_found = True + self.assertEqual(_iface['addresses'], [addr]) + + with mock.patch.object(host_network, 'CLIHelper') as mock_cli: + mock_cli.return_value = mock.MagicMock() + helper = host_network.HostNetworkingHelper() + iface = helper.get_interface_with_addr(addr) + self.assertEqual(iface.addresses, [addr]) + # these should no longer be called + self.assertFalse(mock_cli.return_value.ip_addr.called) + self.assertFalse(mock_cli.return_value.ip_netns.called) + self.assertFalse(mock_cli.return_value.ns_ip_addr.called) + self.assertFalse(mock_cli.return_value.ip_link.called) + self.assertFalse(mock_cli.return_value.ethtool.called) + + self.assertTrue(iface_found) + + def test_get_ns_interfaces(self): + expected = ['lo', 'rfp-984c22fd-6@if2', 'qr-3a70b31c-3f'] + helper = host_network.HostNetworkingHelper() + ns = 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5' + ifaces = helper.get_ns_interfaces(ns) + names = [iface.name for iface in ifaces] + self.assertEqual(names, expected) diff --git a/tests/unit/host_helpers/test_packaging.py b/tests/unit/host_helpers/test_packaging.py new file mode 100644 index 000000000..30ea429b6 --- /dev/null +++ b/tests/unit/host_helpers/test_packaging.py @@ -0,0 +1,400 @@ + +from hotsos.core.host_helpers import packaging as host_pack + +from .. import utils + + +class TestAPTPackageHelper(utils.BaseTestCase): + """ Unit tests for apt helper """ + def test_core_packages(self): + expected = {'systemd': '245.4-4ubuntu3.15', + 'systemd-container': '245.4-4ubuntu3.15', + 'systemd-sysv': '245.4-4ubuntu3.15', + 'systemd-timesyncd': '245.4-4ubuntu3.15'} + obj = host_pack.APTPackageHelper(["systemd"]) + self.assertEqual(obj.all, expected) + # lookup package already loaded + self.assertEqual(obj.get_version("systemd"), "245.4-4ubuntu3.15") + # lookup package not already loaded + self.assertEqual(obj.get_version("apt"), "2.0.6") + + def test_all_packages(self): + expected = {'python3-systemd': '234-3build2', + 'systemd': '245.4-4ubuntu3.15', + 'systemd-container': '245.4-4ubuntu3.15', + 'systemd-sysv': '245.4-4ubuntu3.15', + 'systemd-timesyncd': '245.4-4ubuntu3.15'} + obj = host_pack.APTPackageHelper(["systemd"], ["python3?-systemd"]) + self.assertEqual(obj.all, expected) + + def test_formatted(self): + expected = ['systemd 245.4-4ubuntu3.15', + 'systemd-container 245.4-4ubuntu3.15', + 'systemd-sysv 245.4-4ubuntu3.15', + 'systemd-timesyncd 245.4-4ubuntu3.15'] + obj = host_pack.APTPackageHelper(["systemd"]) + self.assertEqual(obj.all_formatted, expected) + + +class TestSnapPackageHelper(utils.BaseTestCase): + """ Unit tests for snap helper """ + def test_all(self): + expected = {'core20': {'channel': 'latest/stable', + 'version': '20220114'}} + obj = host_pack.SnapPackageHelper(["core20"]) + self.assertEqual(obj.all, expected) + # lookup package already loaded + self.assertEqual(obj.get_version("core20"), "20220114") + # lookup package not already loaded + self.assertEqual(obj.get_version("lxd"), "4.22") + + def test_formatted(self): + expected = ['core20 20220114'] + obj = host_pack.SnapPackageHelper(["core20"]) + self.assertEqual(obj.all_formatted, expected) + + +class TestDPKGVersion(utils.BaseTestCase): # noqa, pylint: disable=too-many-public-methods + """ Unit tests for dpkg version helper """ + def test_dpkg_normalize_string_repr(self): + data = [ + {"ge": "8.9"}, {"lt": "4"}, {"ge": "6.3", "lt": "7.2"} + ] + + self.assertEqual( + repr(host_pack.DPKGVersion.normalize_version_criteria(data)), + "[{'ge': 8.9}, {'ge': 6.3, 'lt': 7.2}, {'lt': 4}]") + + def test_dpkg_version_comparison(self): + data = [ + {"gt": '1.2~1'}, {"gt": '1.2~2'}, {"gt": '1.2'} + ] + + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'gt': '1.2'}, + {'gt': '1.2~2', 'le': '1.2'}, + {'gt': '1.2~1', 'le': '1.2~2'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_00(self): + for elem in ['eq', 'ge', 'gt', 'le', 'lt', 'min', 'max']: + data = [ + { + elem: '1' + } + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + self.assertEqual(result, data) + + def test_dpkg_version_normalize_01(self): + for elem_a in ['eq', 'ge', 'gt', 'le', 'lt']: + for elem_b in ['eq', 'ge', 'gt', 'le', 'lt']: + if elem_a.startswith(elem_b[0]): + continue + data = [ + { + elem_a: '3', elem_b: '4' + }, + { + elem_a: '1', elem_b: '2' + }, + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + self.assertEqual(result, data) + + def test_dpkg_version_normalize_02(self): + data = [ + {'gt': '1'}, {'gt': '2'}, {'lt': '4'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'lt': '4', 'gt': '4'}, + {'gt': '2', 'le': '4'}, + {'gt': '1', 'le': '2'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_03(self): + data = [ + {'gt': '1', 'lt': '2'}, {'gt': '3'}, {'gt': '4'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'gt': '4'}, + {'gt': '3', 'le': '4'}, + {'gt': '1', 'lt': '2'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_04(self): + data = [ + {'gt': '1', 'lt': '2'}, {'gt': '3', 'lt': '4'}, {'gt': '4'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'gt': '4'}, + {'gt': '3', 'lt': '4'}, + {'gt': '1', 'lt': '2'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_05(self): + data = [ + {'gt': '1'}, {'gt': '3', 'lt': '4'}, {'gt': '4'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'gt': '4'}, + {'gt': '3', 'lt': '4'}, + {'gt': '1', 'le': '3'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_06(self): + data = [ + {'lt': '2'}, {'lt': '3'}, {'lt': '4'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'ge': '3', 'lt': '4'}, + {'lt': '2'}, + {'ge': '2', 'lt': '3'}, + ] + self.assertEqual(result, expected) + + def test_dpkg_version_normalize_07(self): + data = [ + {'min': '2', 'max': '3'} + ] + result = host_pack.DPKGVersion.normalize_version_criteria(data) + expected = [ + {'ge': '2', 'le': '3'} + ] + self.assertEqual(result, expected) + + def test_dpkg_version_pkgver_bad_version(self): + with self.assertRaises(host_pack.DPKGBadVersionSyntax): + host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'lt': 'immabadversion'}] + ) + + def test_dpkg_version_pkgver_lt_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'lt': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_lt_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'lt': '1:8.2p1-4ubuntu0.5'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_le_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'le': '1:8.2p1-4ubuntu0.3'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_le_true_eq(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'le': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_le_true_less(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'le': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_gt_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.5'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_gt_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.3'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_ge_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'ge': '1:8.2p1-4ubuntu0.5'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_ge_true_eq(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'ge': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_ge_true_greater(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'ge': '1:8.2p1-4ubuntu0.3'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_is_equal_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'eq': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_is_equal_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'eq': '1:8.2p1-4ubuntu0.3'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_within_ranges_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.2', + 'max': '1:8.3'}]) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.2', + 'max': '1:8.2'}]) + self.assertFalse(result) + + def test_dpkg_version_pkgver_within_ranges_multi(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.0', + 'max': '1:8.1'}, + {'min': '1:8.2', + 'max': '1:8.3'}]) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_no_max_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.0'}, + {'min': '1:8.1'}, + {'min': '1:8.2'}, + {'min': '1:8.3'}]) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_no_max_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.3'}, + {'min': '1:8.4'}, + {'min': '1:8.5'}]) + self.assertFalse(result) + + def test_dpkg_version_pkgver_within_ranges_mixed_true(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.0'}, + {'min': '1:8.1', + 'max': '1:8.1.1'}, + {'min': '1:8.2'}, + {'min': '1:8.3'}]) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_mixed_false(self): + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'min': '1:8.0'}, + {'min': '1:8.1', + 'max': '1:8.1.1'}, + {'min': '1:8.2', + 'max': '1:8.2'}, + {'min': '1:8.3'}]) + self.assertFalse(result) + + def test_dpkg_version_pkgver_within_ranges_mixed_lg_false(self): + # 1:8.2p1-4ubuntu0.4 + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.1', + 'lt': '1:8.1.1'}, + {'gt': '1:8.2', + 'lt': '1:8.2'}, + {'gt': '1:8.3'}] + ) + self.assertFalse(result) + + def test_dpkg_version_pkgver_within_ranges_mixed_lg_true(self): + # 1:8.2p1-4ubuntu0.4 + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.1', + 'lt': '1:8.1.1'}, + {'gt': '1:8.2p1-4ubuntu0.3', + 'lt': '1:8.2p1-4ubuntu0.5'}, + {'gt': '1:8.3'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_first_true(self): + # 1:8.2p1-4ubuntu0.4 + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'ge': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.1', + 'lt': '1:8.1.1'}, + {'gt': '1:8.2p1-4ubuntu0.3', + 'lt': '1:8.2p1-4ubuntu0.5'}, + {'gt': '1:8.3'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_mid_true(self): + # 1:8.2p1-4ubuntu0.4 + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.1', + 'le': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.2p1-4ubuntu0.6', + 'lt': '1:8.2p1-4ubuntu0.9'}, + {'gt': '1:8.3'}] + ) + self.assertTrue(result) + + def test_dpkg_version_pkgver_within_ranges_last_true(self): + # 1:8.2p1-4ubuntu0.4 + result = host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'gt': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.1', + 'le': '1:8.2p1-4ubuntu0.4'}, + {'gt': '1:8.2p1-4ubuntu0.6', + 'lt': '1:8.2p1-4ubuntu0.9'}, + {'eq': '1:8.2p1-4ubuntu0.4'}] + ) + self.assertTrue(result) + + def test_dpkg_version_invalid_op_name(self): + with self.assertRaises(Exception): + # 1:8.2p1-4ubuntu0.4 + host_pack.DPKGVersion.is_version_within_ranges( + '1:8.2p1-4ubuntu0.4', + [{'foo': '1:8.2p1-4ubuntu0.4'}] + ) diff --git a/tests/unit/host_helpers/test_pebble.py b/tests/unit/host_helpers/test_pebble.py new file mode 100644 index 000000000..0b7e64657 --- /dev/null +++ b/tests/unit/host_helpers/test_pebble.py @@ -0,0 +1,36 @@ + +from hotsos.core.host_helpers import pebble as host_pebble + +from .. import utils + +PEBBLE_SERVICES = """Service Startup Current Since +nova-conductor enabled backoff today at 10:25 UTC +""" + +# pylint: disable=C0301 +PEBBLE_PS = """USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND +root 1 0.0 0.0 717708 10516 ? Ssl 08:43 0:01 /charm/bin/pebble run --create-dirs --hold --http :38814 --verbose +root 3048 0.0 0.0 2620 600 pts/0 Ss 10:14 0:00 sh -c bash +root 3055 0.0 0.0 7372 4036 pts/0 S 10:14 0:00 bash +root 3225 0.0 0.2 80748 65780 ? R 10:42 0:00 /usr/bin/python3 /usr/bin/nova-conductor +""" # noqa + + +class TestPebbleHelper(utils.BaseTestCase): + """ Unit tests for pebble helper """ + @utils.create_data_root({'sos_commands/pebble/pebble_services': + PEBBLE_SERVICES}) + def test_service_factory(self): + svc = getattr(host_pebble.ServiceFactory(), 'nova-conductor') + self.assertEqual(svc.state, 'backoff') + + self.assertIsNone(host_pebble.ServiceFactory().noexist) + + @utils.create_data_root({'sos_commands/pebble/pebble_services': + PEBBLE_SERVICES, + 'ps': PEBBLE_PS}) + def test_pebble_helper(self): + expected = {'ps': ['nova-conductor (1)'], + 'pebble': {'backoff': ['nova-conductor']}} + s = host_pebble.PebbleHelper([r'nova\S+']) + self.assertEqual(s.summary, expected) diff --git a/tests/unit/host_helpers/test_sysctl.py b/tests/unit/host_helpers/test_sysctl.py new file mode 100644 index 000000000..60af460e9 --- /dev/null +++ b/tests/unit/host_helpers/test_sysctl.py @@ -0,0 +1,29 @@ +import os + +from hotsos.core.config import HotSOSConfig +from hotsos.core.host_helpers import sysctl as host_sysctl + +from .. import utils + + +class TestSysctlHelper(utils.BaseTestCase): + """ Unit tests for sysctl helper """ + def test_sysctlhelper(self): + self.assertEqual(getattr(host_sysctl.SYSCtlFactory(), + 'net.core.somaxconn'), '4096') + + def test_sysctlconfhelper(self): + path = os.path.join(HotSOSConfig.data_root, 'etc/sysctl.d') + path = os.path.join(path, '50-nova-compute.conf') + sysctl = host_sysctl.SYSCtlConfHelper(path) + setters = {'net.ipv4.neigh.default.gc_thresh1': '128', + 'net.ipv4.neigh.default.gc_thresh2': '28672', + 'net.ipv4.neigh.default.gc_thresh3': '32768', + 'net.ipv6.neigh.default.gc_thresh1': '128', + 'net.ipv6.neigh.default.gc_thresh2': '28672', + 'net.ipv6.neigh.default.gc_thresh3': '32768', + 'net.nf_conntrack_max': '1000000', + 'net.netfilter.nf_conntrack_buckets': '204800', + 'net.netfilter.nf_conntrack_max': '1000000'} + self.assertEqual(sysctl.setters, setters) + self.assertEqual(sysctl.unsetters, {}) diff --git a/tests/unit/host_helpers/test_systemd.py b/tests/unit/host_helpers/test_systemd.py new file mode 100644 index 000000000..62a71d9d4 --- /dev/null +++ b/tests/unit/host_helpers/test_systemd.py @@ -0,0 +1,155 @@ + +from hotsos.core.host_helpers import systemd as host_systemd + +from .. import utils + + +class TestSystemdHelper(utils.BaseTestCase): + """ Unit tests for systemd helper """ + rsyslog_systemctl_status_template = r""" +* rsyslog.service - System Logging Service + Loaded: loaded (/lib/systemd/system/rsyslog.service; enabled;) + Active: active (running) since Wed 2022-02-09 22:38:17 {}; 17h ago + +Feb 09 22:38:17 compute4 systemd[1]: Starting System Logging Service... + +* secureboot-db.service - Secure Boot updates for DB and DBX""" + + def test_service_factory(self): + svc = host_systemd.ServiceFactory().rsyslog + self.assertEqual(svc.start_time_secs, 1644446297.0) + + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice', + 'sos_commands/systemd/systemctl_status_--all']) + def test_service_factory_no_journal(self): + svc = host_systemd.ServiceFactory().rsyslog + self.assertEqual(svc.start_time_secs, 1644446297.0) + + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("+08")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_plus8(self): + svc = host_systemd.ServiceFactory().rsyslog + # 2022-02-09 22:38:17 UTC+08 is 2022-02-09 14:38:17 UTC + self.assertEqual(svc.start_time_secs, 1644417497.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("+0530")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_plus0530(self): + svc = host_systemd.ServiceFactory().rsyslog + # 2022-02-09 22:38:17 UTC+0530 is 2022-02-09 17:08:17 UTC + self.assertEqual(svc.start_time_secs, 1644426497.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("-0730")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_minus0730(self): + svc = host_systemd.ServiceFactory().rsyslog + # 2022-02-09 22:38:17 UTC-0730 is 2022-02-09 17:08:17 UTC + self.assertEqual(svc.start_time_secs, 1644473297.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("HKT")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_hkt(self): + # Hong Kong Time (UTC+08) + # 2022-02-09 22:38:17 HKT is 2022-02-09 14:38:17 UTC + svc = host_systemd.ServiceFactory().rsyslog + self.assertEqual(svc.start_time_secs, 1644417497.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("HST")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_hst(self): + # Hawaii-Aleutian Standard Time (UTC-10) + # 2022-02-09 22:38:17 UTC-10 is 2022-02-10 08:38:17 UTC + svc = host_systemd.ServiceFactory().rsyslog + self.assertEqual(svc.start_time_secs, 1644482297.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + rsyslog_systemctl_status_template.format("ACST")}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_service_factory_no_journal_non_utc_acst(self): + # Australian Central Standard Time (UTC+09:30) + # 2022-02-09 22:38:17 ACST is 2022-02-09 13:08:17 UTC + svc = host_systemd.ServiceFactory().rsyslog + self.assertEqual(svc.start_time_secs, 1644412097.0) + self.assertIsNone(host_systemd.ServiceFactory().noexist) + + def test_systemd_helper(self): + expected = {'ps': ['nova-api-metadata (5)', 'nova-compute (1)'], + 'systemd': {'enabled': + ['nova-api-metadata', 'nova-compute']}} + s = host_systemd.SystemdHelper([r'nova\S+']) + self.assertEqual(s.summary, expected) + + @utils.create_data_root( + {}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files', + 'sys/fs/cgroup/memory/system.slice']) + def test_systemd_service_focal(self): + s = host_systemd.SystemdHelper([r'nova\S+']) + svc = s.services['nova-compute'] + self.assertEqual(svc.memory_current_kb, int(1517744128 / 1024)) + + @utils.create_data_root( + {'sys/fs/cgroup/system.slice/nova-compute.service/memory.current': + '7168'}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files']) + def test_systemd_service_jammy(self): + s = host_systemd.SystemdHelper([r'nova\S+']) + svc = s.services['nova-compute'] + self.assertEqual(svc.memory_current_kb, 7) + + @utils.create_data_root( + {'sos_commands/systemd/systemctl_status_--all': + """ + ● neutron-ovs-cleanup.service + Loaded: masked (Reason: Unit neutron-ovs-cleanup.service is masked.) + Active: inactive (dead) + """}, + copy_from_original=['sos_commands/systemd/systemctl_list-units', + 'sos_commands/systemd/systemctl_list-unit-files']) + def test_start_time_svc_not_active(self): + with self.assertLogs(logger='hotsos', level='WARNING') as log: + svc = getattr(host_systemd.ServiceFactory(), + 'neutron-ovs-cleanup') + self.assertEqual(svc.start_time_secs, 0) + # If not active, log.warning() will have been called + self.assertEqual(len(log.output), 1) + self.assertIn('no active status found for neutron-ovs-cleanup', + log.output[0]) diff --git a/tests/unit/host_helpers/test_uptime.py b/tests/unit/host_helpers/test_uptime.py new file mode 100644 index 000000000..82646a893 --- /dev/null +++ b/tests/unit/host_helpers/test_uptime.py @@ -0,0 +1,37 @@ + +from hotsos.core.host_helpers import uptime as host_uptime + +from .. import utils + + +class TestUptimeHelper(utils.BaseTestCase): + """ Unit tests for uptime helper """ + def test_loadavg(self): + self.assertEqual(host_uptime.UptimeHelper().loadavg, + "3.58, 3.27, 2.58") + + def test_uptime(self): + uptime = host_uptime.UptimeHelper() + self.assertEqual(uptime.in_seconds, 63660) + self.assertEqual(uptime.in_hours, 17) + self.assertEqual(repr(uptime), '0d:17h:41m') + + @utils.create_data_root({'uptime': + (' 14:51:10 up 1 day, 6:27, 1 user, ' + 'load average: 0.55, 0.73, 0.70')}) + def test_uptime_alt_format(self): + uptime = host_uptime.UptimeHelper() + self.assertEqual(uptime.in_seconds, 109620) + self.assertEqual(uptime.in_hours, 30) + self.assertEqual(repr(uptime), '1d:6h:27m') + + @utils.create_data_root({'uptime': + (' 19:12:40 up 1:55, 2 users, ' + 'load average: 3.92, 4.05, 3.90')}) + def test_uptime_alt_format2(self): + uptime = host_uptime.UptimeHelper() + self.assertEqual(uptime.in_seconds, 6900) + self.assertEqual(uptime.in_hours, 1) + self.assertEqual(uptime.loadavg, + '3.92, 4.05, 3.90') + self.assertEqual(repr(uptime), '0d:1h:55m') diff --git a/tests/unit/test_host_helpers.py b/tests/unit/test_host_helpers.py deleted file mode 100644 index 54fcbbc15..000000000 --- a/tests/unit/test_host_helpers.py +++ /dev/null @@ -1,1032 +0,0 @@ -import os -import subprocess -from unittest import mock - -from hotsos.core.config import HotSOSConfig -from hotsos.core import host_helpers -from hotsos.core.host_helpers.packaging import ( - DPKGVersion, - DPKGBadVersionSyntax -) -from hotsos.core.host_helpers.config import ConfigBase -from hotsos.core.host_helpers.filestat import FileFactory - -from . import utils - -DUMMY_CONFIG = """ -[a-section] -a-key = 1023 -b-key = 10-23 -c-key = 2-8,10-31 -""" - -PEBBLE_SERVICES = """Service Startup Current Since -nova-conductor enabled backoff today at 10:25 UTC -""" - -# pylint: disable=C0301 -PEBBLE_PS = """USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND -root 1 0.0 0.0 717708 10516 ? Ssl 08:43 0:01 /charm/bin/pebble run --create-dirs --hold --http :38814 --verbose -root 3048 0.0 0.0 2620 600 pts/0 Ss 10:14 0:00 sh -c bash -root 3055 0.0 0.0 7372 4036 pts/0 S 10:14 0:00 bash -root 3225 0.0 0.2 80748 65780 ? R 10:42 0:00 /usr/bin/python3 /usr/bin/nova-conductor -""" # noqa - - -class TestHostNetworkingHelper(utils.BaseTestCase): - """ Unit tests for networking helper. """ - def test_get_host_interfaces(self): - expected = ['lo', 'ens3', 'ens4', 'ens5', 'ens6', 'ens7', 'ens8', - 'ens9', 'br-ens3', 'ovs-system', 'br-tun', 'br-int', - 'br-ex', 'br-data', 'lxdbr0', 'veth1883dceb@if16', - 'veth5cc250bc@if18', 'veth396824c3@if20', - 'vethe7aaf6c3@if22', 'veth59e22e6f@if24', - 'veth8aa19e05@if26', 'veth0d284c32@if28', - 'vxlan_sys_4789', 'tap0e778df8-ca'] - helper = host_helpers.HostNetworkingHelper() - ifaces = helper.host_interfaces - names = [iface.name for iface in ifaces] - self.assertEqual(names, expected) - - def test_get_host_interfaces_w_ns(self): - expected = ['lo', 'ens3', 'ens4', 'ens5', 'ens6', 'ens7', 'ens8', - 'ens9', 'br-ens3', 'ovs-system', 'br-tun', 'br-int', - 'br-ex', 'br-data', 'lxdbr0', 'veth1883dceb@if16', - 'veth5cc250bc@if18', 'veth396824c3@if20', - 'vethe7aaf6c3@if22', 'veth59e22e6f@if24', - 'veth8aa19e05@if26', 'veth0d284c32@if28', - 'vxlan_sys_4789', 'tap0e778df8-ca', 'lo', - 'fpr-984c22fd-6@if2', 'fg-c8dcce74-c4', 'lo', - 'rfp-984c22fd-6@if2', 'qr-3a70b31c-3f', 'lo', - 'ha-550dc175-c0', 'qg-14f81a43-69', 'sg-189f4c40-9d'] - helper = host_helpers.HostNetworkingHelper() - ifaces = helper.host_interfaces_all - names = [iface.name for iface in ifaces] - self.assertEqual(names, expected) - - def test_get_interface_with_addr_not_exists(self): - helper = host_helpers.HostNetworkingHelper() - iface = helper.get_interface_with_addr('1.2.3.4') - self.assertIsNone(iface) - - def test_get_interface_with_addr_exists(self): - expected = {'br-ens3': { - 'addresses': ['10.0.0.128'], - 'hwaddr': '22:c2:7b:1c:12:1b', - 'mtu': 1500, - 'state': 'UP', - 'speed': 'unknown'}} - helper = host_helpers.HostNetworkingHelper() - iface = helper.get_interface_with_addr('10.0.0.128') - self.assertEqual(iface.to_dict(), expected) - - @mock.patch.object(host_helpers.network, 'CLIHelper') - def test_get_interface_with_speed_exists(self, mock_cli): - cli = host_helpers.CLIHelper() - orig_ip_addr = cli.ip_addr() - orig_ip_link = cli.ip_link() - mock_cli.return_value = mock.MagicMock() - mock_cli.return_value.ethtool.return_value = ['Speed: 100000Mb/s\n'] - mock_cli.return_value.ip_addr.return_value = orig_ip_addr - mock_cli.return_value.ip_link.return_value = orig_ip_link - expected = {'br-ens3': {'addresses': ['10.0.0.128'], - 'hwaddr': '22:c2:7b:1c:12:1b', - 'mtu': 1500, - 'state': 'UP', - 'speed': '100000Mb/s'}} - helper = host_helpers.HostNetworkingHelper() - iface = helper.get_interface_with_addr('10.0.0.128') - self.assertEqual(iface.to_dict(), expected) - - def test_get_interface_stats(self): - expected = {'rx': {'dropped': 0, - 'errors': 0, - 'overrun': 0, - 'packets': 1628707}, - 'tx': {'dropped': 0, - 'errors': 0, - 'packets': 1520974}} - helper = host_helpers.HostNetworkingHelper() - iface = helper.get_interface_with_addr('10.0.0.128') - self.assertEqual(iface.stats, expected) - - def test_get_interfaces_cached(self): - helper = host_helpers.HostNetworkingHelper() - _ = helper.host_interfaces_all - ifaces = helper.cache.get('interfaces') # pylint: disable=E1111 - expected = ['lo', - 'ens3', - 'ens4', - 'ens5', - 'ens6', - 'ens7', - 'ens8', - 'ens9', - 'br-ens3', - 'ovs-system', - 'br-tun', - 'br-int', - 'br-ex', - 'br-data', - 'lxdbr0', - 'veth1883dceb@if16', - 'veth5cc250bc@if18', - 'veth396824c3@if20', - 'vethe7aaf6c3@if22', - 'veth59e22e6f@if24', - 'veth8aa19e05@if26', - 'veth0d284c32@if28', - 'vxlan_sys_4789', - 'tap0e778df8-ca'] - self.assertEqual([i['name'] for i in ifaces], expected) - ns_ifaces = helper.cache.get('ns-interfaces') # pylint: disable=E1111 - expected = [('lo', - 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), - ('fpr-984c22fd-6@if2', - 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), - ('fg-c8dcce74-c4', - 'fip-32981f34-497a-4fae-914a-8576055c8d0d'), - ('lo', 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('rfp-984c22fd-6@if2', - 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('qr-3a70b31c-3f', - 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('lo', 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('ha-550dc175-c0', - 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('qg-14f81a43-69', - 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5'), - ('sg-189f4c40-9d', - 'snat-984c22fd-64b3-4fa1-8ddd-87090f401ce5')] - self.assertEqual([(i['name'], i['namespace']) for i in ns_ifaces], - expected) - - addr = '10.0.0.128' - iface = helper.get_interface_with_addr(addr) - # do this to cache stats - _ = iface.stats - helper = host_helpers.HostNetworkingHelper() - data = helper.cache_load('interfaces') - iface_found = False - for _iface in data: - if _iface['name'] == iface.name: - iface_found = True - self.assertEqual(_iface['addresses'], [addr]) - - with mock.patch.object(host_helpers.network, 'CLIHelper') as mock_cli: - mock_cli.return_value = mock.MagicMock() - helper = host_helpers.HostNetworkingHelper() - iface = helper.get_interface_with_addr(addr) - self.assertEqual(iface.addresses, [addr]) - # these should no longer be called - self.assertFalse(mock_cli.return_value.ip_addr.called) - self.assertFalse(mock_cli.return_value.ip_netns.called) - self.assertFalse(mock_cli.return_value.ns_ip_addr.called) - self.assertFalse(mock_cli.return_value.ip_link.called) - self.assertFalse(mock_cli.return_value.ethtool.called) - - self.assertTrue(iface_found) - - def test_get_ns_interfaces(self): - expected = ['lo', 'rfp-984c22fd-6@if2', 'qr-3a70b31c-3f'] - helper = host_helpers.HostNetworkingHelper() - ns = 'qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5' - ifaces = helper.get_ns_interfaces(ns) - names = [iface.name for iface in ifaces] - self.assertEqual(names, expected) - - -class TestCLIHelper(utils.BaseTestCase): - """ - NOTE: remember that a data_root is configured so helpers will always - use fake_data_root if possible. If you write a test that wants to - test a scenario where no data root is set (i.e. no sosreport) you need - to unset it as part of the test. - """ - - def test_journalctl(self): - HotSOSConfig.use_all_logs = False - HotSOSConfig.max_logrotate_depth = 7 - self.assertEqual(host_helpers.cli.JournalctlBase().since_date, - "2022-02-09") - HotSOSConfig.use_all_logs = True - self.assertEqual(host_helpers.cli.JournalctlBase().since_date, - "2022-02-03") - HotSOSConfig.max_logrotate_depth = 1000 - self.assertEqual(host_helpers.cli.JournalctlBase().since_date, - "2019-05-17") - - def test_ns_ip_addr(self): - ns = "qrouter-984c22fd-64b3-4fa1-8ddd-87090f401ce5" - out = host_helpers.cli.CLIHelper().ns_ip_addr(namespace=ns) - self.assertIsInstance(out, list) - self.assertEqual(len(out), 18) - - def test_udevadm_info_dev(self): - out = host_helpers.cli.CLIHelper().udevadm_info_dev(device='/dev/vdb') - self.assertEqual(out, []) - - @mock.patch.object(host_helpers.cli, 'subprocess') - def test_ps(self, mock_subprocess): - path = os.path.join(HotSOSConfig.data_root, "ps") - with open(path, 'r', encoding='utf-8') as fd: - out = fd.readlines() - - self.assertEqual(host_helpers.cli.CLIHelper().ps(), out) - self.assertFalse(mock_subprocess.called) - - def test_get_date_local(self): - HotSOSConfig.data_root = '/' - self.assertEqual(type(host_helpers.cli.CLIHelper().date()), str) - - def test_get_date(self): - self.assertEqual(host_helpers.cli.CLIHelper().date(), '1644509957') - - @utils.create_data_root({'sos_commands/date/date': - 'Thu Mar 25 10:55:05 2021'}) - def test_get_date_no_tz(self): - self.assertEqual(host_helpers.cli.CLIHelper().date(), '1616669705') - - @utils.create_data_root({'sos_commands/date/date': - 'Thu Mar 25 10:55:05 -03 2021'}) - def test_get_date_w_numeric_tz(self): - self.assertEqual(host_helpers.cli.CLIHelper().date(), '1616680505') - - @utils.create_data_root({'sos_commands/date/date': - 'Thu Mar 25 10:55:05 UTC 2021'}) - def test_get_date_w_tz(self): - self.assertEqual(host_helpers.cli.CLIHelper().date(), '1616669705') - - @utils.create_data_root({'sos_commands/date/date': - 'Thu Mar 25 10:55:05 123UTC 2021'}) - def test_get_date_w_invalid_tz(self): - with self.assertLogs(logger='hotsos', level='ERROR') as log: - self.assertEqual(host_helpers.cli.CLIHelper().date(), "") - # If invalid date, log.error() will have been called - self.assertEqual(len(log.output), 1) - self.assertIn('has invalid date string', log.output[0]) - - def test_ovs_ofctl_bin_w_errors(self): - - def fake_run(cmd, *_args, **_kwargs): - if 'OpenFlow13' in cmd: - m = mock.MagicMock() - m.returncode = 0 - m.stdout = 'testdata'.encode(encoding='utf_8', errors='strict') - m.stderr = '' - return m - - raise subprocess.CalledProcessError(1, 'ofctl') - - HotSOSConfig.data_root = '/' - with mock.patch.object(host_helpers.cli.subprocess, 'run') as \ - mock_run: - mock_run.side_effect = fake_run - - # Test errors with eventual success - helper = host_helpers.cli.CLIHelper() - self.assertEqual(helper.ovs_ofctl(command='show', args='br-int'), - ['testdata']) - - mock_run.side_effect = \ - subprocess.CalledProcessError(1, 'ofctl') - - # Ensure that if all fails the result is always iterable - helper = host_helpers.cli.CLIHelper() - self.assertEqual(helper.ovs_ofctl(command='show', args='br-int'), - []) - - @mock.patch.object(host_helpers.cli.CLIHelper, 'command_catalog', - {'sleep': [host_helpers.cli.BinCmd('time sleep 2')]}) - def test_cli_timeout(self): - cli = host_helpers.cli.CLIHelper() - orig_cfg = HotSOSConfig.CONFIG - try: - # ensure bin command executed - HotSOSConfig.data_root = '/' - HotSOSConfig.command_timeout = 1 - out = cli.sleep() - # a returned [] implies an exception was raised and caught - self.assertEqual(out, []) - finally: - # restore - HotSOSConfig.set(**orig_cfg) - - @mock.patch.object(host_helpers.cli.CLIHelper, 'command_catalog', - {'sleep': [host_helpers.cli.BinCmd('time sleep 1')]}) - def test_cli_no_timeout(self): - cli = host_helpers.cli.CLIHelper() - orig_cfg = HotSOSConfig.CONFIG - try: - # ensure bin command executed - HotSOSConfig.data_root = '/' - out = cli.sleep() - self.assertEqual(len(out), 2) - finally: - # restore - HotSOSConfig.set(**orig_cfg) - - def test_clitempfile(self): - with host_helpers.cli.CLIHelperFile() as cli: - self.assertEqual(os.path.basename(cli.date()), 'date') - - with host_helpers.cli.CLIHelperFile() as cli: - orig_cfg = HotSOSConfig.CONFIG - try: - # ensure bin command executed - HotSOSConfig.data_root = '/' - self.assertEqual(cli.date(), cli.output_file) - finally: - # restore - HotSOSConfig.set(**orig_cfg) - - -class TestSystemdHelper(utils.BaseTestCase): - """ Unit tests for systemd helper """ - rsyslog_systemctl_status_template = r""" -* rsyslog.service - System Logging Service - Loaded: loaded (/lib/systemd/system/rsyslog.service; enabled;) - Active: active (running) since Wed 2022-02-09 22:38:17 {}; 17h ago - -Feb 09 22:38:17 compute4 systemd[1]: Starting System Logging Service... - -* secureboot-db.service - Secure Boot updates for DB and DBX""" - - def test_service_factory(self): - svc = host_helpers.systemd.ServiceFactory().rsyslog - self.assertEqual(svc.start_time_secs, 1644446297.0) - - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice', - 'sos_commands/systemd/systemctl_status_--all']) - def test_service_factory_no_journal(self): - svc = host_helpers.systemd.ServiceFactory().rsyslog - self.assertEqual(svc.start_time_secs, 1644446297.0) - - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("+08")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_plus8(self): - svc = host_helpers.systemd.ServiceFactory().rsyslog - # 2022-02-09 22:38:17 UTC+08 is 2022-02-09 14:38:17 UTC - self.assertEqual(svc.start_time_secs, 1644417497.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("+0530")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_plus0530(self): - svc = host_helpers.systemd.ServiceFactory().rsyslog - # 2022-02-09 22:38:17 UTC+0530 is 2022-02-09 17:08:17 UTC - self.assertEqual(svc.start_time_secs, 1644426497.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("-0730")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_minus0730(self): - svc = host_helpers.systemd.ServiceFactory().rsyslog - # 2022-02-09 22:38:17 UTC-0730 is 2022-02-09 17:08:17 UTC - self.assertEqual(svc.start_time_secs, 1644473297.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("HKT")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_hkt(self): - # Hong Kong Time (UTC+08) - # 2022-02-09 22:38:17 HKT is 2022-02-09 14:38:17 UTC - svc = host_helpers.systemd.ServiceFactory().rsyslog - self.assertEqual(svc.start_time_secs, 1644417497.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("HST")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_hst(self): - # Hawaii-Aleutian Standard Time (UTC-10) - # 2022-02-09 22:38:17 UTC-10 is 2022-02-10 08:38:17 UTC - svc = host_helpers.systemd.ServiceFactory().rsyslog - self.assertEqual(svc.start_time_secs, 1644482297.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - rsyslog_systemctl_status_template.format("ACST")}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_service_factory_no_journal_non_utc_acst(self): - # Australian Central Standard Time (UTC+09:30) - # 2022-02-09 22:38:17 ACST is 2022-02-09 13:08:17 UTC - svc = host_helpers.systemd.ServiceFactory().rsyslog - self.assertEqual(svc.start_time_secs, 1644412097.0) - self.assertIsNone(host_helpers.systemd.ServiceFactory().noexist) - - def test_systemd_helper(self): - expected = {'ps': ['nova-api-metadata (5)', 'nova-compute (1)'], - 'systemd': {'enabled': - ['nova-api-metadata', 'nova-compute']}} - s = host_helpers.systemd.SystemdHelper([r'nova\S+']) - self.assertEqual(s.summary, expected) - - @utils.create_data_root( - {}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files', - 'sys/fs/cgroup/memory/system.slice']) - def test_systemd_service_focal(self): - s = host_helpers.systemd.SystemdHelper([r'nova\S+']) - svc = s.services['nova-compute'] - self.assertEqual(svc.memory_current_kb, int(1517744128 / 1024)) - - @utils.create_data_root( - {'sys/fs/cgroup/system.slice/nova-compute.service/memory.current': - '7168'}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files']) - def test_systemd_service_jammy(self): - s = host_helpers.systemd.SystemdHelper([r'nova\S+']) - svc = s.services['nova-compute'] - self.assertEqual(svc.memory_current_kb, 7) - - @utils.create_data_root( - {'sos_commands/systemd/systemctl_status_--all': - """ - ● neutron-ovs-cleanup.service - Loaded: masked (Reason: Unit neutron-ovs-cleanup.service is masked.) - Active: inactive (dead) - """}, - copy_from_original=['sos_commands/systemd/systemctl_list-units', - 'sos_commands/systemd/systemctl_list-unit-files']) - def test_start_time_svc_not_active(self): - with self.assertLogs(logger='hotsos', level='WARNING') as log: - svc = getattr(host_helpers.systemd.ServiceFactory(), - 'neutron-ovs-cleanup') - self.assertEqual(svc.start_time_secs, 0) - # If not active, log.warning() will have been called - self.assertEqual(len(log.output), 1) - self.assertIn('no active status found for neutron-ovs-cleanup', - log.output[0]) - - -class TestPebbleHelper(utils.BaseTestCase): - """ Unit tests for pebble helper """ - @utils.create_data_root({'sos_commands/pebble/pebble_services': - PEBBLE_SERVICES}) - def test_service_factory(self): - svc = getattr(host_helpers.pebble.ServiceFactory(), 'nova-conductor') - self.assertEqual(svc.state, 'backoff') - - self.assertIsNone(host_helpers.pebble.ServiceFactory().noexist) - - @utils.create_data_root({'sos_commands/pebble/pebble_services': - PEBBLE_SERVICES, - 'ps': PEBBLE_PS}) - def test_pebble_helper(self): - expected = {'ps': ['nova-conductor (1)'], - 'pebble': {'backoff': ['nova-conductor']}} - s = host_helpers.pebble.PebbleHelper([r'nova\S+']) - self.assertEqual(s.summary, expected) - - -class TestFileStatHelper(utils.BaseTestCase): - """ Unit tests for filesstat helper """ - @utils.create_data_root({'foo': 'bar'}) - def test_filestat_factory(self): - fpath = os.path.join(HotSOSConfig.data_root, 'foo') - fileobj = FileFactory().foo - self.assertEqual(fileobj.mtime, os.path.getmtime(fpath)) - - fileobj = FileFactory().noexist - self.assertEqual(fileobj.mtime, 0) - - -class TestUptimeHelper(utils.BaseTestCase): - """ Unit tests for uptime helper """ - def test_loadavg(self): - self.assertEqual(host_helpers.UptimeHelper().loadavg, - "3.58, 3.27, 2.58") - - def test_uptime(self): - uptime = host_helpers.UptimeHelper() - self.assertEqual(uptime.in_seconds, 63660) - self.assertEqual(uptime.in_hours, 17) - self.assertEqual(repr(uptime), '0d:17h:41m') - - @utils.create_data_root({'uptime': - (' 14:51:10 up 1 day, 6:27, 1 user, ' - 'load average: 0.55, 0.73, 0.70')}) - def test_uptime_alt_format(self): - uptime = host_helpers.UptimeHelper() - self.assertEqual(uptime.in_seconds, 109620) - self.assertEqual(uptime.in_hours, 30) - self.assertEqual(repr(uptime), '1d:6h:27m') - - @utils.create_data_root({'uptime': - (' 19:12:40 up 1:55, 2 users, ' - 'load average: 3.92, 4.05, 3.90')}) - def test_uptime_alt_format2(self): - uptime = host_helpers.UptimeHelper() - self.assertEqual(uptime.in_seconds, 6900) - self.assertEqual(uptime.in_hours, 1) - self.assertEqual(uptime.loadavg, - '3.92, 4.05, 3.90') - self.assertEqual(repr(uptime), '0d:1h:55m') - - -class TestSysctlHelper(utils.BaseTestCase): - """ Unit tests for sysctl helper """ - def test_sysctlhelper(self): - self.assertEqual(getattr(host_helpers.SYSCtlFactory(), - 'net.core.somaxconn'), '4096') - - def test_sysctlconfhelper(self): - path = os.path.join(HotSOSConfig.data_root, 'etc/sysctl.d') - path = os.path.join(path, '50-nova-compute.conf') - sysctl = host_helpers.SYSCtlConfHelper(path) - setters = {'net.ipv4.neigh.default.gc_thresh1': '128', - 'net.ipv4.neigh.default.gc_thresh2': '28672', - 'net.ipv4.neigh.default.gc_thresh3': '32768', - 'net.ipv6.neigh.default.gc_thresh1': '128', - 'net.ipv6.neigh.default.gc_thresh2': '28672', - 'net.ipv6.neigh.default.gc_thresh3': '32768', - 'net.nf_conntrack_max': '1000000', - 'net.netfilter.nf_conntrack_buckets': '204800', - 'net.netfilter.nf_conntrack_max': '1000000'} - self.assertEqual(sysctl.setters, setters) - self.assertEqual(sysctl.unsetters, {}) - - -class TestAPTPackageHelper(utils.BaseTestCase): - """ Unit tests for apt helper """ - def test_core_packages(self): - expected = {'systemd': '245.4-4ubuntu3.15', - 'systemd-container': '245.4-4ubuntu3.15', - 'systemd-sysv': '245.4-4ubuntu3.15', - 'systemd-timesyncd': '245.4-4ubuntu3.15'} - obj = host_helpers.APTPackageHelper(["systemd"]) - self.assertEqual(obj.all, expected) - # lookup package already loaded - self.assertEqual(obj.get_version("systemd"), "245.4-4ubuntu3.15") - # lookup package not already loaded - self.assertEqual(obj.get_version("apt"), "2.0.6") - - def test_all_packages(self): - expected = {'python3-systemd': '234-3build2', - 'systemd': '245.4-4ubuntu3.15', - 'systemd-container': '245.4-4ubuntu3.15', - 'systemd-sysv': '245.4-4ubuntu3.15', - 'systemd-timesyncd': '245.4-4ubuntu3.15'} - obj = host_helpers.APTPackageHelper(["systemd"], ["python3?-systemd"]) - self.assertEqual(obj.all, expected) - - def test_formatted(self): - expected = ['systemd 245.4-4ubuntu3.15', - 'systemd-container 245.4-4ubuntu3.15', - 'systemd-sysv 245.4-4ubuntu3.15', - 'systemd-timesyncd 245.4-4ubuntu3.15'] - obj = host_helpers.APTPackageHelper(["systemd"]) - self.assertEqual(obj.all_formatted, expected) - - -class TestSnapPackageHelper(utils.BaseTestCase): - """ Unit tests for snap helper """ - def test_all(self): - expected = {'core20': {'channel': 'latest/stable', - 'version': '20220114'}} - obj = host_helpers.SnapPackageHelper(["core20"]) - self.assertEqual(obj.all, expected) - # lookup package already loaded - self.assertEqual(obj.get_version("core20"), "20220114") - # lookup package not already loaded - self.assertEqual(obj.get_version("lxd"), "4.22") - - def test_formatted(self): - expected = ['core20 20220114'] - obj = host_helpers.SnapPackageHelper(["core20"]) - self.assertEqual(obj.all_formatted, expected) - - -class TestConfigHelper(utils.BaseTestCase): - """ Unit tests for config helper """ - @utils.create_data_root({'test.conf': DUMMY_CONFIG}) - def test_iniconfig_base(self): - conf = os.path.join(HotSOSConfig.data_root, 'test.conf') - cfg = host_helpers.IniConfigBase(conf) - self.assertTrue(cfg.exists) - self.assertEqual(cfg.get('a-key'), '1023') - self.assertEqual(cfg.get('a-key', section='a-section'), '1023') - self.assertIsNone(cfg.get('a-key', section='missing-section')) - self.assertEqual(cfg.get('a-key', expand_to_list=True), [1023]) - - expanded = cfg.get('b-key', expand_to_list=True) - self.assertEqual(expanded, list(range(10, 24))) - self.assertEqual(cfg.squash_int_range(expanded), '10-23') - - expanded = cfg.get('c-key', expand_to_list=True) - self.assertEqual(expanded, list(range(2, 9)) + list(range(10, 32))) - self.assertEqual(cfg.squash_int_range(expanded), '2-8,10-31') - - @utils.create_data_root({'test.conf': DUMMY_CONFIG}) - def test_squash_int_range(self): - self.assertEqual(ConfigBase.squash_int_range([]), '') - expanded = list(range(2, 9)) - self.assertEqual(ConfigBase.squash_int_range(expanded), '2-8') - expanded = list(range(2, 9)) + list(range(10, 32)) - self.assertEqual(ConfigBase.squash_int_range(expanded), '2-8,10-31') - expanded = list(range(2, 9)) + [10] + list(range(12, 32)) - self.assertEqual(ConfigBase.squash_int_range(expanded), '2-8,10,12-31') - - -class TestApparmorHelper(utils.BaseTestCase): - """ Unit tests for apparmor helper """ - def test_aa_status_profiles(self): - helper = host_helpers.ApparmorHelper() - profiles = helper.profiles - num_profiles = 2 - self.assertEqual(profiles['enforce']['count'], 253) - self.assertEqual(len(profiles['enforce']['profiles']), 253) - self.assertEqual(profiles['enforce']['profiles'][-1], 'virt-aa-helper') - self.assertEqual(helper.profiles_enforce, - profiles['enforce']['profiles']) - self.assertEqual(profiles['complain']['count'], 0) - self.assertEqual(len(profiles['complain']['profiles']), 0) - self.assertEqual(helper.profiles_complain, []) - if 'kill' in profiles: - num_profiles += 1 - self.assertEqual(profiles['kill']['count'], 0) - self.assertEqual(len(profiles['kill']['profiles']), 0) - self.assertEqual(helper.profiles_kill, []) - if 'unconfined' in profiles: - num_profiles += 1 - self.assertEqual(profiles['unconfined']['count'], 0) - self.assertEqual(len(profiles['unconfined']['profiles']), 0) - self.assertEqual(helper.profiles_unconfined, []) - self.assertEqual(len(profiles), num_profiles) - - def test_aa_profile_factory(self): - profile = getattr(host_helpers.apparmor.AAProfileFactory(), - 'virt-aa-helper') - self.assertEqual(profile.name, 'virt-aa-helper') - self.assertEqual(profile.mode, 'enforce') - - -class TestDPKGVersion(utils.BaseTestCase): # noqa, pylint: disable=too-many-public-methods - """ Unit tests for dpkg version helper """ - def test_dpkg_normalize_string_repr(self): - data = [ - {"ge": "8.9"}, {"lt": "4"}, {"ge": "6.3", "lt": "7.2"} - ] - - self.assertEqual(repr(DPKGVersion.normalize_version_criteria(data)), - "[{'ge': 8.9}, {'ge': 6.3, 'lt': 7.2}, {'lt': 4}]") - - def test_dpkg_version_comparison(self): - data = [ - {"gt": '1.2~1'}, {"gt": '1.2~2'}, {"gt": '1.2'} - ] - - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'gt': '1.2'}, - {'gt': '1.2~2', 'le': '1.2'}, - {'gt': '1.2~1', 'le': '1.2~2'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_00(self): - for elem in ['eq', 'ge', 'gt', 'le', 'lt', 'min', 'max']: - data = [ - { - elem: '1' - } - ] - result = DPKGVersion.normalize_version_criteria(data) - self.assertEqual(result, data) - - def test_dpkg_version_normalize_01(self): - for elem_a in ['eq', 'ge', 'gt', 'le', 'lt']: - for elem_b in ['eq', 'ge', 'gt', 'le', 'lt']: - if elem_a.startswith(elem_b[0]): - continue - data = [ - { - elem_a: '3', elem_b: '4' - }, - { - elem_a: '1', elem_b: '2' - }, - ] - result = DPKGVersion.normalize_version_criteria(data) - self.assertEqual(result, data) - - def test_dpkg_version_normalize_02(self): - data = [ - {'gt': '1'}, {'gt': '2'}, {'lt': '4'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'lt': '4', 'gt': '4'}, - {'gt': '2', 'le': '4'}, - {'gt': '1', 'le': '2'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_03(self): - data = [ - {'gt': '1', 'lt': '2'}, {'gt': '3'}, {'gt': '4'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'gt': '4'}, - {'gt': '3', 'le': '4'}, - {'gt': '1', 'lt': '2'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_04(self): - data = [ - {'gt': '1', 'lt': '2'}, {'gt': '3', 'lt': '4'}, {'gt': '4'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'gt': '4'}, - {'gt': '3', 'lt': '4'}, - {'gt': '1', 'lt': '2'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_05(self): - data = [ - {'gt': '1'}, {'gt': '3', 'lt': '4'}, {'gt': '4'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'gt': '4'}, - {'gt': '3', 'lt': '4'}, - {'gt': '1', 'le': '3'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_06(self): - data = [ - {'lt': '2'}, {'lt': '3'}, {'lt': '4'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'ge': '3', 'lt': '4'}, - {'lt': '2'}, - {'ge': '2', 'lt': '3'}, - ] - self.assertEqual(result, expected) - - def test_dpkg_version_normalize_07(self): - data = [ - {'min': '2', 'max': '3'} - ] - result = DPKGVersion.normalize_version_criteria(data) - expected = [ - {'ge': '2', 'le': '3'} - ] - self.assertEqual(result, expected) - - def test_dpkg_version_pkgver_bad_version(self): - with self.assertRaises(DPKGBadVersionSyntax): - DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'lt': 'immabadversion'}] - ) - - def test_dpkg_version_pkgver_lt_false(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'lt': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_lt_true(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'lt': '1:8.2p1-4ubuntu0.5'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_le_false(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'le': '1:8.2p1-4ubuntu0.3'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_le_true_eq(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'le': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_le_true_less(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'le': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_gt_false(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.5'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_gt_true(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.3'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_ge_false(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'ge': '1:8.2p1-4ubuntu0.5'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_ge_true_eq(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'ge': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_ge_true_greater(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'ge': '1:8.2p1-4ubuntu0.3'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_is_equal_true(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'eq': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_is_equal_false(self): - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'eq': '1:8.2p1-4ubuntu0.3'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_within_ranges_true(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.2', - 'max': '1:8.3'}]) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_false(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.2', - 'max': '1:8.2'}]) - self.assertFalse(result) - - def test_dpkg_version_pkgver_within_ranges_multi(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.0', - 'max': '1:8.1'}, - {'min': '1:8.2', - 'max': '1:8.3'}]) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_no_max_true(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.0'}, - {'min': '1:8.1'}, - {'min': '1:8.2'}, - {'min': '1:8.3'}]) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_no_max_false(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.3'}, - {'min': '1:8.4'}, - {'min': '1:8.5'}]) - self.assertFalse(result) - - def test_dpkg_version_pkgver_within_ranges_mixed_true(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.0'}, - {'min': '1:8.1', - 'max': '1:8.1.1'}, - {'min': '1:8.2'}, - {'min': '1:8.3'}]) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_mixed_false(self): - result = DPKGVersion.is_version_within_ranges('1:8.2p1-4ubuntu0.4', - [{'min': '1:8.0'}, - {'min': '1:8.1', - 'max': '1:8.1.1'}, - {'min': '1:8.2', - 'max': '1:8.2'}, - {'min': '1:8.3'}]) - self.assertFalse(result) - - def test_dpkg_version_pkgver_within_ranges_mixed_lg_false(self): - # 1:8.2p1-4ubuntu0.4 - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.1', - 'lt': '1:8.1.1'}, - {'gt': '1:8.2', - 'lt': '1:8.2'}, - {'gt': '1:8.3'}] - ) - self.assertFalse(result) - - def test_dpkg_version_pkgver_within_ranges_mixed_lg_true(self): - # 1:8.2p1-4ubuntu0.4 - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.1', - 'lt': '1:8.1.1'}, - {'gt': '1:8.2p1-4ubuntu0.3', - 'lt': '1:8.2p1-4ubuntu0.5'}, - {'gt': '1:8.3'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_first_true(self): - # 1:8.2p1-4ubuntu0.4 - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'ge': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.1', - 'lt': '1:8.1.1'}, - {'gt': '1:8.2p1-4ubuntu0.3', - 'lt': '1:8.2p1-4ubuntu0.5'}, - {'gt': '1:8.3'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_mid_true(self): - # 1:8.2p1-4ubuntu0.4 - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.1', - 'le': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.2p1-4ubuntu0.6', - 'lt': '1:8.2p1-4ubuntu0.9'}, - {'gt': '1:8.3'}] - ) - self.assertTrue(result) - - def test_dpkg_version_pkgver_within_ranges_last_true(self): - # 1:8.2p1-4ubuntu0.4 - result = DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'gt': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.1', - 'le': '1:8.2p1-4ubuntu0.4'}, - {'gt': '1:8.2p1-4ubuntu0.6', - 'lt': '1:8.2p1-4ubuntu0.9'}, - {'eq': '1:8.2p1-4ubuntu0.4'}] - ) - self.assertTrue(result) - - def test_dpkg_version_invalid_op_name(self): - with self.assertRaises(Exception): - # 1:8.2p1-4ubuntu0.4 - DPKGVersion.is_version_within_ranges( - '1:8.2p1-4ubuntu0.4', - [{'foo': '1:8.2p1-4ubuntu0.4'}] - )