|
Packit Service |
a04d08 |
from unittest import mock
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
import pytest
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
from cloudinit import net
|
|
Packit Service |
a04d08 |
from cloudinit.distros.networking import (
|
|
Packit Service |
a04d08 |
BSDNetworking,
|
|
Packit Service |
a04d08 |
LinuxNetworking,
|
|
Packit Service |
a04d08 |
Networking,
|
|
Packit Service |
a04d08 |
)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# See https://docs.pytest.org/en/stable/example
|
|
Packit Service |
a04d08 |
# /parametrize.html#parametrizing-conditional-raising
|
|
Packit Service |
a04d08 |
from contextlib import ExitStack as does_not_raise
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@pytest.yield_fixture
|
|
Packit Service |
a04d08 |
def generic_networking_cls():
|
|
Packit Service |
a04d08 |
"""Returns a direct Networking subclass which errors on /sys usage.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
This enables the direct testing of functionality only present on the
|
|
Packit Service |
a04d08 |
``Networking`` super-class, and provides a check on accidentally using /sys
|
|
Packit Service |
a04d08 |
in that context.
|
|
Packit Service |
a04d08 |
"""
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestNetworking(Networking):
|
|
Packit Service |
a04d08 |
def is_physical(self, *args, **kwargs):
|
|
Packit Service |
a04d08 |
raise NotImplementedError
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def settle(self, *args, **kwargs):
|
|
Packit Service |
a04d08 |
raise NotImplementedError
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
error = AssertionError("Unexpectedly used /sys in generic networking code")
|
|
Packit Service |
a04d08 |
with mock.patch(
|
|
Packit Service |
a04d08 |
"cloudinit.net.get_sys_class_path", side_effect=error,
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
yield TestNetworking
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@pytest.yield_fixture
|
|
Packit Service |
a04d08 |
def sys_class_net(tmpdir):
|
|
Packit Service |
a04d08 |
sys_class_net_path = tmpdir.join("sys/class/net")
|
|
Packit Service |
a04d08 |
sys_class_net_path.ensure_dir()
|
|
Packit Service |
a04d08 |
with mock.patch(
|
|
Packit Service |
a04d08 |
"cloudinit.net.get_sys_class_path",
|
|
Packit Service |
a04d08 |
return_value=sys_class_net_path.strpath + "/",
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
yield sys_class_net_path
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestBSDNetworkingIsPhysical:
|
|
Packit Service |
a04d08 |
def test_raises_notimplementederror(self):
|
|
Packit Service |
a04d08 |
with pytest.raises(NotImplementedError):
|
|
Packit Service |
a04d08 |
BSDNetworking().is_physical("eth0")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestLinuxNetworkingIsPhysical:
|
|
Packit Service |
a04d08 |
def test_returns_false_by_default(self, sys_class_net):
|
|
Packit Service |
a04d08 |
assert not LinuxNetworking().is_physical("eth0")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_returns_false_if_devname_exists_but_not_physical(
|
|
Packit Service |
a04d08 |
self, sys_class_net
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
devname = "eth0"
|
|
Packit Service |
a04d08 |
sys_class_net.join(devname).mkdir()
|
|
Packit Service |
a04d08 |
assert not LinuxNetworking().is_physical(devname)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_returns_true_if_device_is_physical(self, sys_class_net):
|
|
Packit Service |
a04d08 |
devname = "eth0"
|
|
Packit Service |
a04d08 |
device_dir = sys_class_net.join(devname)
|
|
Packit Service |
a04d08 |
device_dir.mkdir()
|
|
Packit Service |
a04d08 |
device_dir.join("device").write("")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
assert LinuxNetworking().is_physical(devname)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestBSDNetworkingSettle:
|
|
Packit Service |
a04d08 |
def test_settle_doesnt_error(self):
|
|
Packit Service |
a04d08 |
# This also implicitly tests that it doesn't use subp.subp
|
|
Packit Service |
a04d08 |
BSDNetworking().settle()
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@pytest.mark.usefixtures("sys_class_net")
|
|
Packit Service |
a04d08 |
@mock.patch("cloudinit.distros.networking.util.udevadm_settle", autospec=True)
|
|
Packit Service |
a04d08 |
class TestLinuxNetworkingSettle:
|
|
Packit Service |
a04d08 |
def test_no_arguments(self, m_udevadm_settle):
|
|
Packit Service |
a04d08 |
LinuxNetworking().settle()
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
assert [mock.call(exists=None)] == m_udevadm_settle.call_args_list
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_exists_argument(self, m_udevadm_settle):
|
|
Packit Service |
a04d08 |
LinuxNetworking().settle(exists="ens3")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
expected_path = net.sys_dev_path("ens3")
|
|
Packit Service |
a04d08 |
assert [
|
|
Packit Service |
a04d08 |
mock.call(exists=expected_path)
|
|
Packit Service |
a04d08 |
] == m_udevadm_settle.call_args_list
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestNetworkingWaitForPhysDevs:
|
|
Packit Service |
a04d08 |
@pytest.fixture
|
|
Packit Service |
a04d08 |
def wait_for_physdevs_netcfg(self):
|
|
Packit Service |
a04d08 |
"""This config is shared across all the tests in this class."""
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def ethernet(mac, name, driver=None, device_id=None):
|
|
Packit Service |
a04d08 |
v2_cfg = {"set-name": name, "match": {"macaddress": mac}}
|
|
Packit Service |
a04d08 |
if driver:
|
|
Packit Service |
a04d08 |
v2_cfg["match"].update({"driver": driver})
|
|
Packit Service |
a04d08 |
if device_id:
|
|
Packit Service |
a04d08 |
v2_cfg["match"].update({"device_id": device_id})
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
return v2_cfg
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
|
|
Packit Service |
a04d08 |
["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
"version": 2,
|
|
Packit Service |
a04d08 |
"ethernets": {args[1]: ethernet(*args) for args in physdevs},
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
return netcfg
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_skips_settle_if_all_present(
|
|
Packit Service |
a04d08 |
self, generic_networking_cls, wait_for_physdevs_netcfg,
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
networking = generic_networking_cls()
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "get_interfaces_by_mac"
|
|
Packit Service |
a04d08 |
) as m_get_interfaces_by_mac:
|
|
Packit Service |
a04d08 |
m_get_interfaces_by_mac.side_effect = iter(
|
|
Packit Service |
a04d08 |
[{"aa:bb:cc:dd:ee:ff": "eth0", "00:11:22:33:44:55": "ens3"}]
|
|
Packit Service |
a04d08 |
)
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "settle", autospec=True
|
|
Packit Service |
a04d08 |
) as m_settle:
|
|
Packit Service |
a04d08 |
networking.wait_for_physdevs(wait_for_physdevs_netcfg)
|
|
Packit Service |
a04d08 |
assert 0 == m_settle.call_count
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_calls_udev_settle_on_missing(
|
|
Packit Service |
a04d08 |
self, generic_networking_cls, wait_for_physdevs_netcfg,
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
networking = generic_networking_cls()
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "get_interfaces_by_mac"
|
|
Packit Service |
a04d08 |
) as m_get_interfaces_by_mac:
|
|
Packit Service |
a04d08 |
m_get_interfaces_by_mac.side_effect = iter(
|
|
Packit Service |
a04d08 |
[
|
|
Packit Service |
a04d08 |
{
|
|
Packit Service |
a04d08 |
"aa:bb:cc:dd:ee:ff": "eth0"
|
|
Packit Service |
a04d08 |
}, # first call ens3 is missing
|
|
Packit Service |
a04d08 |
{
|
|
Packit Service |
a04d08 |
"aa:bb:cc:dd:ee:ff": "eth0",
|
|
Packit Service |
a04d08 |
"00:11:22:33:44:55": "ens3",
|
|
Packit Service |
a04d08 |
}, # second call has both
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
)
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "settle", autospec=True
|
|
Packit Service |
a04d08 |
) as m_settle:
|
|
Packit Service |
a04d08 |
networking.wait_for_physdevs(wait_for_physdevs_netcfg)
|
|
Packit Service |
a04d08 |
m_settle.assert_called_with(exists="ens3")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@pytest.mark.parametrize(
|
|
Packit Service |
a04d08 |
"strict,expectation",
|
|
Packit Service |
a04d08 |
[(True, pytest.raises(RuntimeError)), (False, does_not_raise())],
|
|
Packit Service |
a04d08 |
)
|
|
Packit Service |
a04d08 |
def test_retrying_and_strict_behaviour(
|
|
Packit Service |
a04d08 |
self,
|
|
Packit Service |
a04d08 |
strict,
|
|
Packit Service |
a04d08 |
expectation,
|
|
Packit Service |
a04d08 |
generic_networking_cls,
|
|
Packit Service |
a04d08 |
wait_for_physdevs_netcfg,
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
networking = generic_networking_cls()
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "get_interfaces_by_mac"
|
|
Packit Service |
a04d08 |
) as m_get_interfaces_by_mac:
|
|
Packit Service |
a04d08 |
m_get_interfaces_by_mac.return_value = {}
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
with mock.patch.object(
|
|
Packit Service |
a04d08 |
networking, "settle", autospec=True
|
|
Packit Service |
a04d08 |
) as m_settle:
|
|
Packit Service |
a04d08 |
with expectation:
|
|
Packit Service |
a04d08 |
networking.wait_for_physdevs(
|
|
Packit Service |
a04d08 |
wait_for_physdevs_netcfg, strict=strict
|
|
Packit Service |
a04d08 |
)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
assert (
|
|
Packit Service |
a04d08 |
5 * len(wait_for_physdevs_netcfg["ethernets"])
|
|
Packit Service |
a04d08 |
== m_settle.call_count
|
|
Packit Service |
a04d08 |
)
|