testing: monkeypatch system_info call in unit tests (SC-533) (#1117)

testing: monkeypatch system_info call in unit tests

system_info can make calls that read or write from the filesystem, which
should require special mocking. It is also decorated with 'lru_cache',
which means test authors often don't realize they need to be mocking.
Also, we don't actually want the results from the user's local
machine, so monkeypatching it across all tests should be reasonable.

Additionally, moved some of 'system_info` into a helper function to
reduce the surface area of the monkeypatch, added tests for the new
function (and fixed a bug as a result), and removed related mocks that
should be no longer needed.

cloudinit/analyze/tests/test_boot.py

@@ -9,20 +9,11 @@
 
 class TestDistroChecker(CiTestCase):
 
-    @mock.patch('cloudinit.util.system_info', return_value={'dist': ('', '',
-                                                                     ''),
-                                                            'system': ''})
-    @mock.patch('cloudinit.util.get_linux_distro', return_value=('', '', ''))
-    @mock.patch('cloudinit.util.is_FreeBSD', return_value=False)
-    def test_blank_distro(self, m_sys_info, m_get_linux_distro, m_free_bsd):
+    def test_blank_distro(self):
         self.assertEqual(err_code, dist_check_timestamp())
 
-    @mock.patch('cloudinit.util.system_info', return_value={'dist': ('', '',
-                                                                     '')})
-    @mock.patch('cloudinit.util.get_linux_distro', return_value=('', '', ''))
     @mock.patch('cloudinit.util.is_FreeBSD', return_value=True)
-    def test_freebsd_gentoo_cant_find(self, m_sys_info,
-                                      m_get_linux_distro, m_is_FreeBSD):
+    def test_freebsd_gentoo_cant_find(self, m_is_FreeBSD):
         self.assertEqual(err_code, dist_check_timestamp())
 
     @mock.patch('cloudinit.subp.subp', return_value=(0, 1))

cloudinit/config/tests/test_set_passwords.py

@@ -121,13 +121,11 @@ def test_bsd_calls_custom_pw_cmds_to_set_and_expire_passwords(
             m_subp.call_args_list)
 
     @mock.patch(MODPATH + "util.multi_log")
-    @mock.patch(MODPATH + "util.is_BSD")
     @mock.patch(MODPATH + "subp.subp")
     def test_handle_on_chpasswd_list_creates_random_passwords(
-        self, m_subp, m_is_bsd, m_multi_log
+        self, m_subp, m_multi_log
     ):
         """handle parses command set random passwords."""
-        m_is_bsd.return_value = False
         cloud = self.tmp_cloud(distro='ubuntu')
         valid_random_pwds = [
             'root:R',

cloudinit/tests/test_util.py

@@ -815,6 +815,44 @@ def test_get_linux_distro_plat_data(self, m_platform_dist,
         self.assertEqual(('foo', '1.1', 'aarch64'), dist)
 
 
+class TestGetVariant:
+    @pytest.mark.parametrize('info, expected_variant', [
+        ({'system': 'Linux', 'dist': ('almalinux',)}, 'almalinux'),
+        ({'system': 'linux', 'dist': ('alpine',)}, 'alpine'),
+        ({'system': 'linux', 'dist': ('arch',)}, 'arch'),
+        ({'system': 'linux', 'dist': ('centos',)}, 'centos'),
+        ({'system': 'linux', 'dist': ('cloudlinux',)}, 'cloudlinux'),
+        ({'system': 'linux', 'dist': ('debian',)}, 'debian'),
+        ({'system': 'linux', 'dist': ('eurolinux',)}, 'eurolinux'),
+        ({'system': 'linux', 'dist': ('fedora',)}, 'fedora'),
+        ({'system': 'linux', 'dist': ('openEuler',)}, 'openeuler'),
+        ({'system': 'linux', 'dist': ('photon',)}, 'photon'),
+        ({'system': 'linux', 'dist': ('rhel',)}, 'rhel'),
+        ({'system': 'linux', 'dist': ('rocky',)}, 'rocky'),
+        ({'system': 'linux', 'dist': ('suse',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('virtuozzo',)}, 'virtuozzo'),
+        ({'system': 'linux', 'dist': ('ubuntu',)}, 'ubuntu'),
+        ({'system': 'linux', 'dist': ('linuxmint',)}, 'ubuntu'),
+        ({'system': 'linux', 'dist': ('mint',)}, 'ubuntu'),
+        ({'system': 'linux', 'dist': ('redhat',)}, 'rhel'),
+        ({'system': 'linux', 'dist': ('opensuse',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('opensuse-tumbleweed',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('opensuse-leap',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('sles',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('sle_hpc',)}, 'suse'),
+        ({'system': 'linux', 'dist': ('my_distro',)}, 'linux'),
+        ({'system': 'Windows', 'dist': ('dontcare',)}, 'windows'),
+        ({'system': 'Darwin', 'dist': ('dontcare',)}, 'darwin'),
+        ({'system': 'Freebsd', 'dist': ('dontcare',)}, 'freebsd'),
+        ({'system': 'Netbsd', 'dist': ('dontcare',)}, 'netbsd'),
+        ({'system': 'Openbsd', 'dist': ('dontcare',)}, 'openbsd'),
+        ({'system': 'Dragonfly', 'dist': ('dontcare',)}, 'dragonfly'),
+    ])
+    def test_get_variant(self, info, expected_variant):
+        """Verify we get the correct variant name"""
+        assert util._get_variant(info) == expected_variant
+
+
 class TestJsonDumps(CiTestCase):
     def test_is_str(self):
         """json_dumps should return a string."""

cloudinit/util.py

@@ -533,42 +533,45 @@ def get_linux_distro():
     return (distro_name, distro_version, flavor)
 
 
-@lru_cache()
-def system_info():
-    info = {
-        'platform': platform.platform(),
-        'system': platform.system(),
-        'release': platform.release(),
-        'python': platform.python_version(),
-        'uname': list(platform.uname()),
-        'dist': get_linux_distro()
-    }
+def _get_variant(info):
     system = info['system'].lower()
-    var = 'unknown'
+    variant = 'unknown'
     if system == "linux":
         linux_dist = info['dist'][0].lower()
         if linux_dist in (
                 'almalinux', 'alpine', 'arch', 'centos', 'cloudlinux',
-                'debian', 'eurolinux', 'fedora', 'openEuler', 'photon',
+                'debian', 'eurolinux', 'fedora', 'openeuler', 'photon',
                 'rhel', 'rocky', 'suse', 'virtuozzo'):
-            var = linux_dist
+            variant = linux_dist
         elif linux_dist in ('ubuntu', 'linuxmint', 'mint'):
-            var = 'ubuntu'
+            variant = 'ubuntu'
         elif linux_dist == 'redhat':
-            var = 'rhel'
+            variant = 'rhel'
         elif linux_dist in (
                 'opensuse', 'opensuse-tumbleweed', 'opensuse-leap',
                 'sles', 'sle_hpc'):
-            var = 'suse'
+            variant = 'suse'
         else:
-            var = 'linux'
+            variant = 'linux'
     elif system in (
             'windows', 'darwin', "freebsd", "netbsd",
             "openbsd", "dragonfly"):
-        var = system
+        variant = system
+
+    return variant
 
-    info['variant'] = var
 
+@lru_cache()
+def system_info():
+    info = {
+        'platform': platform.platform(),
+        'system': platform.system(),
+        'release': platform.release(),
+        'python': platform.python_version(),
+        'uname': list(platform.uname()),
+        'dist': get_linux_distro()
+    }
+    info['variant'] = _get_variant(info)
     return info
 
 

conftest.py

@@ -12,7 +12,7 @@
 
 import pytest
 
-from cloudinit import helpers, subp
+from cloudinit import helpers, subp, util
 
 
 class _FixtureUtils:
@@ -201,3 +201,19 @@ def paths(tmpdir):
         "run_dir": tmpdir.mkdir("run_dir").strpath,
     }
     return helpers.Paths(dirs)
+
+
+@pytest.fixture(autouse=True, scope='session')
+def monkeypatch_system_info():
+    def my_system_info():
+        return {
+            "platform": "invalid",
+            "system": "invalid",
+            "release": "invalid",
+            "python": "invalid",
+            "uname": ["invalid"] * 6,
+            "dist": ("Distro", "-1.1", "Codename"),
+            "variant": "ubuntu"
+        }
+
+    util.system_info = my_system_info

tests/unittests/test_datasource/test_azure.py

@@ -1789,9 +1789,8 @@ def test_fallback_network_config(self, mock_fallback, mock_dd,
             config_driver=True)
 
     @mock.patch(MOCKPATH + 'net.get_interfaces', autospec=True)
-    @mock.patch(MOCKPATH + 'util.is_FreeBSD')
     def test_blacklist_through_distro(
-            self, m_is_freebsd, m_net_get_interfaces):
+            self, m_net_get_interfaces):
         """Verify Azure DS updates blacklist drivers in the distro's
            networking object."""
         odata = {'HostName': "myhost", 'UserName': "myuser"}
@@ -1805,7 +1804,6 @@ def test_blacklist_through_distro(
         self.assertEqual(distro.networking.blacklist_drivers,
                          dsaz.BLACKLIST_DRIVERS)
 
-        m_is_freebsd.return_value = False
         distro.networking.get_interfaces_by_mac()
         m_net_get_interfaces.assert_called_with(
             blacklist_drivers=dsaz.BLACKLIST_DRIVERS)
@@ -3219,7 +3217,6 @@ def test_poll_imds_report_ready_failure_raises_exc_and_doesnt_write_marker(
 @mock.patch(MOCKPATH + 'DataSourceAzure._report_ready', mock.MagicMock())
 @mock.patch(MOCKPATH + 'subp.subp', mock.MagicMock())
 @mock.patch(MOCKPATH + 'util.write_file', mock.MagicMock())
-@mock.patch(MOCKPATH + 'util.is_FreeBSD')
 @mock.patch('cloudinit.sources.helpers.netlink.'
             'wait_for_media_disconnect_connect')
 @mock.patch('cloudinit.net.dhcp.EphemeralIPv4Network', autospec=True)
@@ -3236,10 +3233,8 @@ def setUp(self):
 
     def test_poll_imds_returns_ovf_env(self, m_request,
                                        m_dhcp, m_net,
-                                       m_media_switch,
-                                       m_is_bsd):
+                                       m_media_switch):
         """The _poll_imds method should return the ovf_env.xml."""
-        m_is_bsd.return_value = False
         m_media_switch.return_value = None
         m_dhcp.return_value = [{
             'interface': 'eth9', 'fixed-address': '192.168.2.9',
@@ -3268,10 +3263,8 @@ def test_poll_imds_returns_ovf_env(self, m_request,
 
     def test__reprovision_calls__poll_imds(self, m_request,
                                            m_dhcp, m_net,
-                                           m_media_switch,
-                                           m_is_bsd):
+                                           m_media_switch):
         """The _reprovision method should call poll IMDS."""
-        m_is_bsd.return_value = False
         m_media_switch.return_value = None
         m_dhcp.return_value = [{
             'interface': 'eth9', 'fixed-address': '192.168.2.9',

tests/unittests/test_datasource/test_azure_helper.py

@@ -129,9 +129,7 @@ def test_from_dhcp_client(self):
         self.dhcp_options.return_value = {"eth0": {"unknown_245": "5:4:3:2"}}
         self.assertEqual('5.4.3.2', wa_shim.find_endpoint(None))
 
-    @mock.patch('cloudinit.sources.helpers.azure.util.is_FreeBSD')
-    def test_latest_lease_used(self, m_is_freebsd):
-        m_is_freebsd.return_value = False  # To avoid hitting load_file
+    def test_latest_lease_used(self):
         encoded_addresses = ['5:4:3:2', '4:3:2:1']
         file_content = '\n'.join([self._build_lease_content(encoded_address)
                                   for encoded_address in encoded_addresses])

tests/unittests/test_datasource/test_configdrive.py

@@ -486,11 +486,10 @@ def test_subplatform_config_drive_when_starts_with_dev(self):
                                           None,
                                           helpers.Paths({}))
         with mock.patch(M_PATH + 'find_candidate_devs') as m_find_devs:
-            with mock.patch(M_PATH + 'util.is_FreeBSD', return_value=False):
-                with mock.patch(M_PATH + 'util.mount_cb'):
-                    with mock.patch(M_PATH + 'on_first_boot'):
-                        m_find_devs.return_value = ['/dev/anything']
-                        self.assertEqual(True, cfg_ds.get_data())
+            with mock.patch(M_PATH + 'util.mount_cb'):
+                with mock.patch(M_PATH + 'on_first_boot'):
+                    m_find_devs.return_value = ['/dev/anything']
+                    self.assertEqual(True, cfg_ds.get_data())
         self.assertEqual('config-disk (/dev/anything)', cfg_ds.subplatform)
 
 

tests/unittests/test_distros/test_netconfig.py

@@ -241,8 +241,6 @@ class TestNetCfgDistroBase(FilesystemMockingTestCase):
     def setUp(self):
         super(TestNetCfgDistroBase, self).setUp()
         self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
-        self.add_patch('cloudinit.util.system_info', 'm_sysinfo')
-        self.m_sysinfo.return_value = {'dist': ('Distro', '99.1', 'Codename')}
 
     def _get_distro(self, dname, renderers=None):
         cls = distros.fetch(dname)
@@ -783,7 +781,10 @@ def test_apply_network_config_v1_with_netplan(self):
                 """),
         }
 
-        with mock.patch('cloudinit.util.is_FreeBSD', return_value=False):
+        with mock.patch(
+            'cloudinit.net.netplan.get_devicelist',
+            return_value=[]
+        ):
             self._apply_and_verify(self.distro.apply_network_config,
                                    V1_NET_CFG,
                                    expected_cfgs=expected_cfgs.copy(),

tests/unittests/test_distros/test_user_data_normalize.py

@@ -26,8 +26,6 @@ class TestUGNormalize(TestCase):
     def setUp(self):
         super(TestUGNormalize, self).setUp()
         self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
-        self.add_patch('cloudinit.util.system_info', 'm_sysinfo')
-        self.m_sysinfo.return_value = {'dist': ('Distro', '99.1', 'Codename')}
 
     def _make_distro(self, dtype, def_user=None):
         cfg = dict(settings.CFG_BUILTIN)

tests/unittests/test_handler/test_handler_ntp.py

@@ -37,8 +37,6 @@ def setUp(self):
         self.new_root = self.tmp_dir()
         self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
         self.m_snappy.return_value = False
-        self.add_patch('cloudinit.util.system_info', 'm_sysinfo')
-        self.m_sysinfo.return_value = {'dist': ('Distro', '99.1', 'Codename')}
         self.new_root = self.reRoot()
         self._get_cloud = partial(
             get_cloud,
@@ -510,35 +508,38 @@ def test_ntp_the_whole_package(self, m_sysd, m_select, m_subp, m_dsubp):
 
             self.assertEqual(expected_content, util.load_file(confpath))
 
-    def test_opensuse_picks_chrony(self):
+    @mock.patch('cloudinit.util.system_info')
+    def test_opensuse_picks_chrony(self, m_sysinfo):
         """Test opensuse picks chrony or ntp on certain distro versions"""
         #  < 15.0  => ntp
-        self.m_sysinfo.return_value = {'dist':
-                                       ('openSUSE', '13.2', 'Harlequin')}
+        m_sysinfo.return_value = {
+            'dist': ('openSUSE', '13.2', 'Harlequin')
+        }
         mycloud = self._get_cloud('opensuse')
         expected_client = mycloud.distro.preferred_ntp_clients[0]
         self.assertEqual('ntp', expected_client)
 
         #  >= 15.0 and  not openSUSE => chrony
-        self.m_sysinfo.return_value = {'dist':
-                                       ('SLES', '15.0',
-                                        'SUSE Linux Enterprise Server 15')}
+        m_sysinfo.return_value = {
+            'dist': ('SLES', '15.0', 'SUSE Linux Enterprise Server 15')
+        }
         mycloud = self._get_cloud('sles')
         expected_client = mycloud.distro.preferred_ntp_clients[0]
         self.assertEqual('chrony', expected_client)
 
         #  >= 15.0 and  openSUSE and ver != 42  => chrony
-        self.m_sysinfo.return_value = {'dist': ('openSUSE Tumbleweed',
-                                                '20180326',
-                                                'timbleweed')}
+        m_sysinfo.return_value = {
+            'dist': ('openSUSE Tumbleweed', '20180326', 'timbleweed')
+        }
         mycloud = self._get_cloud('opensuse')
         expected_client = mycloud.distro.preferred_ntp_clients[0]
         self.assertEqual('chrony', expected_client)
 
-    def test_ubuntu_xenial_picks_ntp(self):
+    @mock.patch('cloudinit.util.system_info')
+    def test_ubuntu_xenial_picks_ntp(self, m_sysinfo):
         """Test Ubuntu picks ntp on xenial release"""
 
-        self.m_sysinfo.return_value = {'dist': ('Ubuntu', '16.04', 'xenial')}
+        m_sysinfo.return_value = {'dist': ('Ubuntu', '16.04', 'xenial')}
         mycloud = self._get_cloud('ubuntu')
         expected_client = mycloud.distro.preferred_ntp_clients[0]
         self.assertEqual('ntp', expected_client)

tests/unittests/test_net.py

@@ -5373,21 +5373,20 @@ def test_select_none_found_raises(self, m_eni_avail, m_sysc_avail):
                           priority=['sysconfig', 'eni'])
 
     @mock.patch("cloudinit.net.sysconfig.available_sysconfig")
-    @mock.patch("cloudinit.util.get_linux_distro")
-    def test_sysconfig_available_uses_variant_mapping(self, m_distro, m_avail):
+    @mock.patch("cloudinit.util.system_info")
+    def test_sysconfig_available_uses_variant_mapping(self, m_info, m_avail):
         m_avail.return_value = True
-        distro_values = [
-            ('opensuse', '', ''),
-            ('opensuse-leap', '', ''),
-            ('opensuse-tumbleweed', '', ''),
-            ('sles', '', ''),
-            ('centos', '', ''),
-            ('eurolinux', '', ''),
-            ('fedora', '', ''),
-            ('redhat', '', ''),
+        variants = [
+            'suse',
+            'centos',
+            'eurolinux',
+            'fedora',
+            'rhel',
         ]
-        for (distro_name, distro_version, flavor) in distro_values:
-            m_distro.return_value = (distro_name, distro_version, flavor)
+        for distro_name in variants:
+            m_info.return_value = {
+                "variant": distro_name
+            }
             if hasattr(util.system_info, "cache_clear"):
                 util.system_info.cache_clear()
             result = sysconfig.available()