summaryrefslogtreecommitdiff
path: root/cloudinit/distros/tests
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit/distros/tests')
-rw-r--r--cloudinit/distros/tests/__init__.py0
-rw-r--r--cloudinit/distros/tests/test_init.py156
-rw-r--r--cloudinit/distros/tests/test_networking.py223
3 files changed, 0 insertions, 379 deletions
diff --git a/cloudinit/distros/tests/__init__.py b/cloudinit/distros/tests/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/cloudinit/distros/tests/__init__.py
+++ /dev/null
diff --git a/cloudinit/distros/tests/test_init.py b/cloudinit/distros/tests/test_init.py
deleted file mode 100644
index db534654..00000000
--- a/cloudinit/distros/tests/test_init.py
+++ /dev/null
@@ -1,156 +0,0 @@
-# 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
deleted file mode 100644
index ec508f4d..00000000
--- a/cloudinit/distros/tests/test_networking.py
+++ /dev/null
@@ -1,223 +0,0 @@
-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
-
- def try_set_link_up(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 TestBSDNetworkingTrySetLinkUp:
- def test_raises_notimplementederror(self):
- with pytest.raises(NotImplementedError):
- BSDNetworking().try_set_link_up("eth0")
-
-
-@mock.patch("cloudinit.net.is_up")
-@mock.patch("cloudinit.distros.networking.subp.subp")
-class TestLinuxNetworkingTrySetLinkUp:
- def test_calls_subp_return_true(self, m_subp, m_is_up):
- devname = "eth0"
- m_is_up.return_value = True
- is_success = LinuxNetworking().try_set_link_up(devname)
-
- assert (mock.call(['ip', 'link', 'set', devname, 'up']) ==
- m_subp.call_args_list[-1])
- assert is_success
-
- def test_calls_subp_return_false(self, m_subp, m_is_up):
- devname = "eth0"
- m_is_up.return_value = False
- is_success = LinuxNetworking().try_set_link_up(devname)
-
- assert (mock.call(['ip', 'link', 'set', devname, 'up']) ==
- m_subp.call_args_list[-1])
- assert not is_success
-
-
-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
- )
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-10 05:17:44 +0000