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
)