webvirtcloud/interfaces/tests/test_forms.py

"""Host interface form: every value is validated before it goes into the interface XML."""

from unittest.mock import MagicMock, patch

import libvirt

from computes.models import Compute
from django.contrib.auth import get_user_model
from django.test import SimpleTestCase, TestCase
from django.urls import reverse
from lxml import etree, html
from vrtManager.interface import wvmInterfaces

from interfaces.forms import AddInterface

BASE = {
    "name": "br1", "itype": "bridge", "start_mode": "onboot", "netdev": "eth1",
    "ipv4_type": "static", "ipv4_addr": "192.0.2.10/24", "ipv4_gw": "192.0.2.1",
    "ipv6_type": "static", "stp": "on", "delay": 0,
}


# the host's network devices, as interfaces/views.py passes them
NETDEVS = ["eth0", "eth1", "enp129s0f0np0", "eth0.100", "bond-lan"]


def form(**values):
    return AddInterface(data={**BASE, **values}, netdevs=NETDEVS)


class Ipv6FormTestCase(SimpleTestCase):
    def test_valid_addresses_are_normalized(self):
        f = form(ipv6_addr="2001:DB8::10/64", ipv6_gw="2001:DB8::1")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["ipv6_addr"], "2001:db8::10/64")
        self.assertEqual(f.cleaned_data["ipv6_gw"], "2001:db8::1")

    def test_address_without_prefix_gets_one(self):
        f = form(ipv6_addr="2001:db8::10", ipv6_gw="")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["ipv6_addr"], "2001:db8::10/128")

    def test_values_that_only_start_like_an_address_are_rejected(self):
        for field, value in (
            ("ipv6_addr", "2001:db8::10/64'/><evil/>"),
            ("ipv6_gw", "2001:db8::1'/><evil/>"),
            ("ipv6_addr", "2001:db8::zz/64"),
            ("ipv6_addr", "192.0.2.10/24"),
        ):
            with self.subTest(field=field, value=value):
                f = form(**{"ipv6_addr": "2001:db8::10/64", "ipv6_gw": "", field: value})
                self.assertFalse(f.is_valid())
                self.assertIn(field, f.errors)

    def test_static_needs_an_address(self):
        f = form(ipv6_addr="", ipv6_gw="")
        self.assertFalse(f.is_valid())
        self.assertIn("ipv6_addr", f.errors)

    def test_dhcp_needs_no_address(self):
        self.assertTrue(form(ipv6_type="dhcp", ipv6_addr="", ipv6_gw="").is_valid())


class Ipv6XmlTestCase(SimpleTestCase):
    def xml(self, gw):
        conn = wvmInterfaces.__new__(wvmInterfaces)
        conn.define_iface = MagicMock()
        conn.get_iface = MagicMock()
        conn.get_iface.return_value.XMLDesc.return_value = "<interface type='ethernet' name='eth1'/>"
        conn.create_iface("br1", "bridge", "onboot", "eth1", "none", "", "", "static", "2001:db8::10/64", gw, "on", 0)
        return etree.fromstring(conn.define_iface.call_args[0][0])

    def test_static_address_with_gateway(self):
        proto = self.xml("2001:db8::1").find("protocol[@family='ipv6']")
        self.assertEqual(proto.find("ip").attrib, {"address": "2001:db8::10", "prefix": "64"})
        self.assertEqual(proto.find("route").get("gateway"), "2001:db8::1")

    def test_no_empty_route_without_gateway(self):
        self.assertIsNone(self.xml("").find("protocol[@family='ipv6']/route"))


