Merge pull request #18 from zdc/T2117-equuleus-20.3

T2117: Cloud-init updated to 20.3

3 files changed, 348 insertions, 0 deletions

diff --git a/cloudinit/distros/tests/__init__.py b/cloudinit/distros/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/cloudinit/distros/tests/__init__.py
diff --git a/cloudinit/distros/tests/test_init.py b/cloudinit/distros/tests/test_init.py
new file mode 100644
index 00000000..db534654
--- /dev/null
+++ b/cloudinit/distros/tests/test_init.py
@@ -0,0 +1,156 @@
+# Copyright (C) 2020 Canonical Ltd.
+#
+# Author: Daniel Watkins <oddbloke@ubuntu.com>
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+"""Tests for cloudinit/distros/__init__.py"""
+
+from unittest import mock
+
+import pytest
+
+from cloudinit.distros import _get_package_mirror_info, LDH_ASCII_CHARS
+
+
+# Define a set of characters we would expect to be replaced
+INVALID_URL_CHARS = [
+    chr(x) for x in range(127) if chr(x) not in LDH_ASCII_CHARS
+]
+for separator in [":", ".", "/", "#", "?", "@", "[", "]"]:
+    # Remove from the set characters that either separate hostname parts (":",
+    # "."), terminate hostnames ("/", "#", "?", "@"), or cause Python to be
+    # unable to parse URLs ("[", "]").
+    INVALID_URL_CHARS.remove(separator)
+
+
+class TestGetPackageMirrorInfo:
+    """
+    Tests for cloudinit.distros._get_package_mirror_info.
+
+    These supplement the tests in tests/unittests/test_distros/test_generic.py
+    which are more focused on testing a single production-like configuration.
+    These tests are more focused on specific aspects of the unit under test.
+    """
+
+    @pytest.mark.parametrize('mirror_info,expected', [
+        # Empty info gives empty return
+        ({}, {}),
+        # failsafe values used if present
+        ({'failsafe': {'primary': 'http://value', 'security': 'http://other'}},
+         {'primary': 'http://value', 'security': 'http://other'}),
+        # search values used if present
+        ({'search': {'primary': ['http://value'],
+                     'security': ['http://other']}},
+         {'primary': ['http://value'], 'security': ['http://other']}),
+        # failsafe values used if search value not present
+        ({'search': {'primary': ['http://value']},
+          'failsafe': {'security': 'http://other'}},
+         {'primary': ['http://value'], 'security': 'http://other'})
+    ])
+    def test_get_package_mirror_info_failsafe(self, mirror_info, expected):
+        """
+        Test the interaction between search and failsafe inputs
+
+        (This doesn't test the case where the mirror_filter removes all search
+        options; test_failsafe_used_if_all_search_results_filtered_out covers
+        that.)
+        """
+        assert expected == _get_package_mirror_info(mirror_info,
+                                                    mirror_filter=lambda x: x)
+
+    def test_failsafe_used_if_all_search_results_filtered_out(self):
+        """Test the failsafe option used if all search options eliminated."""
+        mirror_info = {
+            'search': {'primary': ['http://value']},
+            'failsafe': {'primary': 'http://other'}
+        }
+        assert {'primary': 'http://other'} == _get_package_mirror_info(
+            mirror_info, mirror_filter=lambda x: False)
+
+    @pytest.mark.parametrize('allow_ec2_mirror, platform_type', [
+        (True, 'ec2')
+    ])
+    @pytest.mark.parametrize('availability_zone,region,patterns,expected', (
+        # Test ec2_region alone
+        ('fk-fake-1f', None, ['http://EC2-%(ec2_region)s/ubuntu'],
+         ['http://ec2-fk-fake-1/ubuntu']),
+        # Test availability_zone alone
+        ('fk-fake-1f', None, ['http://AZ-%(availability_zone)s/ubuntu'],
+         ['http://az-fk-fake-1f/ubuntu']),
+        # Test region alone
+        (None, 'fk-fake-1', ['http://RG-%(region)s/ubuntu'],
+         ['http://rg-fk-fake-1/ubuntu']),
+        # Test that ec2_region is not available for non-matching AZs
+        ('fake-fake-1f', None,
+         ['http://EC2-%(ec2_region)s/ubuntu',
+          'http://AZ-%(availability_zone)s/ubuntu'],
+         ['http://az-fake-fake-1f/ubuntu']),
+        # Test that template order maintained
+        (None, 'fake-region',
+         ['http://RG-%(region)s-2/ubuntu', 'http://RG-%(region)s-1/ubuntu'],
+         ['http://rg-fake-region-2/ubuntu', 'http://rg-fake-region-1/ubuntu']),
+        # Test that non-ASCII hostnames are IDNA encoded;
+        # "IDNA-ТεЅТ̣".encode('idna') == b"xn--idna--4kd53hh6aba3q"
+        (None, 'ТεЅТ̣', ['http://www.IDNA-%(region)s.com/ubuntu'],
+         ['http://www.xn--idna--4kd53hh6aba3q.com/ubuntu']),
+        # Test that non-ASCII hostnames with a port are IDNA encoded;
+        # "IDNA-ТεЅТ̣".encode('idna') == b"xn--idna--4kd53hh6aba3q"
+        (None, 'ТεЅТ̣', ['http://www.IDNA-%(region)s.com:8080/ubuntu'],
+         ['http://www.xn--idna--4kd53hh6aba3q.com:8080/ubuntu']),
+        # Test that non-ASCII non-hostname parts of URLs are unchanged
+        (None, 'ТεЅТ̣', ['http://www.example.com/%(region)s/ubuntu'],
+         ['http://www.example.com/ТεЅТ̣/ubuntu']),
+        # Test that IPv4 addresses are unchanged
+        (None, 'fk-fake-1', ['http://192.168.1.1:8080/%(region)s/ubuntu'],
+         ['http://192.168.1.1:8080/fk-fake-1/ubuntu']),
+        # Test that IPv6 addresses are unchanged
+        (None, 'fk-fake-1',
+         ['http://[2001:67c:1360:8001::23]/%(region)s/ubuntu'],
+         ['http://[2001:67c:1360:8001::23]/fk-fake-1/ubuntu']),
+        # Test that unparseable URLs are filtered out of the mirror list
+        (None, 'inv[lid',
+         ['http://%(region)s.in.hostname/should/be/filtered',
+          'http://but.not.in.the.path/%(region)s'],
+         ['http://but.not.in.the.path/inv[lid']),
+        (None, '-some-region-',
+         ['http://-lead-ing.%(region)s.trail-ing-.example.com/ubuntu'],
+         ['http://lead-ing.some-region.trail-ing.example.com/ubuntu']),
+    ) + tuple(
+        # Dynamically generate a test case for each non-LDH
+        # (Letters/Digits/Hyphen) ASCII character, testing that it is
+        # substituted with a hyphen
+        (None, 'fk{0}fake{0}1'.format(invalid_char),
+         ['http://%(region)s/ubuntu'], ['http://fk-fake-1/ubuntu'])
+        for invalid_char in INVALID_URL_CHARS
+    ))
+    def test_valid_substitution(self,
+                                allow_ec2_mirror,
+                                platform_type,
+                                availability_zone,
+                                region,
+                                patterns,
+                                expected):
+        """Test substitution works as expected."""
+        flag_path = "cloudinit.distros." \
+                    "ALLOW_EC2_MIRRORS_ON_NON_AWS_INSTANCE_TYPES"
+
+        m_data_source = mock.Mock(
+            availability_zone=availability_zone,
+            region=region,
+            platform_type=platform_type
+        )
+        mirror_info = {'search': {'primary': patterns}}
+
+        with mock.patch(flag_path, allow_ec2_mirror):
+            ret = _get_package_mirror_info(
+                mirror_info,
+                data_source=m_data_source,
+                mirror_filter=lambda x: x
+            )
+        print(allow_ec2_mirror)
+        print(platform_type)
+        print(availability_zone)
+        print(region)
+        print(patterns)
+        print(expected)
+        assert {'primary': expected} == ret
diff --git a/cloudinit/distros/tests/test_networking.py b/cloudinit/distros/tests/test_networking.py
new file mode 100644
index 00000000..b9a63842
--- /dev/null
+++ b/cloudinit/distros/tests/test_networking.py
@@ -0,0 +1,192 @@
+from unittest import mock
+
+import pytest
+
+from cloudinit import net
+from cloudinit.distros.networking import (
+    BSDNetworking,
+    LinuxNetworking,
+    Networking,
+)
+
+# See https://docs.pytest.org/en/stable/example
+# /parametrize.html#parametrizing-conditional-raising
+from contextlib import ExitStack as does_not_raise
+
+
+@pytest.yield_fixture
+def generic_networking_cls():
+    """Returns a direct Networking subclass which errors on /sys usage.
+
+    This enables the direct testing of functionality only present on the
+    ``Networking`` super-class, and provides a check on accidentally using /sys
+    in that context.
+    """
+
+    class TestNetworking(Networking):
+        def is_physical(self, *args, **kwargs):
+            raise NotImplementedError
+
+        def settle(self, *args, **kwargs):
+            raise NotImplementedError
+
+    error = AssertionError("Unexpectedly used /sys in generic networking code")
+    with mock.patch(
+        "cloudinit.net.get_sys_class_path", side_effect=error,
+    ):
+        yield TestNetworking
+
+
+@pytest.yield_fixture
+def sys_class_net(tmpdir):
+    sys_class_net_path = tmpdir.join("sys/class/net")
+    sys_class_net_path.ensure_dir()
+    with mock.patch(
+        "cloudinit.net.get_sys_class_path",
+        return_value=sys_class_net_path.strpath + "/",
+    ):
+        yield sys_class_net_path
+
+
+class TestBSDNetworkingIsPhysical:
+    def test_raises_notimplementederror(self):
+        with pytest.raises(NotImplementedError):
+            BSDNetworking().is_physical("eth0")
+
+
+class TestLinuxNetworkingIsPhysical:
+    def test_returns_false_by_default(self, sys_class_net):
+        assert not LinuxNetworking().is_physical("eth0")
+
+    def test_returns_false_if_devname_exists_but_not_physical(
+        self, sys_class_net
+    ):
+        devname = "eth0"
+        sys_class_net.join(devname).mkdir()
+        assert not LinuxNetworking().is_physical(devname)
+
+    def test_returns_true_if_device_is_physical(self, sys_class_net):
+        devname = "eth0"
+        device_dir = sys_class_net.join(devname)
+        device_dir.mkdir()
+        device_dir.join("device").write("")
+
+        assert LinuxNetworking().is_physical(devname)
+
+
+class TestBSDNetworkingSettle:
+    def test_settle_doesnt_error(self):
+        # This also implicitly tests that it doesn't use subp.subp
+        BSDNetworking().settle()
+
+
+@pytest.mark.usefixtures("sys_class_net")
+@mock.patch("cloudinit.distros.networking.util.udevadm_settle", autospec=True)
+class TestLinuxNetworkingSettle:
+    def test_no_arguments(self, m_udevadm_settle):
+        LinuxNetworking().settle()
+
+        assert [mock.call(exists=None)] == m_udevadm_settle.call_args_list
+
+    def test_exists_argument(self, m_udevadm_settle):
+        LinuxNetworking().settle(exists="ens3")
+
+        expected_path = net.sys_dev_path("ens3")
+        assert [
+            mock.call(exists=expected_path)
+        ] == m_udevadm_settle.call_args_list
+
+
+class TestNetworkingWaitForPhysDevs:
+    @pytest.fixture
+    def wait_for_physdevs_netcfg(self):
+        """This config is shared across all the tests in this class."""
+
+        def ethernet(mac, name, driver=None, device_id=None):
+            v2_cfg = {"set-name": name, "match": {"macaddress": mac}}
+            if driver:
+                v2_cfg["match"].update({"driver": driver})
+            if device_id:
+                v2_cfg["match"].update({"device_id": device_id})
+
+            return v2_cfg
+
+        physdevs = [
+            ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
+            ["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
+        ]
+        netcfg = {
+            "version": 2,
+            "ethernets": {args[1]: ethernet(*args) for args in physdevs},
+        }
+        return netcfg
+
+    def test_skips_settle_if_all_present(
+        self, generic_networking_cls, wait_for_physdevs_netcfg,
+    ):
+        networking = generic_networking_cls()
+        with mock.patch.object(
+            networking, "get_interfaces_by_mac"
+        ) as m_get_interfaces_by_mac:
+            m_get_interfaces_by_mac.side_effect = iter(
+                [{"aa:bb:cc:dd:ee:ff": "eth0", "00:11:22:33:44:55": "ens3"}]
+            )
+            with mock.patch.object(
+                networking, "settle", autospec=True
+            ) as m_settle:
+                networking.wait_for_physdevs(wait_for_physdevs_netcfg)
+            assert 0 == m_settle.call_count
+
+    def test_calls_udev_settle_on_missing(
+        self, generic_networking_cls, wait_for_physdevs_netcfg,
+    ):
+        networking = generic_networking_cls()
+        with mock.patch.object(
+            networking, "get_interfaces_by_mac"
+        ) as m_get_interfaces_by_mac:
+            m_get_interfaces_by_mac.side_effect = iter(
+                [
+                    {
+                        "aa:bb:cc:dd:ee:ff": "eth0"
+                    },  # first call ens3 is missing
+                    {
+                        "aa:bb:cc:dd:ee:ff": "eth0",
+                        "00:11:22:33:44:55": "ens3",
+                    },  # second call has both
+                ]
+            )
+            with mock.patch.object(
+                networking, "settle", autospec=True
+            ) as m_settle:
+                networking.wait_for_physdevs(wait_for_physdevs_netcfg)
+            m_settle.assert_called_with(exists="ens3")
+
+    @pytest.mark.parametrize(
+        "strict,expectation",
+        [(True, pytest.raises(RuntimeError)), (False, does_not_raise())],
+    )
+    def test_retrying_and_strict_behaviour(
+        self,
+        strict,
+        expectation,
+        generic_networking_cls,
+        wait_for_physdevs_netcfg,
+    ):
+        networking = generic_networking_cls()
+        with mock.patch.object(
+            networking, "get_interfaces_by_mac"
+        ) as m_get_interfaces_by_mac:
+            m_get_interfaces_by_mac.return_value = {}
+
+            with mock.patch.object(
+                networking, "settle", autospec=True
+            ) as m_settle:
+                with expectation:
+                    networking.wait_for_physdevs(
+                        wait_for_physdevs_netcfg, strict=strict
+                    )
+
+        assert (
+            5 * len(wait_for_physdevs_netcfg["ethernets"])
+            == m_settle.call_count
+        )