diff options
Diffstat (limited to 'test/units/module_utils/facts/test_collectors.py')
| -rw-r--r-- | test/units/module_utils/facts/test_collectors.py | 510 |
1 files changed, 510 insertions, 0 deletions
diff --git a/test/units/module_utils/facts/test_collectors.py b/test/units/module_utils/facts/test_collectors.py new file mode 100644 index 0000000..c480602 --- /dev/null +++ b/test/units/module_utils/facts/test_collectors.py @@ -0,0 +1,510 @@ +# unit tests for ansible fact collectors +# -*- coding: utf-8 -*- +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. +# + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import pytest + +from units.compat.mock import Mock, patch + +from . base import BaseFactsTest + +from ansible.module_utils.facts import collector + +from ansible.module_utils.facts.system.apparmor import ApparmorFactCollector +from ansible.module_utils.facts.system.caps import SystemCapabilitiesFactCollector +from ansible.module_utils.facts.system.cmdline import CmdLineFactCollector +from ansible.module_utils.facts.system.distribution import DistributionFactCollector +from ansible.module_utils.facts.system.dns import DnsFactCollector +from ansible.module_utils.facts.system.env import EnvFactCollector +from ansible.module_utils.facts.system.fips import FipsFactCollector +from ansible.module_utils.facts.system.pkg_mgr import PkgMgrFactCollector, OpenBSDPkgMgrFactCollector +from ansible.module_utils.facts.system.platform import PlatformFactCollector +from ansible.module_utils.facts.system.python import PythonFactCollector +from ansible.module_utils.facts.system.selinux import SelinuxFactCollector +from ansible.module_utils.facts.system.service_mgr import ServiceMgrFactCollector +from ansible.module_utils.facts.system.ssh_pub_keys import SshPubKeyFactCollector +from ansible.module_utils.facts.system.user import UserFactCollector + +from ansible.module_utils.facts.virtual.base import VirtualCollector +from ansible.module_utils.facts.network.base import NetworkCollector +from ansible.module_utils.facts.hardware.base import HardwareCollector + + +class CollectorException(Exception): + pass + + +class ExceptionThrowingCollector(collector.BaseFactCollector): + name = 'exc_throwing' + + def __init__(self, collectors=None, namespace=None, exception=None): + super(ExceptionThrowingCollector, self).__init__(collectors, namespace) + self._exception = exception or CollectorException('collection failed') + + def collect(self, module=None, collected_facts=None): + raise self._exception + + +class TestExceptionThrowingCollector(BaseFactsTest): + __test__ = True + gather_subset = ['exc_throwing'] + valid_subsets = ['exc_throwing'] + collector_class = ExceptionThrowingCollector + + def test_collect(self): + module = self._mock_module() + fact_collector = self.collector_class() + self.assertRaises(CollectorException, + fact_collector.collect, + module=module, + collected_facts=self.collected_facts) + + def test_collect_with_namespace(self): + module = self._mock_module() + fact_collector = self.collector_class() + self.assertRaises(CollectorException, + fact_collector.collect_with_namespace, + module=module, + collected_facts=self.collected_facts) + + +class TestApparmorFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'apparmor'] + valid_subsets = ['apparmor'] + fact_namespace = 'ansible_apparmor' + collector_class = ApparmorFactCollector + + def test_collect(self): + facts_dict = super(TestApparmorFacts, self).test_collect() + self.assertIn('status', facts_dict['apparmor']) + + +class TestCapsFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'caps'] + valid_subsets = ['caps'] + fact_namespace = 'ansible_system_capabilities' + collector_class = SystemCapabilitiesFactCollector + + def _mock_module(self): + mock_module = Mock() + mock_module.params = {'gather_subset': self.gather_subset, + 'gather_timeout': 10, + 'filter': '*'} + mock_module.get_bin_path = Mock(return_value='/usr/sbin/capsh') + mock_module.run_command = Mock(return_value=(0, 'Current: =ep', '')) + return mock_module + + +class TestCmdLineFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'cmdline'] + valid_subsets = ['cmdline'] + fact_namespace = 'ansible_cmdline' + collector_class = CmdLineFactCollector + + def test_parse_proc_cmdline_uefi(self): + uefi_cmdline = r'initrd=\70ef65e1a04a47aea04f7b5145ea3537\4.10.0-19-generic\initrd root=UUID=50973b75-4a66-4bf0-9764-2b7614489e64 ro quiet' + expected = {'initrd': r'\70ef65e1a04a47aea04f7b5145ea3537\4.10.0-19-generic\initrd', + 'root': 'UUID=50973b75-4a66-4bf0-9764-2b7614489e64', + 'quiet': True, + 'ro': True} + fact_collector = self.collector_class() + facts_dict = fact_collector._parse_proc_cmdline(uefi_cmdline) + + self.assertDictEqual(facts_dict, expected) + + def test_parse_proc_cmdline_fedora(self): + cmdline_fedora = r'BOOT_IMAGE=/vmlinuz-4.10.16-200.fc25.x86_64 root=/dev/mapper/fedora-root ro rd.lvm.lv=fedora/root rd.luks.uuid=luks-c80b7537-358b-4a07-b88c-c59ef187479b rd.lvm.lv=fedora/swap rhgb quiet LANG=en_US.UTF-8' # noqa + + expected = {'BOOT_IMAGE': '/vmlinuz-4.10.16-200.fc25.x86_64', + 'LANG': 'en_US.UTF-8', + 'quiet': True, + 'rd.luks.uuid': 'luks-c80b7537-358b-4a07-b88c-c59ef187479b', + 'rd.lvm.lv': 'fedora/swap', + 'rhgb': True, + 'ro': True, + 'root': '/dev/mapper/fedora-root'} + + fact_collector = self.collector_class() + facts_dict = fact_collector._parse_proc_cmdline(cmdline_fedora) + + self.assertDictEqual(facts_dict, expected) + + def test_parse_proc_cmdline_dup_console(self): + example = r'BOOT_IMAGE=/boot/vmlinuz-4.4.0-72-generic root=UUID=e12e46d9-06c9-4a64-a7b3-60e24b062d90 ro console=tty1 console=ttyS0' + + # FIXME: Two 'console' keywords? Using a dict for the fact value here loses info. Currently the 'last' one wins + expected = {'BOOT_IMAGE': '/boot/vmlinuz-4.4.0-72-generic', + 'root': 'UUID=e12e46d9-06c9-4a64-a7b3-60e24b062d90', + 'ro': True, + 'console': 'ttyS0'} + + fact_collector = self.collector_class() + facts_dict = fact_collector._parse_proc_cmdline(example) + + # TODO: fails because we lose a 'console' + self.assertDictEqual(facts_dict, expected) + + +class TestDistributionFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'distribution'] + valid_subsets = ['distribution'] + fact_namespace = 'ansible_distribution' + collector_class = DistributionFactCollector + + +class TestDnsFacts(BaseFactsTest): + + __test__ = True + gather_subset = ['!all', 'dns'] + valid_subsets = ['dns'] + fact_namespace = 'ansible_dns' + collector_class = DnsFactCollector + + +class TestEnvFacts(BaseFactsTest): + + __test__ = True + gather_subset = ['!all', 'env'] + valid_subsets = ['env'] + fact_namespace = 'ansible_env' + collector_class = EnvFactCollector + + def test_collect(self): + facts_dict = super(TestEnvFacts, self).test_collect() + self.assertIn('HOME', facts_dict['env']) + + +class TestFipsFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'fips'] + valid_subsets = ['fips'] + fact_namespace = 'ansible_fips' + collector_class = FipsFactCollector + + +class TestHardwareCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'hardware'] + valid_subsets = ['hardware'] + fact_namespace = 'ansible_hardware' + collector_class = HardwareCollector + collected_facts = {'ansible_architecture': 'x86_64'} + + +class TestNetworkCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'network'] + valid_subsets = ['network'] + fact_namespace = 'ansible_network' + collector_class = NetworkCollector + + +class TestPkgMgrFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'pkg_mgr'] + valid_subsets = ['pkg_mgr'] + fact_namespace = 'ansible_pkgmgr' + collector_class = PkgMgrFactCollector + collected_facts = { + "ansible_distribution": "Fedora", + "ansible_distribution_major_version": "28", + "ansible_os_family": "RedHat" + } + + def test_collect(self): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + + +class TestMacOSXPkgMgrFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'pkg_mgr'] + valid_subsets = ['pkg_mgr'] + fact_namespace = 'ansible_pkgmgr' + collector_class = PkgMgrFactCollector + collected_facts = { + "ansible_distribution": "MacOSX", + "ansible_distribution_major_version": "11", + "ansible_os_family": "Darwin" + } + + @patch('ansible.module_utils.facts.system.pkg_mgr.os.path.exists', side_effect=lambda x: x == '/opt/homebrew/bin/brew') + def test_collect_opt_homebrew(self, p_exists): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + self.assertEqual(facts_dict['pkg_mgr'], 'homebrew') + + @patch('ansible.module_utils.facts.system.pkg_mgr.os.path.exists', side_effect=lambda x: x == '/usr/local/bin/brew') + def test_collect_usr_homebrew(self, p_exists): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + self.assertEqual(facts_dict['pkg_mgr'], 'homebrew') + + @patch('ansible.module_utils.facts.system.pkg_mgr.os.path.exists', side_effect=lambda x: x == '/opt/local/bin/port') + def test_collect_macports(self, p_exists): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + self.assertEqual(facts_dict['pkg_mgr'], 'macports') + + +def _sanitize_os_path_apt_get(path): + if path == '/usr/bin/apt-get': + return True + else: + return False + + +class TestPkgMgrFactsAptFedora(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'pkg_mgr'] + valid_subsets = ['pkg_mgr'] + fact_namespace = 'ansible_pkgmgr' + collector_class = PkgMgrFactCollector + collected_facts = { + "ansible_distribution": "Fedora", + "ansible_distribution_major_version": "28", + "ansible_os_family": "RedHat", + "ansible_pkg_mgr": "apt" + } + + @patch('ansible.module_utils.facts.system.pkg_mgr.os.path.exists', side_effect=_sanitize_os_path_apt_get) + def test_collect(self, mock_os_path_exists): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + + +class TestOpenBSDPkgMgrFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'pkg_mgr'] + valid_subsets = ['pkg_mgr'] + fact_namespace = 'ansible_pkgmgr' + collector_class = OpenBSDPkgMgrFactCollector + + def test_collect(self): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, collected_facts=self.collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertIn('pkg_mgr', facts_dict) + self.assertEqual(facts_dict['pkg_mgr'], 'openbsd_pkg') + + +class TestPlatformFactCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'platform'] + valid_subsets = ['platform'] + fact_namespace = 'ansible_platform' + collector_class = PlatformFactCollector + + +class TestPythonFactCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'python'] + valid_subsets = ['python'] + fact_namespace = 'ansible_python' + collector_class = PythonFactCollector + + +class TestSelinuxFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'selinux'] + valid_subsets = ['selinux'] + fact_namespace = 'ansible_selinux' + collector_class = SelinuxFactCollector + + def test_no_selinux(self): + with patch('ansible.module_utils.facts.system.selinux.HAVE_SELINUX', False): + module = self._mock_module() + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['selinux']['status'], 'Missing selinux Python library') + return facts_dict + + +class TestServiceMgrFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'service_mgr'] + valid_subsets = ['service_mgr'] + fact_namespace = 'ansible_service_mgr' + collector_class = ServiceMgrFactCollector + + # TODO: dedupe some of this test code + + @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) + @patch('ansible.module_utils.facts.system.service_mgr.ServiceMgrFactCollector.is_systemd_managed', return_value=False) + @patch('ansible.module_utils.facts.system.service_mgr.ServiceMgrFactCollector.is_systemd_managed_offline', return_value=False) + @patch('ansible.module_utils.facts.system.service_mgr.os.path.exists', return_value=False) + @pytest.mark.skip(reason='faulty test') + def test_service_mgr_runit_one(self, mock_gfc, mock_ism, mock_ismo, mock_ope): + # no /proc/1/comm, ps returns non-0 + # should fallback to 'service' + module = self._mock_module() + module.run_command = Mock(return_value=(1, '', 'wat')) + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['service_mgr'], 'service') + + @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) + def test_no_proc1_ps_random_init(self, mock_gfc): + # no /proc/1/comm, ps returns '/sbin/sys11' which we dont know + # should end up return 'sys11' + module = self._mock_module() + module.run_command = Mock(return_value=(0, '/sbin/sys11', '')) + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['service_mgr'], 'sys11') + + @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) + @patch('ansible.module_utils.facts.system.service_mgr.ServiceMgrFactCollector.is_systemd_managed', return_value=False) + @patch('ansible.module_utils.facts.system.service_mgr.ServiceMgrFactCollector.is_systemd_managed_offline', return_value=False) + @patch('ansible.module_utils.facts.system.service_mgr.os.path.exists', return_value=False) + @pytest.mark.skip(reason='faulty test') + def test_service_mgr_runit_two(self, mock_gfc, mock_ism, mock_ismo, mock_ope): + # no /proc/1/comm, ps fails, distro and system are clowncar + # should end up return 'sys11' + module = self._mock_module() + module.run_command = Mock(return_value=(1, '', '')) + collected_facts = {'distribution': 'clowncar', + 'system': 'ClownCarOS'} + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, + collected_facts=collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['service_mgr'], 'service') + + @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value='runit-init') + @patch('ansible.module_utils.facts.system.service_mgr.os.path.islink', side_effect=lambda x: x == '/sbin/init') + @patch('ansible.module_utils.facts.system.service_mgr.os.readlink', side_effect=lambda x: '/sbin/runit-init' if x == '/sbin/init' else '/bin/false') + def test_service_mgr_runit(self, mock_gfc, mock_opl, mock_orl): + # /proc/1/comm contains 'runit-init', ps fails, service manager is runit + # should end up return 'runit' + module = self._mock_module() + module.run_command = Mock(return_value=(1, '', '')) + collected_facts = {'ansible_system': 'Linux'} + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, + collected_facts=collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['service_mgr'], 'runit') + + @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) + @patch('ansible.module_utils.facts.system.service_mgr.os.path.islink', side_effect=lambda x: x == '/sbin/init') + @patch('ansible.module_utils.facts.system.service_mgr.os.readlink', side_effect=lambda x: '/sbin/runit-init' if x == '/sbin/init' else '/bin/false') + def test_service_mgr_runit_no_comm(self, mock_gfc, mock_opl, mock_orl): + # no /proc/1/comm, ps returns 'COMMAND\n', service manager is runit + # should end up return 'runit' + module = self._mock_module() + module.run_command = Mock(return_value=(1, 'COMMAND\n', '')) + collected_facts = {'ansible_system': 'Linux'} + fact_collector = self.collector_class() + facts_dict = fact_collector.collect(module=module, + collected_facts=collected_facts) + self.assertIsInstance(facts_dict, dict) + self.assertEqual(facts_dict['service_mgr'], 'runit') + + # TODO: reenable these tests when we can mock more easily + +# @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) +# def test_sunos_fallback(self, mock_gfc): +# # no /proc/1/comm, ps fails, 'system' is SunOS +# # should end up return 'smf'? +# module = self._mock_module() +# # FIXME: the result here is a kluge to at least cover more of service_mgr.collect +# # TODO: remove +# # FIXME: have to force a pid for results here to get into any of the system/distro checks +# module.run_command = Mock(return_value=(1, ' 37 ', '')) +# collected_facts = {'system': 'SunOS'} +# fact_collector = self.collector_class(module=module) +# facts_dict = fact_collector.collect(collected_facts=collected_facts) +# print('facts_dict: %s' % facts_dict) +# self.assertIsInstance(facts_dict, dict) +# self.assertEqual(facts_dict['service_mgr'], 'smf') + +# @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) +# def test_aix_fallback(self, mock_gfc): +# # no /proc/1/comm, ps fails, 'system' is SunOS +# # should end up return 'smf'? +# module = self._mock_module() +# module.run_command = Mock(return_value=(1, '', '')) +# collected_facts = {'system': 'AIX'} +# fact_collector = self.collector_class(module=module) +# facts_dict = fact_collector.collect(collected_facts=collected_facts) +# print('facts_dict: %s' % facts_dict) +# self.assertIsInstance(facts_dict, dict) +# self.assertEqual(facts_dict['service_mgr'], 'src') + +# @patch('ansible.module_utils.facts.system.service_mgr.get_file_content', return_value=None) +# def test_linux_fallback(self, mock_gfc): +# # no /proc/1/comm, ps fails, 'system' is SunOS +# # should end up return 'smf'? +# module = self._mock_module() +# module.run_command = Mock(return_value=(1, ' 37 ', '')) +# collected_facts = {'system': 'Linux'} +# fact_collector = self.collector_class(module=module) +# facts_dict = fact_collector.collect(collected_facts=collected_facts) +# print('facts_dict: %s' % facts_dict) +# self.assertIsInstance(facts_dict, dict) +# self.assertEqual(facts_dict['service_mgr'], 'sdfadf') + + +class TestSshPubKeyFactCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'ssh_pub_keys'] + valid_subsets = ['ssh_pub_keys'] + fact_namespace = 'ansible_ssh_pub_leys' + collector_class = SshPubKeyFactCollector + + +class TestUserFactCollector(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'user'] + valid_subsets = ['user'] + fact_namespace = 'ansible_user' + collector_class = UserFactCollector + + +class TestVirtualFacts(BaseFactsTest): + __test__ = True + gather_subset = ['!all', 'virtual'] + valid_subsets = ['virtual'] + fact_namespace = 'ansible_virtual' + collector_class = VirtualCollector |