class Ipv4FormTestCase(SimpleTestCase):
    def test_dhcp_and_none_need_no_address(self):
        for ipv4_type in ("dhcp", "none"):
            with self.subTest(ipv4_type=ipv4_type):
                f = form(ipv4_type=ipv4_type, ipv4_addr="", ipv4_gw="", ipv6_type="none")
                self.assertTrue(f.is_valid(), f.errors)

    def test_valid_address_and_gateway(self):
        f = form(ipv4_addr="192.0.2.10/24", ipv4_gw="192.0.2.1", ipv6_type="none")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/24")

    def test_address_without_prefix_gets_one(self):
        f = form(ipv4_addr="192.0.2.10", ipv4_gw="", ipv6_type="none")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/32")

    def test_invalid_values_are_rejected(self):
        for field, value in (
            ("ipv4_addr", "192.0.2.300/24"),
            ("ipv4_addr", "192.0.2.10/24/1"),
            ("ipv4_addr", "192.0.2.10/33"),
            ("ipv4_gw", "192.0.2.300"),
            ("ipv4_gw", "192.0.2.1/24"),
        ):
            with self.subTest(field=field, value=value):
                f = form(**{"ipv6_type": "none", field: value})
                self.assertFalse(f.is_valid())
                self.assertIn(field, f.errors)

    def test_static_needs_an_address(self):
        f = form(ipv4_addr="", ipv4_gw="", ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("ipv4_addr", f.errors)


class InterfaceNameTestCase(SimpleTestCase):
    def test_linux_interface_names_are_accepted(self):
        for name in ("br0", "bond-lan", "br_mgmt", "vlan.100", "BR0", "a" * 15):
            with self.subTest(name=name):
                f = form(name=name, ipv6_type="none")
                self.assertTrue(f.is_valid(), f.errors)
        for netdev in ("enp129s0f0np0", "eth0.100", "bond-lan"):
            with self.subTest(netdev=netdev):
                f = form(netdev=netdev, ipv6_type="none")
                self.assertTrue(f.is_valid(), f.errors)

    def test_other_names_are_rejected(self):
        for field in ("name", "netdev"):
            for value in ("br 1", "br/1", "br'1", "br1\nx", "a" * 16, "", ".", ".."):
                with self.subTest(field=field, value=value):
                    f = form(**{"ipv6_type": "none", field: value})
                    self.assertFalse(f.is_valid())
                    self.assertIn(field, f.errors)

    def test_surrounding_whitespace_is_stripped(self):
        f = form(name="br1\n", ipv6_type="none")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["name"], "br1")

    def test_a_new_interface_name_has_no_alias_colon(self):
        f = form(name="br0:1", ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("name", f.errors)


class Ipv4XmlTestCase(SimpleTestCase):
    def xml(self, addr, gw):
        conn = wvmInterfaces.__new__(wvmInterfaces)
        conn.define_iface = MagicMock()
        conn.get_iface = MagicMock()
        conn.get_iface.return_value.XMLDesc.return_value = "<interface type='ethernet' name='eth1'/>"
        conn.create_iface("br1", "bridge", "onboot", "eth1", "static", addr, gw, "none", "", "", "on", 0)
        return etree.fromstring(conn.define_iface.call_args[0][0])

    def test_static_address_with_gateway(self):
        proto = self.xml("192.0.2.10/24", "192.0.2.1").find("protocol[@family='ipv4']")
        self.assertEqual(proto.find("ip").attrib, {"address": "192.0.2.10", "prefix": "24"})
        self.assertEqual(proto.find("route").get("gateway"), "192.0.2.1")

    def test_no_empty_route_without_gateway(self):
        self.assertIsNone(self.xml("192.0.2.10/24", "").find("protocol[@family='ipv4']/route"))


class HostDeviceTestCase(SimpleTestCase):
    def test_a_bridge_member_must_be_a_host_device(self):
        for netdev in ("eth9", "eth0:1"):
            with self.subTest(netdev=netdev):
                f = form(netdev=netdev, ipv6_type="none")
                self.assertFalse(f.is_valid())
                self.assertIn("netdev", f.errors)

    def test_a_bridge_cannot_contain_itself(self):
        f = form(name="eth1", netdev="eth1", ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("netdev", f.errors)

    def test_an_ethernet_interface_configures_a_host_device(self):
        self.assertTrue(form(itype="ethernet", name="eth0", ipv6_type="none").is_valid())
        f = form(itype="ethernet", name="eth9", ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("name", f.errors)


def no_interface(name):
    err = libvirt.libvirtError("Interface not found: no interface with matching name '%s'" % name)
    err.err = (libvirt.VIR_ERR_NO_INTERFACE, libvirt.VIR_FROM_INTERFACE, str(err), 2, name, None, None, 0, 0)
    return err


class BridgeMemberXmlTestCase(SimpleTestCase):
    """A bridge member is declared as the host defines it: a bond keeps its
    mode and devices, a vlan its tag and device."""

    def member(self, host_xml):
        conn = wvmInterfaces.__new__(wvmInterfaces)
        conn.define_iface = MagicMock()
        created = MagicMock()

        def get_iface(name):
            if name == "br1":
                return created
            if host_xml is None:
                raise no_interface(name)
            iface = MagicMock()
            iface.XMLDesc.return_value = host_xml
            return iface

        conn.get_iface = MagicMock(side_effect=get_iface)
        conn.create_iface("br1", "bridge", "onboot", "bond0", "none", "", "", "none", "", "", "on", 0)
        self.assertTrue(created.create.called)
        members = etree.fromstring(conn.define_iface.call_args[0][0]).findall("bridge/interface")
        self.assertEqual(len(members), 1)
        return members[0]

    def test_a_bond_keeps_its_mode_and_devices(self):
        member = self.member(
            "<interface type='bond' name='bond0'><start mode='onboot'/><mtu size='9000'/>"
            "<protocol family='ipv4'><dhcp/></protocol>"
            "<bond mode='active-backup'><miimon freq='100'/>"
            "<interface type='ethernet' name='eth0'><mac address='52:54:00:00:00:01'/></interface>"
            "<interface type='ethernet' name='eth1'/></bond></interface>"
        )
        self.assertEqual((member.get("type"), member.get("name")), ("bond", "bond0"))
        self.assertEqual(member.find("bond").get("mode"), "active-backup")
        self.assertEqual([i.get("name") for i in member.findall("bond/interface")], ["eth0", "eth1"])
        self.assertEqual(member.find("bond/miimon").get("freq"), "100")
        # what only a top-level interface has: the bridge holds the addresses
        for tag in ("start", "mtu", "protocol"):
            self.assertIsNone(member.find(tag), tag)

    def test_a_vlan_keeps_its_tag_and_device(self):
        member = self.member(
            "<interface type='vlan' name='bond0'><start mode='onboot'/>"
            "<vlan tag='42'><interface name='eth0'/></vlan></interface>"
        )
        self.assertEqual(member.get("type"), "vlan")
        self.assertEqual(member.find("vlan").get("tag"), "42")
        self.assertEqual(member.find("vlan/interface").get("name"), "eth0")
        self.assertIsNone(member.find("start"))

    def test_a_device_the_host_does_not_define_is_ethernet(self):
        member = self.member(None)
        self.assertEqual(member.attrib, {"type": "ethernet", "name": "bond0"})

    def test_other_lookup_errors_are_not_hidden(self):
        conn = wvmInterfaces.__new__(wvmInterfaces)
        conn.define_iface = MagicMock()
        err = libvirt.libvirtError("cannot recv data")
        err.err = (libvirt.VIR_ERR_SYSTEM_ERROR, libvirt.VIR_FROM_RPC, str(err), 2, None, None, None, 0, 0)
        conn.get_iface = MagicMock(side_effect=err)
        with self.assertRaises(libvirt.libvirtError):
            conn.create_iface("br1", "bridge", "onboot", "bond0", "none", "", "", "none", "", "", "on", 0)
        self.assertFalse(conn.define_iface.called)


class BridgeSettingsTestCase(SimpleTestCase):
    def test_a_bridge_needs_stp_and_a_delay(self):
        for field in ("stp", "delay"):
            with self.subTest(field=field):
                f = form(**{"ipv6_type": "none", field: ""})
                self.assertFalse(f.is_valid())
                self.assertIn(field, f.errors)

    def test_the_delay_is_not_negative(self):
        f = form(delay=-1, ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("delay", f.errors)


class AddressNotationTestCase(SimpleTestCase):
    def test_ipv4_netmask_notation_becomes_a_prefix(self):
        f = form(ipv4_addr="192.0.2.10/255.255.255.0", ipv6_type="none")
        self.assertTrue(f.is_valid(), f.errors)
        self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/24")

    def test_ipv6_scope_ids_are_rejected(self):
        for field, value in (("ipv6_addr", "fe80::1%eth0/64"), ("ipv6_gw", "fe80::1%eth0")):
            with self.subTest(field=field):
                f = form(**{"ipv6_addr": "2001:db8::10/64", "ipv6_gw": "", field: value})
                self.assertFalse(f.is_valid())
                self.assertIn(field, f.errors)


class CreateViewTestCase(TestCase):
    """The page validates the device against the host's devices before it
    defines an interface."""

    def test_only_a_host_device_becomes_a_bridge_member(self):
        admin = get_user_model().objects.create_superuser("iface_admin", "a@example.com", "x")
        compute = Compute.objects.create(name="iface-c", hostname="127.0.0.1:1", login="root", password="", type=1)
        self.client.force_login(admin)
        for netdev, created in (("eth9", False), ("eth0", True)):
            with self.subTest(netdev=netdev), patch("interfaces.views.wvmInterfaces") as conn_cls:
                conn = conn_cls.return_value
                conn.get_ifaces.return_value = []
                conn.can_change_interfaces.return_value = True
                conn.get_net_devices.return_value = ["eth0", "eth1"]
                self.client.post(
                    reverse("interfaces", args=[compute.id]),
                    {**BASE, "netdev": netdev, "ipv6_type": "none", "create": ""},
                )
                self.assertEqual(conn.create_iface.called, created)

    def test_unreadable_host_devices_are_reported_not_blamed_on_the_input(self):
        admin = get_user_model().objects.create_superuser("iface_admin2", "b@example.com", "x")
        compute = Compute.objects.create(name="iface-c2", hostname="127.0.0.1:1", login="root", password="", type=1)
        self.client.force_login(admin)
        with patch("interfaces.views.wvmInterfaces") as conn_cls:
            conn = conn_cls.return_value
            conn.get_ifaces.return_value = []
            conn.can_change_interfaces.return_value = True
            conn.get_net_devices.side_effect = RuntimeError("nodedev driver not running")
            response = self.client.post(
                reverse("interfaces", args=[compute.id]),
                {**BASE, "netdev": "eth0", "ipv6_type": "none", "create": ""},
            )
        self.assertFalse(conn.create_iface.called)
        messages = [str(m) for m in response.context["messages"]]
        self.assertTrue(any("nodedev driver not running" in m for m in messages), messages)
        self.assertFalse(any("not on this host" in m for m in messages), messages)


class UnchangeableHostTestCase(TestCase):
    """A host whose libvirt cannot change interfaces (udev backend) gets no
    create form and no start/stop/delete, only a note; posts change nothing."""

    NOTE = "does not support changing network interfaces"

    def setUp(self):
        self.client.force_login(get_user_model().objects.create_superuser("iface_ro", "ro@example.com", "x"))
        self.compute = Compute.objects.create(name="iface-ro", hostname="127.0.0.1:1", login="root", password="", type=1)

    def interfaces_page(self, changeable, post=None):
        with patch("interfaces.views.wvmInterfaces") as conn_cls:
            conn = conn_cls.return_value
            conn.get_ifaces.return_value = []
            conn.can_change_interfaces.return_value = changeable
            conn.get_net_devices.return_value = ["eth0", "eth1"]
            url = reverse("interfaces", args=[self.compute.id])
            response = self.client.post(url, post) if post else self.client.get(url)
        return conn, response.content.decode()

    def test_the_interfaces_page_offers_create_only_where_it_works(self):
        for changeable in (True, False):
            with self.subTest(changeable=changeable):
                conn, page = self.interfaces_page(changeable)
                self.assertEqual("#AddInterface" in page, changeable)
                self.assertEqual(self.NOTE in page, not changeable)
                # the devices are only the form's choices
                self.assertEqual(conn.get_net_devices.called, changeable)

    def test_a_create_post_creates_nothing(self):
        conn, page = self.interfaces_page(False, {**BASE, "netdev": "eth1", "ipv6_type": "none", "create": ""})
        conn.create_iface.assert_not_called()
        self.assertIn(self.NOTE, page)

    def interface_page(self, changeable, state, post=None):
        with patch("interfaces.views.wvmInterface") as conn_cls:
            conn = conn_cls.return_value
            conn.can_change_interfaces.return_value = changeable
            conn.is_active.return_value = state
            conn.get_bridge_slave_ifaces.return_value = []
            url = reverse("interface", args=[self.compute.id, "br0"])
            response = self.client.post(url, post) if post else self.client.get(url)
        return conn, response

    def test_the_interface_page_shows_the_state_without_actions(self):
        for state, shown in ((1, "Active"), (0, "Inactive")):
            with self.subTest(state=state):
                _conn, response = self.interface_page(False, state)
                page = response.content.decode()
                self.assertNotIn('name="start"', page)
                self.assertNotIn('name="stop"', page)
                state_cell = html.fromstring(page).xpath("//dt[normalize-space()='State']/following-sibling::dd[1]")[0]
                self.assertEqual(state_cell.text_content().strip(), shown)
                self.assertIn(self.NOTE, page)
        _conn, response = self.interface_page(True, 0)
        self.assertIn('name="start"', response.content.decode())

    def test_a_host_that_could_not_be_read_keeps_the_page_as_it_was(self):
        with patch("interfaces.views.wvmInterface", side_effect=libvirt.libvirtError("unreachable")):
            page = self.client.get(reverse("interface", args=[self.compute.id, "br0"])).content.decode()
        self.assertNotIn(self.NOTE, page)
        self.assertIn("Interface start/stop/delete form", page)

    def test_start_stop_and_delete_posts_change_nothing(self):
        for action in ("start", "stop", "delete"):
            with self.subTest(action=action):
                conn, response = self.interface_page(False, 1, {action: ""})
                self.assertEqual(response.status_code, 200)
                for method in (conn.start_iface, conn.stop_iface, conn.delete_iface):
                    method.assert_not_called()


class EthernetDeviceTestCase(SimpleTestCase):
    def test_an_ethernet_interface_needs_no_bridge_member(self):
        self.assertTrue(form(itype="ethernet", name="eth0", netdev="", ipv6_type="none").is_valid())

    def test_a_bridge_needs_a_member(self):
        f = form(netdev="", ipv6_type="none")
        self.assertFalse(f.is_valid())
        self.assertIn("netdev", f.errors)