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

        )