diff options
Diffstat (limited to 'tests/unittests/test_net.py')
| -rw-r--r-- | tests/unittests/test_net.py | 4833 |
1 files changed, 2911 insertions, 1922 deletions
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py index b5c38c55..66a47b0f 100644 --- a/tests/unittests/test_net.py +++ b/tests/unittests/test_net.py @@ -1,20 +1,5 @@ # This file is part of cloud-init. See LICENSE file for license information. -from cloudinit import net -from cloudinit import distros -from cloudinit.net import cmdline -from cloudinit.net import ( - eni, interface_has_own_mac, natural_sort_key, netplan, network_state, - renderers, sysconfig, networkd) -from cloudinit.sources.helpers import openstack -from cloudinit import temp_utils -from cloudinit import subp -from cloudinit import util -from cloudinit import safeyaml as yaml - -from tests.unittests.helpers import ( - CiTestCase, FilesystemMockingTestCase, dir2dict, mock, populate_dir) - import base64 import copy import gzip @@ -23,9 +8,32 @@ import json import os import re import textwrap -from yaml.serializer import Serializer import pytest +from yaml.serializer import Serializer + +from cloudinit import distros, net +from cloudinit import safeyaml as yaml +from cloudinit import subp, temp_utils, util +from cloudinit.net import ( + cmdline, + eni, + interface_has_own_mac, + natural_sort_key, + netplan, + network_state, + networkd, + renderers, + sysconfig, +) +from cloudinit.sources.helpers import openstack +from tests.unittests.helpers import ( + CiTestCase, + FilesystemMockingTestCase, + dir2dict, + mock, + populate_dir, +) DHCP_CONTENT_1 = """ DEVICE='eth0' @@ -48,15 +56,19 @@ DOMAINSEARCH='foo.com' """ DHCP_EXPECTED_1 = { - 'name': 'eth0', - 'type': 'physical', - 'subnets': [{'broadcast': '192.168.122.255', - 'control': 'manual', - 'gateway': '192.168.122.1', - 'dns_search': ['foo.com'], - 'type': 'dhcp', - 'netmask': '255.255.255.0', - 'dns_nameservers': ['192.168.122.1']}], + "name": "eth0", + "type": "physical", + "subnets": [ + { + "broadcast": "192.168.122.255", + "control": "manual", + "gateway": "192.168.122.1", + "dns_search": ["foo.com"], + "type": "dhcp", + "netmask": "255.255.255.0", + "dns_nameservers": ["192.168.122.1"], + } + ], } DHCP6_CONTENT_1 = """ @@ -73,12 +85,17 @@ DNSDOMAIN= """ DHCP6_EXPECTED_1 = { - 'name': 'eno1', - 'type': 'physical', - 'subnets': [{'control': 'manual', - 'dns_nameservers': ['2001:67c:1562:8010::2:1'], - 'netmask': '64', - 'type': 'dhcp6'}]} + "name": "eno1", + "type": "physical", + "subnets": [ + { + "control": "manual", + "dns_nameservers": ["2001:67c:1562:8010::2:1"], + "netmask": "64", + "type": "dhcp6", + } + ], +} STATIC_CONTENT_1 = """ @@ -97,14 +114,20 @@ DOMAINSEARCH='foo.com' """ STATIC_EXPECTED_1 = { - 'name': 'eth1', - 'type': 'physical', - 'subnets': [{'broadcast': '10.0.0.255', 'control': 'manual', - 'gateway': '10.0.0.1', - 'dns_search': ['foo.com'], 'type': 'static', - 'netmask': '255.255.255.0', - 'dns_nameservers': ['10.0.1.1'], - 'address': '10.0.0.2'}], + "name": "eth1", + "type": "physical", + "subnets": [ + { + "broadcast": "10.0.0.255", + "control": "manual", + "gateway": "10.0.0.1", + "dns_search": ["foo.com"], + "type": "static", + "netmask": "255.255.255.0", + "dns_nameservers": ["10.0.1.1"], + "address": "10.0.0.2", + } + ], } V1_NAMESERVER_ALIAS = """ @@ -471,34 +494,42 @@ ethernets: # Examples (and expected outputs for various renderers). OS_SAMPLES = [ { - 'in_data': { + "in_data": { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", - "type": "ipv4", "netmask": "255.255.252.0", - "link": "tap1a81968a-79", - "routes": [{ - "netmask": "0.0.0.0", - "network": "0.0.0.0", - "gateway": "172.19.3.254", - }], - "ip_address": "172.19.1.34", "id": "network0" - }], + "networks": [ + { + "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", + "type": "ipv4", + "netmask": "255.255.252.0", + "link": "tap1a81968a-79", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "172.19.3.254", + } + ], + "ip_address": "172.19.1.34", + "id": "network0", + } + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], }, - 'in_macs': { - 'fa:16:3e:ed:9a:59': 'eth0', + "in_macs": { + "fa:16:3e:ed:9a:59": "eth0", }, - 'out_sysconfig_opensuse': [ - ('etc/sysconfig/network/ifcfg-eth0', - """ + "out_sysconfig_opensuse": [ + ( + "etc/sysconfig/network/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=static @@ -506,26 +537,39 @@ IPADDR=172.19.1.34 LLADDR=fa:16:3e:ed:9a:59 NETMASK=255.255.252.0 STARTMODE=auto -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/85-persistent-net-cloud-init.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))], - 'out_sysconfig_rhel': [ - ('etc/sysconfig/network-scripts/ifcfg-eth0', - """ +""".lstrip(), + ), + ( + "etc/udev/rules.d/85-persistent-net-cloud-init.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], + "out_sysconfig_rhel": [ + ( + "etc/sysconfig/network-scripts/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=none @@ -539,60 +583,82 @@ NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet USERCTL=no -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/70-persistent-net.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))] - +""".lstrip(), + ), + ( + "etc/udev/rules.d/70-persistent-net.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], }, { - 'in_data': { + "in_data": { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "public-ipv4", - "type": "ipv4", "netmask": "255.255.252.0", - "link": "tap1a81968a-79", - "routes": [{ - "netmask": "0.0.0.0", - "network": "0.0.0.0", - "gateway": "172.19.3.254", - }], - "ip_address": "172.19.1.34", "id": "network0" - }, { - "network_id": "private-ipv4", - "type": "ipv4", "netmask": "255.255.255.0", - "link": "tap1a81968a-79", - "routes": [], - "ip_address": "10.0.0.10", "id": "network1" - }], + "networks": [ + { + "network_id": "public-ipv4", + "type": "ipv4", + "netmask": "255.255.252.0", + "link": "tap1a81968a-79", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "172.19.3.254", + } + ], + "ip_address": "172.19.1.34", + "id": "network0", + }, + { + "network_id": "private-ipv4", + "type": "ipv4", + "netmask": "255.255.255.0", + "link": "tap1a81968a-79", + "routes": [], + "ip_address": "10.0.0.10", + "id": "network1", + }, + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], }, - 'in_macs': { - 'fa:16:3e:ed:9a:59': 'eth0', + "in_macs": { + "fa:16:3e:ed:9a:59": "eth0", }, - 'out_sysconfig_opensuse': [ - ('etc/sysconfig/network/ifcfg-eth0', - """ + "out_sysconfig_opensuse": [ + ( + "etc/sysconfig/network/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=static @@ -602,26 +668,39 @@ LLADDR=fa:16:3e:ed:9a:59 NETMASK=255.255.252.0 NETMASK1=255.255.255.0 STARTMODE=auto -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/85-persistent-net-cloud-init.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))], - 'out_sysconfig_rhel': [ - ('etc/sysconfig/network-scripts/ifcfg-eth0', - """ +""".lstrip(), + ), + ( + "etc/udev/rules.d/85-persistent-net-cloud-init.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], + "out_sysconfig_rhel": [ + ( + "etc/sysconfig/network-scripts/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=none @@ -637,80 +716,106 @@ NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet USERCTL=no -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/70-persistent-net.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))] - +""".lstrip(), + ), + ( + "etc/udev/rules.d/70-persistent-net.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], }, { - 'in_data': { + "in_data": { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "public-ipv4", - "type": "ipv4", "netmask": "255.255.252.0", - "link": "tap1a81968a-79", - "routes": [{ - "netmask": "0.0.0.0", - "network": "0.0.0.0", - "gateway": "172.19.3.254", - }], - "ip_address": "172.19.1.34", "id": "network0" - }, { - "network_id": "public-ipv6-a", - "type": "ipv6", "netmask": "", - "link": "tap1a81968a-79", - "routes": [ - { - "gateway": "2001:DB8::1", - "netmask": "::", - "network": "::" - } - ], - "ip_address": "2001:DB8::10", "id": "network1" - }, { - "network_id": "public-ipv6-b", - "type": "ipv6", "netmask": "64", - "link": "tap1a81968a-79", - "routes": [ - ], - "ip_address": "2001:DB9::10", "id": "network2" - }, { - "network_id": "public-ipv6-c", - "type": "ipv6", "netmask": "64", - "link": "tap1a81968a-79", - "routes": [ - ], - "ip_address": "2001:DB10::10", "id": "network3" - }], + "networks": [ + { + "network_id": "public-ipv4", + "type": "ipv4", + "netmask": "255.255.252.0", + "link": "tap1a81968a-79", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "172.19.3.254", + } + ], + "ip_address": "172.19.1.34", + "id": "network0", + }, + { + "network_id": "public-ipv6-a", + "type": "ipv6", + "netmask": "", + "link": "tap1a81968a-79", + "routes": [ + { + "gateway": "2001:DB8::1", + "netmask": "::", + "network": "::", + } + ], + "ip_address": "2001:DB8::10", + "id": "network1", + }, + { + "network_id": "public-ipv6-b", + "type": "ipv6", + "netmask": "64", + "link": "tap1a81968a-79", + "routes": [], + "ip_address": "2001:DB9::10", + "id": "network2", + }, + { + "network_id": "public-ipv6-c", + "type": "ipv6", + "netmask": "64", + "link": "tap1a81968a-79", + "routes": [], + "ip_address": "2001:DB10::10", + "id": "network3", + }, + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], }, - 'in_macs': { - 'fa:16:3e:ed:9a:59': 'eth0', + "in_macs": { + "fa:16:3e:ed:9a:59": "eth0", }, - 'out_sysconfig_opensuse': [ - ('etc/sysconfig/network/ifcfg-eth0', - """ + "out_sysconfig_opensuse": [ + ( + "etc/sysconfig/network/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=static @@ -721,26 +826,39 @@ IPADDR6_2=2001:DB10::10/64 LLADDR=fa:16:3e:ed:9a:59 NETMASK=255.255.252.0 STARTMODE=auto -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/85-persistent-net-cloud-init.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))], - 'out_sysconfig_rhel': [ - ('etc/sysconfig/network-scripts/ifcfg-eth0', - """ +""".lstrip(), + ), + ( + "etc/udev/rules.d/85-persistent-net-cloud-init.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], + "out_sysconfig_rhel": [ + ( + "etc/sysconfig/network-scripts/ifcfg-eth0", + """ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=none @@ -760,24 +878,36 @@ NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet USERCTL=no -""".lstrip()), - ('etc/resolv.conf', - """ +""".lstrip(), + ), + ( + "etc/resolv.conf", + """ ; Created by cloud-init on instance boot automatically, do not edit. ; nameserver 172.19.0.12 -""".lstrip()), - ('etc/NetworkManager/conf.d/99-cloud-init.conf', - """ +""".lstrip(), + ), + ( + "etc/NetworkManager/conf.d/99-cloud-init.conf", + """ # Created by cloud-init on instance boot automatically, do not edit. # [main] dns = none -""".lstrip()), - ('etc/udev/rules.d/70-persistent-net.rules', - "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', - 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))] - } +""".lstrip(), + ), + ( + "etc/udev/rules.d/70-persistent-net.rules", + "".join( + [ + 'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ', + 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n', + ] + ), + ), + ], + }, ] EXAMPLE_ENI = """ @@ -820,8 +950,9 @@ iface eth1 inet static """.lstrip() NETWORK_CONFIGS = { - 'small': { - 'expected_networkd_eth99': textwrap.dedent("""\ + "small": { + "expected_networkd_eth99": textwrap.dedent( + """\ [Match] Name=eth99 MACAddress=c0:d6:9f:2c:e8:80 @@ -837,8 +968,10 @@ NETWORK_CONFIGS = { Gateway=65.61.151.37 Destination=0.0.0.0/0 Metric=10000 - """).rstrip(' '), - 'expected_networkd_eth1': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_networkd_eth1": textwrap.dedent( + """\ [Match] Name=eth1 MACAddress=cf:d6:af:48:e8:80 @@ -846,8 +979,10 @@ NETWORK_CONFIGS = { DHCP=no Domains=wark.maas DNS=1.2.3.4 5.6.7.8 - """).rstrip(' '), - 'expected_eni': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback dns-nameservers 1.2.3.4 5.6.7.8 @@ -865,8 +1000,10 @@ NETWORK_CONFIGS = { dns-search barley.maas sach.maas post-up route add default gw 65.61.151.37 metric 10000 || true pre-down route del default gw 65.61.151.37 metric 10000 || true - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: @@ -892,29 +1029,37 @@ NETWORK_CONFIGS = { to: 0.0.0.0/0 via: 65.61.151.37 set-name: eth99 - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-eth1': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=cf:d6:af:48:e8:80 - STARTMODE=auto"""), - 'ifcfg-eth99': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-eth99": textwrap.dedent( + """\ BOOTPROTO=dhcp4 LLADDR=c0:d6:9f:2c:e8:80 IPADDR=192.168.21.3 NETMASK=255.255.255.0 - STARTMODE=auto"""), + STARTMODE=auto""" + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-eth1': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth1 HWADDR=cf:d6:af:48:e8:80 NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth99': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth99": textwrap.dedent( + """\ BOOTPROTO=dhcp DEFROUTE=yes DEVICE=eth99 @@ -930,9 +1075,11 @@ NETWORK_CONFIGS = { NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), + USERCTL=no""" + ), }, - 'yaml': textwrap.dedent(""" + "yaml": textwrap.dedent( + """ version: 1 config: # Physical interfaces. @@ -961,16 +1108,20 @@ NETWORK_CONFIGS = { - 5.6.7.8 search: - wark.maas - """), + """ + ), }, - 'v4_and_v6': { - 'expected_networkd': textwrap.dedent("""\ + "v4_and_v6": { + "expected_networkd": textwrap.dedent( + """\ [Match] Name=iface0 [Network] DHCP=yes - """).rstrip(' '), - 'expected_eni': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback @@ -979,22 +1130,28 @@ NETWORK_CONFIGS = { # control-alias iface0 iface iface0 inet6 dhcp - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: dhcp4: true dhcp6: true - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp DHCLIENT6_MODE=managed - STARTMODE=auto""") + STARTMODE=auto""" + ) }, - 'yaml': textwrap.dedent("""\ + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1002,10 +1159,12 @@ NETWORK_CONFIGS = { subnets: - {'type': 'dhcp4'} - {'type': 'dhcp6'} - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'v4_and_v6_static': { - 'expected_networkd': textwrap.dedent("""\ + "v4_and_v6_static": { + "expected_networkd": textwrap.dedent( + """\ [Match] Name=iface0 [Link] @@ -1015,8 +1174,10 @@ NETWORK_CONFIGS = { [Address] Address=192.168.14.2/24 Address=2001:1::1/64 - """).rstrip(' '), - 'expected_eni': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback @@ -1029,8 +1190,10 @@ NETWORK_CONFIGS = { iface iface0 inet6 static address 2001:1::1/64 mtu 1500 - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: @@ -1040,8 +1203,10 @@ NETWORK_CONFIGS = { - 2001:1::1/64 ipv6-mtu: 1500 mtu: 9000 - """).rstrip(' '), - 'yaml': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1054,19 +1219,23 @@ NETWORK_CONFIGS = { - type: static address: 2001:1::1/64 mtu: 1500 - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=static IPADDR=192.168.14.2 IPADDR6=2001:1::1/64 NETMASK=255.255.255.0 STARTMODE=auto MTU=9000 - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 IPADDR=192.168.14.2 @@ -1081,17 +1250,21 @@ NETWORK_CONFIGS = { USERCTL=no MTU=9000 IPV6_MTU=1500 - """), + """ + ), }, }, - 'v6_and_v4': { - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + "v6_and_v4": { + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp DHCLIENT6_MODE=managed - STARTMODE=auto""") + STARTMODE=auto""" + ) }, - 'yaml': textwrap.dedent("""\ + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1099,46 +1272,58 @@ NETWORK_CONFIGS = { subnets: - type: dhcp6 - type: dhcp4 - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'dhcpv6_only': { - 'expected_networkd': textwrap.dedent("""\ + "dhcpv6_only": { + "expected_networkd": textwrap.dedent( + """\ [Match] Name=iface0 [Network] DHCP=ipv6 - """).rstrip(' '), - 'expected_eni': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 dhcp - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: dhcp6: true - """).rstrip(' '), - 'yaml': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' name: 'iface0' subnets: - {'type': 'dhcp6'} - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=managed STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 DHCPV6C=yes @@ -1148,27 +1333,33 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'dhcpv6_accept_ra': { - 'expected_eni': textwrap.dedent("""\ + "dhcpv6_accept_ra": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 dhcp accept_ra 1 - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: accept-ra: true dhcp6: true - """).rstrip(' '), - 'yaml_v1': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml_v1": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1176,23 +1367,29 @@ NETWORK_CONFIGS = { subnets: - {'type': 'dhcp6'} accept-ra: true - """).rstrip(' '), - 'yaml_v2': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml_v2": textwrap.dedent( + """\ version: 2 ethernets: iface0: dhcp6: true accept-ra: true - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=managed STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 DHCPV6C=yes @@ -1203,34 +1400,42 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, - 'expected_networkd': textwrap.dedent("""\ + "expected_networkd": textwrap.dedent( + """\ [Match] Name=iface0 [Network] DHCP=ipv6 IPv6AcceptRA=True - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'dhcpv6_reject_ra': { - 'expected_eni': textwrap.dedent("""\ + "dhcpv6_reject_ra": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 dhcp accept_ra 0 - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: accept-ra: false dhcp6: true - """).rstrip(' '), - 'yaml_v1': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml_v1": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1238,23 +1443,29 @@ NETWORK_CONFIGS = { subnets: - {'type': 'dhcp6'} accept-ra: false - """).rstrip(' '), - 'yaml_v2': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml_v2": textwrap.dedent( + """\ version: 2 ethernets: iface0: dhcp6: true accept-ra: false - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=managed STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 DHCPV6C=yes @@ -1265,49 +1476,61 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, - 'expected_networkd': textwrap.dedent("""\ + "expected_networkd": textwrap.dedent( + """\ [Match] Name=iface0 [Network] DHCP=ipv6 IPv6AcceptRA=False - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'ipv6_slaac': { - 'expected_eni': textwrap.dedent("""\ + "ipv6_slaac": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 auto dhcp 0 - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: dhcp6: true - """).rstrip(' '), - 'yaml': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' name: 'iface0' subnets: - {'type': 'ipv6_slaac'} - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=info STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 IPV6_AUTOCONF=yes @@ -1317,11 +1540,13 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'static6': { - 'yaml': textwrap.dedent("""\ + "static6": { + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1330,9 +1555,11 @@ NETWORK_CONFIGS = { subnets: - type: 'static6' address: 2001:1::1/64 - """).rstrip(' '), - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 IPV6ADDR=2001:1::1/64 @@ -1344,42 +1571,52 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'dhcpv6_stateless': { - 'expected_eni': textwrap.dedent("""\ + "dhcpv6_stateless": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 auto dhcp 1 - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: dhcp6: true - """).rstrip(' '), - 'yaml': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' name: 'iface0' subnets: - {'type': 'ipv6_dhcpv6-stateless'} - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=info STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=iface0 DHCPV6C=yes @@ -1391,26 +1628,32 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'dhcpv6_stateful': { - 'expected_eni': textwrap.dedent("""\ + "dhcpv6_stateful": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet6 dhcp - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: iface0: accept-ra: true dhcp6: true - """).rstrip(' '), - 'yaml': textwrap.dedent("""\ + """ + ).rstrip(" "), + "yaml": textwrap.dedent( + """\ version: 1 config: - type: 'physical' @@ -1418,16 +1661,20 @@ NETWORK_CONFIGS = { subnets: - {'type': 'ipv6_dhcpv6-stateful'} accept-ra: true - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp6 DHCLIENT6_MODE=managed STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=iface0 DHCPV6C=yes @@ -1439,75 +1686,93 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'wakeonlan_disabled': { - 'expected_eni': textwrap.dedent("""\ + "wakeonlan_disabled": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet dhcp - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: ethernets: iface0: dhcp4: true wakeonlan: false version: 2 - """), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp4 STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=iface0 NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, - 'yaml_v2': textwrap.dedent("""\ + "yaml_v2": textwrap.dedent( + """\ version: 2 ethernets: iface0: dhcp4: true wakeonlan: false - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'wakeonlan_enabled': { - 'expected_eni': textwrap.dedent("""\ + "wakeonlan_enabled": { + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback auto iface0 iface iface0 inet dhcp ethernet-wol g - """).rstrip(' '), - 'expected_netplan': textwrap.dedent(""" + """ + ).rstrip(" "), + "expected_netplan": textwrap.dedent( + """ network: ethernets: iface0: dhcp4: true wakeonlan: true version: 2 - """), - 'expected_sysconfig_opensuse': { - 'ifcfg-iface0': textwrap.dedent("""\ + """ + ), + "expected_sysconfig_opensuse": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp4 ETHTOOL_OPTS="wol g" STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-iface0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-iface0": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=iface0 ETHTOOL_OPTS="wol g" @@ -1515,18 +1780,21 @@ NETWORK_CONFIGS = { ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, - 'yaml_v2': textwrap.dedent("""\ + "yaml_v2": textwrap.dedent( + """\ version: 2 ethernets: iface0: dhcp4: true wakeonlan: true - """).rstrip(' '), + """ + ).rstrip(" "), }, - 'all': { - 'expected_eni': ("""\ + "all": { + "expected_eni": """\ auto lo iface lo inet loopback dns-nameservers 8.8.8.8 4.4.4.4 8.8.4.4 @@ -1616,8 +1884,9 @@ iface eth0.101 inet static post-up route add -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true -"""), - 'expected_netplan': textwrap.dedent(""" +""", + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: @@ -1713,25 +1982,31 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true - barley.maas - sacchromyces.maas - brettanomyces.maas - """).rstrip(' '), - 'expected_sysconfig_opensuse': { - 'ifcfg-bond0': textwrap.dedent("""\ + """ + ).rstrip(" "), + "expected_sysconfig_opensuse": { + "ifcfg-bond0": textwrap.dedent( + """\ BONDING_MASTER=yes BONDING_MODULE_OPTS="mode=active-backup """ - """xmit_hash_policy=layer3+4 """ - """miimon=100" + """xmit_hash_policy=layer3+4 """ + """miimon=100" BONDING_SLAVE_0=eth1 BONDING_SLAVE_1=eth2 BOOTPROTO=dhcp6 DHCLIENT6_MODE=managed LLADDR=aa:bb:cc:dd:ee:ff - STARTMODE=auto"""), - 'ifcfg-bond0.200': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-bond0.200": textwrap.dedent( + """\ BOOTPROTO=dhcp4 ETHERDEVICE=bond0 STARTMODE=auto - VLAN_ID=200"""), - 'ifcfg-br0': textwrap.dedent("""\ + VLAN_ID=200""" + ), + "ifcfg-br0": textwrap.dedent( + """\ BRIDGE_AGEINGTIME=250 BOOTPROTO=static IPADDR=192.168.14.2 @@ -1741,12 +2016,16 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true BRIDGE_PRIORITY=22 BRIDGE_PORTS='eth3 eth4' STARTMODE=auto - BRIDGE_STP=off"""), - 'ifcfg-eth0': textwrap.dedent("""\ + BRIDGE_STP=off""" + ), + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=c0:d6:9f:2c:e8:80 - STARTMODE=auto"""), - 'ifcfg-eth0.101': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-eth0.101": textwrap.dedent( + """\ BOOTPROTO=static IPADDR=192.168.0.2 IPADDR1=192.168.2.10 @@ -1755,44 +2034,58 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NETMASK1=255.255.255.0 ETHERDEVICE=eth0 STARTMODE=auto - VLAN_ID=101"""), - 'ifcfg-eth1': textwrap.dedent("""\ + VLAN_ID=101""" + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=none LLADDR=aa:d6:9f:2c:e8:80 - STARTMODE=hotplug"""), - 'ifcfg-eth2': textwrap.dedent("""\ + STARTMODE=hotplug""" + ), + "ifcfg-eth2": textwrap.dedent( + """\ BOOTPROTO=none LLADDR=c0:bb:9f:2c:e8:80 - STARTMODE=hotplug"""), - 'ifcfg-eth3': textwrap.dedent("""\ + STARTMODE=hotplug""" + ), + "ifcfg-eth3": textwrap.dedent( + """\ BOOTPROTO=static BRIDGE=yes LLADDR=66:bb:9f:2c:e8:80 - STARTMODE=auto"""), - 'ifcfg-eth4': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-eth4": textwrap.dedent( + """\ BOOTPROTO=static BRIDGE=yes LLADDR=98:bb:9f:2c:e8:80 - STARTMODE=auto"""), - 'ifcfg-eth5': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-eth5": textwrap.dedent( + """\ BOOTPROTO=dhcp LLADDR=98:bb:9f:2c:e8:8a - STARTMODE=manual"""), - 'ifcfg-ib0': textwrap.dedent("""\ + STARTMODE=manual""" + ), + "ifcfg-ib0": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=a0:00:02:20:fe:80:00:00:00:00:00:00:ec:0d:9a:03:00:15:e2:c1 IPADDR=192.168.200.7 MTU=9000 NETMASK=255.255.255.0 STARTMODE=auto - TYPE=InfiniBand"""), + TYPE=InfiniBand""" + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-bond0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-bond0": textwrap.dedent( + """\ BONDING_MASTER=yes BONDING_OPTS="mode=active-backup """ - """xmit_hash_policy=layer3+4 """ - """miimon=100" + """xmit_hash_policy=layer3+4 """ + """miimon=100" BONDING_SLAVE0=eth1 BONDING_SLAVE1=eth2 BOOTPROTO=none @@ -1803,8 +2096,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NM_CONTROLLED=no ONBOOT=yes TYPE=Bond - USERCTL=no"""), - 'ifcfg-bond0.200': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-bond0.200": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=bond0.200 DHCLIENT_SET_DEFAULT_ROUTE=no @@ -1812,8 +2107,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true ONBOOT=yes PHYSDEV=bond0 USERCTL=no - VLAN=yes"""), - 'ifcfg-br0': textwrap.dedent("""\ + VLAN=yes""" + ), + "ifcfg-br0": textwrap.dedent( + """\ AGEING=250 BOOTPROTO=none DEFROUTE=yes @@ -1831,16 +2128,20 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true PRIO=22 STP=no TYPE=Bridge - USERCTL=no"""), - 'ifcfg-eth0': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth0 HWADDR=c0:d6:9f:2c:e8:80 NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth0.101': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth0.101": textwrap.dedent( + """\ BOOTPROTO=none DEFROUTE=yes DEVICE=eth0.101 @@ -1857,8 +2158,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true ONBOOT=yes PHYSDEV=eth0 USERCTL=no - VLAN=yes"""), - 'ifcfg-eth1': textwrap.dedent("""\ + VLAN=yes""" + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth1 HWADDR=aa:d6:9f:2c:e8:80 @@ -1867,8 +2170,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true ONBOOT=yes SLAVE=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth2': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth2": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth2 HWADDR=c0:bb:9f:2c:e8:80 @@ -1877,8 +2182,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true ONBOOT=yes SLAVE=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth3': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth3": textwrap.dedent( + """\ BOOTPROTO=none BRIDGE=br0 DEVICE=eth3 @@ -1886,8 +2193,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth4': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth4": textwrap.dedent( + """\ BOOTPROTO=none BRIDGE=br0 DEVICE=eth4 @@ -1895,8 +2204,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-eth5': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-eth5": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=eth5 DHCLIENT_SET_DEFAULT_ROUTE=no @@ -1904,8 +2215,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NM_CONTROLLED=no ONBOOT=no TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-ib0': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-ib0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=ib0 HWADDR=a0:00:02:20:fe:80:00:00:00:00:00:00:ec:0d:9a:03:00:15:e2:c1 @@ -1915,9 +2228,11 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true NM_CONTROLLED=no ONBOOT=yes TYPE=InfiniBand - USERCTL=no"""), + USERCTL=no""" + ), }, - 'yaml': textwrap.dedent(""" + "yaml": textwrap.dedent( + """ version: 1 config: # Physical interfaces. @@ -2060,10 +2375,12 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true destination: 10.0.0.0/8 gateway: 11.0.0.1 metric: 3 - """).lstrip(), + """ + ).lstrip(), }, - 'bond': { - 'yaml': textwrap.dedent(""" + "bond": { + "yaml": textwrap.dedent( + """ version: 1 config: - type: physical @@ -2109,8 +2426,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true network: 3001:67c:1 netmask: ffff:ffff:0 metric: 10000 - """), - 'expected_netplan': textwrap.dedent(""" + """ + ), + "expected_netplan": textwrap.dedent( + """ network: version: 2 ethernets: @@ -2152,8 +2471,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true - metric: 10000 to: 3001:67c:1/32 via: 3001:67c:1562:1 - """), - 'expected_eni': textwrap.dedent("""\ + """ + ), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback @@ -2215,8 +2536,10 @@ iface bond0 inet6 static || true pre-down route del -A inet6 3001:67c:1/32 gw 3001:67c:1562:1 metric 10000 \ || true - """), - 'yaml-v2': textwrap.dedent(""" + """ + ), + "yaml-v2": textwrap.dedent( + """ version: 2 ethernets: eth0: @@ -2256,8 +2579,10 @@ iface bond0 inet6 static - metric: 10000 to: 3001:67c:1562:8007::1/64 via: 3001:67c:1562:8007::aac:40b2 - """), - 'expected_netplan-v2': textwrap.dedent(""" + """ + ), + "expected_netplan-v2": textwrap.dedent( + """ network: bonds: bond0: @@ -2298,17 +2623,18 @@ iface bond0 inet6 static macaddress: aa:bb:cc:dd:e8:01 set-name: vf0 version: 2 - """), - - 'expected_sysconfig_opensuse': { - 'ifcfg-bond0': textwrap.dedent("""\ + """ + ), + "expected_sysconfig_opensuse": { + "ifcfg-bond0": textwrap.dedent( + """\ BONDING_MASTER=yes BONDING_MODULE_OPTS="mode=active-backup xmit_hash_policy=layer3+4 """ - """miimon=100 num_grat_arp=5 """ - """downdelay=10 updelay=20 """ - """fail_over_mac=active """ - """primary=bond0s0 """ - """primary_reselect=always" + """miimon=100 num_grat_arp=5 """ + """downdelay=10 updelay=20 """ + """fail_over_mac=active """ + """primary=bond0s0 """ + """primary_reselect=always" BONDING_SLAVE_0=bond0s0 BONDING_SLAVE_1=bond0s1 BOOTPROTO=static @@ -2320,27 +2646,33 @@ iface bond0 inet6 static NETMASK=255.255.255.0 NETMASK1=255.255.255.0 STARTMODE=auto - """), - 'ifcfg-bond0s0': textwrap.dedent("""\ + """ + ), + "ifcfg-bond0s0": textwrap.dedent( + """\ BOOTPROTO=none LLADDR=aa:bb:cc:dd:e8:00 STARTMODE=hotplug - """), - 'ifcfg-bond0s1': textwrap.dedent("""\ + """ + ), + "ifcfg-bond0s1": textwrap.dedent( + """\ BOOTPROTO=none LLADDR=aa:bb:cc:dd:e8:01 STARTMODE=hotplug - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-bond0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-bond0": textwrap.dedent( + """\ BONDING_MASTER=yes BONDING_OPTS="mode=active-backup xmit_hash_policy=layer3+4 """ - """miimon=100 num_grat_arp=5 """ - """downdelay=10 updelay=20 """ - """fail_over_mac=active """ - """primary=bond0s0 """ - """primary_reselect=always" + """miimon=100 num_grat_arp=5 """ + """downdelay=10 updelay=20 """ + """fail_over_mac=active """ + """primary=bond0s0 """ + """primary_reselect=always" BONDING_SLAVE0=bond0s0 BONDING_SLAVE1=bond0s1 BOOTPROTO=none @@ -2361,8 +2693,10 @@ iface bond0 inet6 static ONBOOT=yes TYPE=Bond USERCTL=no - """), - 'ifcfg-bond0s0': textwrap.dedent("""\ + """ + ), + "ifcfg-bond0s0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=bond0s0 HWADDR=aa:bb:cc:dd:e8:00 @@ -2372,19 +2706,25 @@ iface bond0 inet6 static SLAVE=yes TYPE=Ethernet USERCTL=no - """), - 'route6-bond0': textwrap.dedent("""\ + """ + ), + "route6-bond0": textwrap.dedent( + """\ # Created by cloud-init on instance boot automatically, do not edit. # 2001:67c:1/ffff:ffff:0 via 2001:67c:1562:1 dev bond0 3001:67c:1/ffff:ffff:0 via 3001:67c:1562:1 metric 10000 dev bond0 - """), - 'route-bond0': textwrap.dedent("""\ + """ + ), + "route-bond0": textwrap.dedent( + """\ ADDRESS0=10.1.3.0 GATEWAY0=192.168.0.3 NETMASK0=255.255.255.0 - """), - 'ifcfg-bond0s1': textwrap.dedent("""\ + """ + ), + "ifcfg-bond0s1": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=bond0s1 HWADDR=aa:bb:cc:dd:e8:01 @@ -2394,11 +2734,13 @@ iface bond0 inet6 static SLAVE=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'vlan': { - 'yaml': textwrap.dedent(""" + "vlan": { + "yaml": textwrap.dedent( + """ version: 1 config: - type: physical @@ -2421,14 +2763,18 @@ iface bond0 inet6 static - gateway: 2001:1::1 netmask: '::' network: '::' - """), - 'expected_sysconfig_opensuse': { + """ + ), + "expected_sysconfig_opensuse": { # TODO RJS: unknown proper BOOTPROTO setting ask Marius - 'ifcfg-en0': textwrap.dedent("""\ + "ifcfg-en0": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=aa:bb:cc:dd:e8:00 - STARTMODE=auto"""), - 'ifcfg-en0.99': textwrap.dedent("""\ + STARTMODE=auto""" + ), + "ifcfg-en0.99": textwrap.dedent( + """\ BOOTPROTO=static IPADDR=192.168.2.2 IPADDR1=192.168.1.2 @@ -2439,18 +2785,22 @@ iface bond0 inet6 static STARTMODE=auto ETHERDEVICE=en0 VLAN_ID=99 - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-en0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-en0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=en0 HWADDR=aa:bb:cc:dd:e8:00 NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet - USERCTL=no"""), - 'ifcfg-en0.99': textwrap.dedent("""\ + USERCTL=no""" + ), + "ifcfg-en0.99": textwrap.dedent( + """\ BOOTPROTO=none DEFROUTE=yes DEVICE=en0.99 @@ -2469,11 +2819,13 @@ iface bond0 inet6 static ONBOOT=yes PHYSDEV=en0 USERCTL=no - VLAN=yes"""), + VLAN=yes""" + ), }, }, - 'bridge': { - 'yaml': textwrap.dedent(""" + "bridge": { + "yaml": textwrap.dedent( + """ version: 1 config: - type: physical @@ -2498,9 +2850,11 @@ iface bond0 inet6 static bridge_bridgeprio: 22 subnets: - type: static - address: 192.168.2.2/24"""), - 'expected_sysconfig_opensuse': { - 'ifcfg-br0': textwrap.dedent("""\ + address: 192.168.2.2/24""" + ), + "expected_sysconfig_opensuse": { + "ifcfg-br0": textwrap.dedent( + """\ BOOTPROTO=static IPADDR=192.168.2.2 NETMASK=255.255.255.0 @@ -2508,24 +2862,30 @@ iface bond0 inet6 static BRIDGE_STP=off BRIDGE_PRIORITY=22 BRIDGE_PORTS='eth0 eth1' - """), - 'ifcfg-eth0': textwrap.dedent("""\ + """ + ), + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=static BRIDGE=yes LLADDR=52:54:00:12:34:00 IPADDR6=2001:1::100/96 STARTMODE=auto - """), - 'ifcfg-eth1': textwrap.dedent("""\ + """ + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=static BRIDGE=yes LLADDR=52:54:00:12:34:01 IPADDR6=2001:1::101/96 STARTMODE=auto - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-br0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-br0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=br0 IPADDR=192.168.2.2 @@ -2536,8 +2896,10 @@ iface bond0 inet6 static STP=no TYPE=Bridge USERCTL=no - """), - 'ifcfg-eth0': textwrap.dedent("""\ + """ + ), + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=none BRIDGE=br0 DEVICE=eth0 @@ -2550,8 +2912,10 @@ iface bond0 inet6 static ONBOOT=yes TYPE=Ethernet USERCTL=no - """), - 'ifcfg-eth1': textwrap.dedent("""\ + """ + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=none BRIDGE=br0 DEVICE=eth1 @@ -2564,11 +2928,13 @@ iface bond0 inet6 static ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, - 'manual': { - 'yaml': textwrap.dedent(""" + "manual": { + "yaml": textwrap.dedent( + """ version: 1 config: - type: physical @@ -2590,8 +2956,10 @@ iface bond0 inet6 static subnets: - type: manual control: manual - """), - 'expected_eni': textwrap.dedent("""\ + """ + ), + "expected_eni": textwrap.dedent( + """\ auto lo iface lo inet loopback @@ -2605,8 +2973,10 @@ iface bond0 inet6 static # control-manual eth2 iface eth2 inet manual - """), - 'expected_netplan': textwrap.dedent("""\ + """ + ), + "expected_netplan": textwrap.dedent( + """\ network: version: 2 @@ -2626,29 +2996,37 @@ iface bond0 inet6 static match: macaddress: 52:54:00:12:34:ff set-name: eth2 - """), - 'expected_sysconfig_opensuse': { - 'ifcfg-eth0': textwrap.dedent("""\ + """ + ), + "expected_sysconfig_opensuse": { + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=52:54:00:12:34:00 IPADDR=192.168.1.2 NETMASK=255.255.255.0 STARTMODE=manual - """), - 'ifcfg-eth1': textwrap.dedent("""\ + """ + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=52:54:00:12:34:aa MTU=1480 STARTMODE=auto - """), - 'ifcfg-eth2': textwrap.dedent("""\ + """ + ), + "ifcfg-eth2": textwrap.dedent( + """\ BOOTPROTO=static LLADDR=52:54:00:12:34:ff STARTMODE=manual - """), + """ + ), }, - 'expected_sysconfig_rhel': { - 'ifcfg-eth0': textwrap.dedent("""\ + "expected_sysconfig_rhel": { + "ifcfg-eth0": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth0 HWADDR=52:54:00:12:34:00 @@ -2658,8 +3036,10 @@ iface bond0 inet6 static ONBOOT=no TYPE=Ethernet USERCTL=no - """), - 'ifcfg-eth1': textwrap.dedent("""\ + """ + ), + "ifcfg-eth1": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth1 HWADDR=52:54:00:12:34:aa @@ -2668,8 +3048,10 @@ iface bond0 inet6 static ONBOOT=yes TYPE=Ethernet USERCTL=no - """), - 'ifcfg-eth2': textwrap.dedent("""\ + """ + ), + "ifcfg-eth2": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eth2 HWADDR=52:54:00:12:34:ff @@ -2677,51 +3059,85 @@ iface bond0 inet6 static ONBOOT=no TYPE=Ethernet USERCTL=no - """), + """ + ), }, }, } CONFIG_V1_EXPLICIT_LOOPBACK = { - 'version': 1, - 'config': [{'name': 'eth0', 'type': 'physical', - 'subnets': [{'control': 'auto', 'type': 'dhcp'}]}, - {'name': 'lo', 'type': 'loopback', - 'subnets': [{'control': 'auto', 'type': 'loopback'}]}, - ]} + "version": 1, + "config": [ + { + "name": "eth0", + "type": "physical", + "subnets": [{"control": "auto", "type": "dhcp"}], + }, + { + "name": "lo", + "type": "loopback", + "subnets": [{"control": "auto", "type": "loopback"}], + }, + ], +} CONFIG_V1_SIMPLE_SUBNET = { - 'version': 1, - 'config': [{'mac_address': '52:54:00:12:34:00', - 'name': 'interface0', - 'subnets': [{'address': '10.0.2.15', - 'gateway': '10.0.2.2', - 'netmask': '255.255.255.0', - 'type': 'static'}], - 'type': 'physical'}]} + "version": 1, + "config": [ + { + "mac_address": "52:54:00:12:34:00", + "name": "interface0", + "subnets": [ + { + "address": "10.0.2.15", + "gateway": "10.0.2.2", + "netmask": "255.255.255.0", + "type": "static", + } + ], + "type": "physical", + } + ], +} CONFIG_V1_MULTI_IFACE = { - 'version': 1, - 'config': [{'type': 'physical', - 'mtu': 1500, - 'subnets': [{'type': 'static', - 'netmask': '255.255.240.0', - 'routes': [{'netmask': '0.0.0.0', - 'network': '0.0.0.0', - 'gateway': '51.68.80.1'}], - 'address': '51.68.89.122', - 'ipv4': True}], - 'mac_address': 'fa:16:3e:25:b4:59', - 'name': 'eth0'}, - {'type': 'physical', - 'mtu': 9000, - 'subnets': [{'type': 'dhcp4'}], - 'mac_address': 'fa:16:3e:b1:ca:29', 'name': 'eth1'}]} + "version": 1, + "config": [ + { + "type": "physical", + "mtu": 1500, + "subnets": [ + { + "type": "static", + "netmask": "255.255.240.0", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "51.68.80.1", + } + ], + "address": "51.68.89.122", + "ipv4": True, + } + ], + "mac_address": "fa:16:3e:25:b4:59", + "name": "eth0", + }, + { + "type": "physical", + "mtu": 9000, + "subnets": [{"type": "dhcp4"}], + "mac_address": "fa:16:3e:b1:ca:29", + "name": "eth1", + }, + ], +} DEFAULT_DEV_ATTRS = { - 'eth1000': { + "eth1000": { "bridge": False, "carrier": False, "dormant": False, @@ -2734,16 +3150,26 @@ DEFAULT_DEV_ATTRS = { } -def _setup_test(tmp_dir, mock_get_devicelist, mock_read_sys_net, - mock_sys_dev_path, dev_attrs=None): +def _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=None, +): if not dev_attrs: dev_attrs = DEFAULT_DEV_ATTRS mock_get_devicelist.return_value = dev_attrs.keys() - def fake_read(devname, path, translate=None, - on_enoent=None, on_keyerror=None, - on_einval=None): + def fake_read( + devname, + path, + translate=None, + on_enoent=None, + on_keyerror=None, + on_einval=None, + ): return dev_attrs[devname][path] mock_read_sys_net.side_effect = fake_read @@ -2753,99 +3179,137 @@ def _setup_test(tmp_dir, mock_get_devicelist, mock_read_sys_net, for dev in dev_attrs: os.makedirs(os.path.join(tmp_dir, dev)) - with open(os.path.join(tmp_dir, dev, 'operstate'), 'w') as fh: - fh.write(dev_attrs[dev]['operstate']) + with open(os.path.join(tmp_dir, dev, "operstate"), "w") as fh: + fh.write(dev_attrs[dev]["operstate"]) os.makedirs(os.path.join(tmp_dir, dev, "device")) - for key in ['device/driver']: + for key in ["device/driver"]: if key in dev_attrs[dev] and dev_attrs[dev][key]: target = dev_attrs[dev][key] link = os.path.join(tmp_dir, dev, key) - print('symlink %s -> %s' % (link, target)) + print("symlink %s -> %s" % (link, target)) os.symlink(target, link) mock_sys_dev_path.side_effect = sys_dev_path class TestGenerateFallbackConfig(CiTestCase): - def setUp(self): super(TestGenerateFallbackConfig, self).setUp() self.add_patch( - "cloudinit.util.get_cmdline", "m_get_cmdline", - return_value="root=/dev/sda1") + "cloudinit.util.get_cmdline", + "m_get_cmdline", + return_value="root=/dev/sda1", + ) @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_device_driver_v2(self, mock_get_devicelist, mock_read_sys_net, - mock_sys_dev_path): + def test_device_driver_v2( + self, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ): """Network configuration for generate_fallback_config is version 2.""" devices = { - 'eth0': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'hv_netsvc', 'device/device': '0x3', - 'name_assign_type': '4'}, - 'eth1': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'mlx4_core', 'device/device': '0x7', - 'name_assign_type': '4'}, - + "eth0": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "hv_netsvc", + "device/device": "0x3", + "name_assign_type": "4", + }, + "eth1": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "mlx4_core", + "device/device": "0x7", + "name_assign_type": "4", + }, } tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - dev_attrs=devices) + _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=devices, + ) network_cfg = net.generate_fallback_config(config_driver=True) expected = { - 'ethernets': {'eth0': {'dhcp4': True, 'set-name': 'eth0', - 'match': {'macaddress': '00:11:22:33:44:55', - 'driver': 'hv_netsvc'}}}, - 'version': 2} + "ethernets": { + "eth0": { + "dhcp4": True, + "set-name": "eth0", + "match": { + "macaddress": "00:11:22:33:44:55", + "driver": "hv_netsvc", + }, + } + }, + "version": 2, + } self.assertEqual(expected, network_cfg) @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_device_driver(self, mock_get_devicelist, mock_read_sys_net, - mock_sys_dev_path): + def test_device_driver( + self, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ): devices = { - 'eth0': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'hv_netsvc', 'device/device': '0x3', - 'name_assign_type': '4'}, - 'eth1': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'mlx4_core', 'device/device': '0x7', - 'name_assign_type': '4'}, - + "eth0": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "hv_netsvc", + "device/device": "0x3", + "name_assign_type": "4", + }, + "eth1": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "mlx4_core", + "device/device": "0x7", + "name_assign_type": "4", + }, } tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - dev_attrs=devices) + _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=devices, + ) network_cfg = net.generate_fallback_config(config_driver=True) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) # don't set rulepath so eni writes them renderer = eni.Renderer( - {'eni_path': 'interfaces', 'netrules_path': 'netrules'}) + {"eni_path": "interfaces", "netrules_path": "netrules"} + ) renderer.render_network_state(ns, target=render_dir) - self.assertTrue(os.path.exists(os.path.join(render_dir, - 'interfaces'))) - with open(os.path.join(render_dir, 'interfaces')) as fh: + self.assertTrue(os.path.exists(os.path.join(render_dir, "interfaces"))) + with open(os.path.join(render_dir, "interfaces")) as fh: contents = fh.read() print(contents) expected = """ @@ -2857,8 +3321,8 @@ iface eth0 inet dhcp """ self.assertEqual(expected.lstrip(), contents.lstrip()) - self.assertTrue(os.path.exists(os.path.join(render_dir, 'netrules'))) - with open(os.path.join(render_dir, 'netrules')) as fh: + self.assertTrue(os.path.exists(os.path.join(render_dir, "netrules"))) + with open(os.path.join(render_dir, "netrules")) as fh: contents = fh.read() print(contents) expected_rule = [ @@ -2868,48 +3332,65 @@ iface eth0 inet dhcp 'ATTR{address}=="00:11:22:33:44:55"', 'NAME="eth0"', ] - self.assertEqual(", ".join(expected_rule) + '\n', contents.lstrip()) + self.assertEqual(", ".join(expected_rule) + "\n", contents.lstrip()) @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_device_driver_blacklist(self, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path): + def test_device_driver_blacklist( + self, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ): devices = { - 'eth1': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'hv_netsvc', 'device/device': '0x3', - 'name_assign_type': '4'}, - 'eth0': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'mlx4_core', 'device/device': '0x7', - 'name_assign_type': '4'}, + "eth1": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "hv_netsvc", + "device/device": "0x3", + "name_assign_type": "4", + }, + "eth0": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "mlx4_core", + "device/device": "0x7", + "name_assign_type": "4", + }, } tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - dev_attrs=devices) + _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=devices, + ) - blacklist = ['mlx4_core'] - network_cfg = net.generate_fallback_config(blacklist_drivers=blacklist, - config_driver=True) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + blacklist = ["mlx4_core"] + network_cfg = net.generate_fallback_config( + blacklist_drivers=blacklist, config_driver=True + ) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) # don't set rulepath so eni writes them renderer = eni.Renderer( - {'eni_path': 'interfaces', 'netrules_path': 'netrules'}) + {"eni_path": "interfaces", "netrules_path": "netrules"} + ) renderer.render_network_state(ns, target=render_dir) - self.assertTrue(os.path.exists(os.path.join(render_dir, - 'interfaces'))) - with open(os.path.join(render_dir, 'interfaces')) as fh: + self.assertTrue(os.path.exists(os.path.join(render_dir, "interfaces"))) + with open(os.path.join(render_dir, "interfaces")) as fh: contents = fh.read() print(contents) expected = """ @@ -2921,8 +3402,8 @@ iface eth1 inet dhcp """ self.assertEqual(expected.lstrip(), contents.lstrip()) - self.assertTrue(os.path.exists(os.path.join(render_dir, 'netrules'))) - with open(os.path.join(render_dir, 'netrules')) as fh: + self.assertTrue(os.path.exists(os.path.join(render_dir, "netrules"))) + with open(os.path.join(render_dir, "netrules")) as fh: contents = fh.read() print(contents) expected_rule = [ @@ -2932,35 +3413,54 @@ iface eth1 inet dhcp 'ATTR{address}=="00:11:22:33:44:55"', 'NAME="eth1"', ] - self.assertEqual(", ".join(expected_rule) + '\n', contents.lstrip()) + self.assertEqual(", ".join(expected_rule) + "\n", contents.lstrip()) @mock.patch("cloudinit.util.get_cmdline") @mock.patch("cloudinit.util.udevadm_settle") @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_unstable_names(self, mock_get_devicelist, mock_read_sys_net, - mock_sys_dev_path, mock_settle, m_get_cmdline): + def test_unstable_names( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + mock_settle, + m_get_cmdline, + ): """verify that udevadm settle is called when we find unstable names""" devices = { - 'eth0': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'hv_netsvc', 'device/device': '0x3', - 'name_assign_type': False}, - 'ens4': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'mlx4_core', 'device/device': '0x7', - 'name_assign_type': '4'}, - + "eth0": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "hv_netsvc", + "device/device": "0x3", + "name_assign_type": False, + }, + "ens4": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "mlx4_core", + "device/device": "0x7", + "name_assign_type": "4", + }, } - m_get_cmdline.return_value = '' + m_get_cmdline.return_value = "" tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - dev_attrs=devices) + _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=devices, + ) net.generate_fallback_config(config_driver=True) self.assertEqual(1, mock_settle.call_count) @@ -2969,52 +3469,73 @@ iface eth1 inet dhcp @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_unstable_names_disabled(self, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - mock_settle, m_get_cmdline): + def test_unstable_names_disabled( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + mock_settle, + m_get_cmdline, + ): """verify udevadm settle not called when cmdline has net.ifnames=0""" devices = { - 'eth0': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'hv_netsvc', 'device/device': '0x3', - 'name_assign_type': False}, - 'ens4': { - 'bridge': False, 'carrier': False, 'dormant': False, - 'operstate': 'down', 'address': '00:11:22:33:44:55', - 'device/driver': 'mlx4_core', 'device/device': '0x7', - 'name_assign_type': '4'}, - + "eth0": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "hv_netsvc", + "device/device": "0x3", + "name_assign_type": False, + }, + "ens4": { + "bridge": False, + "carrier": False, + "dormant": False, + "operstate": "down", + "address": "00:11:22:33:44:55", + "device/driver": "mlx4_core", + "device/device": "0x7", + "name_assign_type": "4", + }, } - m_get_cmdline.return_value = 'net.ifnames=0' + m_get_cmdline.return_value = "net.ifnames=0" tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path, - dev_attrs=devices) + _setup_test( + tmp_dir, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + dev_attrs=devices, + ) net.generate_fallback_config(config_driver=True) self.assertEqual(0, mock_settle.call_count) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", - mock.Mock(return_value=False) + mock.Mock(return_value=False), ) class TestRhelSysConfigRendering(CiTestCase): with_logs = True nm_cfg_file = "/etc/NetworkManager/NetworkManager.conf" - scripts_dir = '/etc/sysconfig/network-scripts' - header = ('# Created by cloud-init on instance boot automatically, ' - 'do not edit.\n#\n') + scripts_dir = "/etc/sysconfig/network-scripts" + header = ( + "# Created by cloud-init on instance boot automatically, " + "do not edit.\n#\n" + ) - expected_name = 'expected_sysconfig_rhel' + expected_name = "expected_sysconfig_rhel" def _get_renderer(self): - distro_cls = distros.fetch('rhel') + distro_cls = distros.fetch("rhel") return sysconfig.Renderer( - config=distro_cls.renderer_configs.get('sysconfig')) + config=distro_cls.renderer_configs.get("sysconfig") + ) def _render_and_read(self, network_config=None, state=None, dir=None): if dir is None: @@ -3032,9 +3553,8 @@ class TestRhelSysConfigRendering(CiTestCase): return dir2dict(dir) def _compare_files_to_expected(self, expected, found): - def _try_load(f): - ''' Attempt to load shell content, otherwise return as-is ''' + """Attempt to load shell content, otherwise return as-is""" try: return util.load_shell_content(f) except ValueError: @@ -3045,12 +3565,15 @@ class TestRhelSysConfigRendering(CiTestCase): orig_maxdiff = self.maxDiff expected_d = dict( (os.path.join(self.scripts_dir, k), _try_load(v)) - for k, v in expected.items()) + for k, v in expected.items() + ) # only compare the files in scripts_dir scripts_found = dict( - (k, _try_load(v)) for k, v in found.items() - if k.startswith(self.scripts_dir)) + (k, _try_load(v)) + for k, v in found.items() + if k.startswith(self.scripts_dir) + ) try: self.maxDiff = None self.assertEqual(expected_d, scripts_found) @@ -3058,9 +3581,14 @@ class TestRhelSysConfigRendering(CiTestCase): self.maxDiff = orig_maxdiff def _assert_headers(self, found): - missing = [f for f in found - if (f.startswith(self.scripts_dir) and - not found[f].startswith(self.header))] + missing = [ + f + for f in found + if ( + f.startswith(self.scripts_dir) + and not found[f].startswith(self.header) + ) + ] if missing: raise AssertionError("Missing headers in: %s" % missing) @@ -3068,16 +3596,22 @@ class TestRhelSysConfigRendering(CiTestCase): @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_default_generation(self, mock_get_devicelist, - mock_read_sys_net, - mock_sys_dev_path, m_get_cmdline): + def test_default_generation( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + m_get_cmdline, + ): tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path) + _setup_test( + tmp_dir, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ) network_cfg = net.generate_fallback_config() - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) @@ -3085,7 +3619,7 @@ class TestRhelSysConfigRendering(CiTestCase): renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) - render_file = 'etc/sysconfig/network-scripts/ifcfg-eth1000' + render_file = "etc/sysconfig/network-scripts/ifcfg-eth1000" with open(os.path.join(render_dir, render_file)) as fh: content = fh.read() expected_content = """ @@ -3105,35 +3639,44 @@ USERCTL=no """ValueError is raised when duplicate ipv4 gateways exist.""" net_json = { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", - "type": "ipv4", "netmask": "255.255.252.0", - "link": "tap1a81968a-79", - "routes": [{ - "netmask": "0.0.0.0", - "network": "0.0.0.0", - "gateway": "172.19.3.254", - }, { - "netmask": "0.0.0.0", # A second default gateway - "network": "0.0.0.0", - "gateway": "172.20.3.254", - }], - "ip_address": "172.19.1.34", "id": "network0" - }], + "networks": [ + { + "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", + "type": "ipv4", + "netmask": "255.255.252.0", + "link": "tap1a81968a-79", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "172.19.3.254", + }, + { + "netmask": "0.0.0.0", # A second default gateway + "network": "0.0.0.0", + "gateway": "172.20.3.254", + }, + ], + "ip_address": "172.19.1.34", + "id": "network0", + } + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], } - macs = {'fa:16:3e:ed:9a:59': 'eth0'} + macs = {"fa:16:3e:ed:9a:59": "eth0"} render_dir = self.tmp_dir() network_cfg = openstack.convert_net_json(net_json, known_macs=macs) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) renderer = self._get_renderer() with self.assertRaises(ValueError): renderer.render_network_state(ns, target=render_dir) @@ -3143,35 +3686,44 @@ USERCTL=no """ValueError is raised when duplicate ipv6 gateways exist.""" net_json = { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "public-ipv6", - "type": "ipv6", "netmask": "", - "link": "tap1a81968a-79", - "routes": [{ - "gateway": "2001:DB8::1", - "netmask": "::", - "network": "::" - }, { - "gateway": "2001:DB9::1", - "netmask": "::", - "network": "::" - }], - "ip_address": "2001:DB8::10", "id": "network1" - }], + "networks": [ + { + "network_id": "public-ipv6", + "type": "ipv6", + "netmask": "", + "link": "tap1a81968a-79", + "routes": [ + { + "gateway": "2001:DB8::1", + "netmask": "::", + "network": "::", + }, + { + "gateway": "2001:DB9::1", + "netmask": "::", + "network": "::", + }, + ], + "ip_address": "2001:DB8::10", + "id": "network1", + } + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], } - macs = {'fa:16:3e:ed:9a:59': 'eth0'} + macs = {"fa:16:3e:ed:9a:59": "eth0"} render_dir = self.tmp_dir() network_cfg = openstack.convert_net_json(net_json, known_macs=macs) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) renderer = self._get_renderer() with self.assertRaises(ValueError): renderer.render_network_state(ns, target=render_dir) @@ -3180,19 +3732,22 @@ USERCTL=no def test_openstack_rendering_samples(self): for os_sample in OS_SAMPLES: render_dir = self.tmp_dir() - ex_input = os_sample['in_data'] - ex_mac_addrs = os_sample['in_macs'] + ex_input = os_sample["in_data"] + ex_mac_addrs = os_sample["in_macs"] network_cfg = openstack.convert_net_json( - ex_input, known_macs=ex_mac_addrs) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ex_input, known_macs=ex_mac_addrs + ) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) renderer = self._get_renderer() # render a multiple times to simulate reboots renderer.render_network_state(ns, target=render_dir) renderer.render_network_state(ns, target=render_dir) renderer.render_network_state(ns, target=render_dir) - for fn, expected_content in os_sample.get('out_sysconfig_rhel', - []): + for fn, expected_content in os_sample.get( + "out_sysconfig_rhel", [] + ): with open(os.path.join(render_dir, fn)) as fh: self.assertEqual(expected_content, fh.read()) @@ -3203,8 +3758,8 @@ USERCTL=no renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) found = dir2dict(render_dir) - nspath = '/etc/sysconfig/network-scripts/' - self.assertNotIn(nspath + 'ifcfg-lo', found.keys()) + nspath = "/etc/sysconfig/network-scripts/" + self.assertNotIn(nspath + "ifcfg-lo", found.keys()) expected = """\ # Created by cloud-init on instance boot automatically, do not edit. # @@ -3220,10 +3775,10 @@ ONBOOT=yes TYPE=Ethernet USERCTL=no """ - self.assertEqual(expected, found[nspath + 'ifcfg-interface0']) + self.assertEqual(expected, found[nspath + "ifcfg-interface0"]) # The configuration has no nameserver information make sure we # do not write the resolv.conf file - respath = '/etc/resolv.conf' + respath = "/etc/resolv.conf" self.assertNotIn(respath, found.keys()) def test_network_config_v1_multi_iface_samples(self): @@ -3233,8 +3788,8 @@ USERCTL=no renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) found = dir2dict(render_dir) - nspath = '/etc/sysconfig/network-scripts/' - self.assertNotIn(nspath + 'ifcfg-lo', found.keys()) + nspath = "/etc/sysconfig/network-scripts/" + self.assertNotIn(nspath + "ifcfg-lo", found.keys()) expected_i1 = """\ # Created by cloud-init on instance boot automatically, do not edit. # @@ -3251,7 +3806,7 @@ ONBOOT=yes TYPE=Ethernet USERCTL=no """ - self.assertEqual(expected_i1, found[nspath + 'ifcfg-eth0']) + self.assertEqual(expected_i1, found[nspath + "ifcfg-eth0"]) expected_i2 = """\ # Created by cloud-init on instance boot automatically, do not edit. # @@ -3265,21 +3820,21 @@ ONBOOT=yes TYPE=Ethernet USERCTL=no """ - self.assertEqual(expected_i2, found[nspath + 'ifcfg-eth1']) + self.assertEqual(expected_i2, found[nspath + "ifcfg-eth1"]) def test_config_with_explicit_loopback(self): ns = network_state.parse_net_config_data(CONFIG_V1_EXPLICIT_LOOPBACK) render_dir = self.tmp_path("render") os.makedirs(render_dir) # write an etc/resolv.conf and expect it to not be modified - resolvconf = os.path.join(render_dir, 'etc/resolv.conf') + resolvconf = os.path.join(render_dir, "etc/resolv.conf") resolvconf_content = "# Original Content" util.write_file(resolvconf, resolvconf_content) renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) found = dir2dict(render_dir) - nspath = '/etc/sysconfig/network-scripts/' - self.assertNotIn(nspath + 'ifcfg-lo', found.keys()) + nspath = "/etc/sysconfig/network-scripts/" + self.assertNotIn(nspath + "ifcfg-lo", found.keys()) expected = """\ # Created by cloud-init on instance boot automatically, do not edit. # @@ -3290,171 +3845,188 @@ ONBOOT=yes TYPE=Ethernet USERCTL=no """ - self.assertEqual(expected, found[nspath + 'ifcfg-eth0']) + self.assertEqual(expected, found[nspath + "ifcfg-eth0"]) # a dhcp only config should not modify resolv.conf - self.assertEqual(resolvconf_content, found['/etc/resolv.conf']) + self.assertEqual(resolvconf_content, found["/etc/resolv.conf"]) def test_bond_config(self): - entry = NETWORK_CONFIGS['bond'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["bond"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_vlan_config(self): - entry = NETWORK_CONFIGS['vlan'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["vlan"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_bridge_config(self): - entry = NETWORK_CONFIGS['bridge'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["bridge"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_manual_config(self): - entry = NETWORK_CONFIGS['manual'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["manual"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_all_config(self): - entry = NETWORK_CONFIGS['all'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["all"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) self.assertNotIn( - 'WARNING: Network config: ignoring eth0.101 device-level mtu', - self.logs.getvalue()) + "WARNING: Network config: ignoring eth0.101 device-level mtu", + self.logs.getvalue(), + ) def test_small_config(self): - entry = NETWORK_CONFIGS['small'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["small"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_v4_and_v6_static_config(self): - entry = NETWORK_CONFIGS['v4_and_v6_static'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6_static"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) expected_msg = ( - 'WARNING: Network config: ignoring iface0 device-level mtu:8999' - ' because ipv4 subnet-level mtu:9000 provided.') + "WARNING: Network config: ignoring iface0 device-level mtu:8999" + " because ipv4 subnet-level mtu:9000 provided." + ) self.assertIn(expected_msg, self.logs.getvalue()) def test_dhcpv6_only_config(self): - entry = NETWORK_CONFIGS['dhcpv6_only'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_only"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_accept_ra_config_v1(self): - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_accept_ra_config_v2(self): - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_reject_ra_config_v1(self): - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_stattic6_from_json(self): net_json = { "services": [{"type": "dns", "address": "172.19.0.12"}], - "networks": [{ - "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", - "type": "ipv4", "netmask": "255.255.252.0", - "link": "tap1a81968a-79", - "routes": [{ - "netmask": "0.0.0.0", - "network": "0.0.0.0", - "gateway": "172.19.3.254", - }, { - "netmask": "0.0.0.0", # A second default gateway - "network": "0.0.0.0", - "gateway": "172.20.3.254", - }], - "ip_address": "172.19.1.34", "id": "network0" - }, { - "network_id": "mgmt", - "netmask": "ffff:ffff:ffff:ffff::", - "link": "interface1", - "mode": "link-local", - "routes": [], - "ip_address": "fe80::c096:67ff:fe5c:6e84", - "type": "static6", - "id": "network1", - "services": [], - "accept-ra": "false" - }], + "networks": [ + { + "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", + "type": "ipv4", + "netmask": "255.255.252.0", + "link": "tap1a81968a-79", + "routes": [ + { + "netmask": "0.0.0.0", + "network": "0.0.0.0", + "gateway": "172.19.3.254", + }, + { + "netmask": "0.0.0.0", # A second default gateway + "network": "0.0.0.0", + "gateway": "172.20.3.254", + }, + ], + "ip_address": "172.19.1.34", + "id": "network0", + }, + { + "network_id": "mgmt", + "netmask": "ffff:ffff:ffff:ffff::", + "link": "interface1", + "mode": "link-local", + "routes": [], + "ip_address": "fe80::c096:67ff:fe5c:6e84", + "type": "static6", + "id": "network1", + "services": [], + "accept-ra": "false", + }, + ], "links": [ { "ethernet_mac_address": "fa:16:3e:ed:9a:59", - "mtu": None, "type": "bridge", "id": - "tap1a81968a-79", - "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + "mtu": None, + "type": "bridge", + "id": "tap1a81968a-79", + "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f", }, ], } - macs = {'fa:16:3e:ed:9a:59': 'eth0'} + macs = {"fa:16:3e:ed:9a:59": "eth0"} render_dir = self.tmp_dir() network_cfg = openstack.convert_net_json(net_json, known_macs=macs) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) renderer = self._get_renderer() with self.assertRaises(ValueError): renderer.render_network_state(ns, target=render_dir) self.assertEqual([], os.listdir(render_dir)) def test_static6_from_yaml(self): - entry = NETWORK_CONFIGS['static6'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml'])) + entry = NETWORK_CONFIGS["static6"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_reject_ra_config_v2(self): - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_stateless_config(self): - entry = NETWORK_CONFIGS['dhcpv6_stateless'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateless"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_stateful_config(self): - entry = NETWORK_CONFIGS['dhcpv6_stateful'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateful"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_wakeonlan_disabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_disabled'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_disabled"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_wakeonlan_enabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_enabled'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_enabled"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) @@ -3465,20 +4037,21 @@ USERCTL=no util.ensure_dir(os.path.dirname(nm_cfg)) # write a template nm.conf, note plugins is a list here - with open(nm_cfg, 'w') as fh: - fh.write('# test_check_ifcfg_rh\n[main]\nplugins=foo,bar\n') + with open(nm_cfg, "w") as fh: + fh.write("# test_check_ifcfg_rh\n[main]\nplugins=foo,bar\n") self.assertTrue(os.path.exists(nm_cfg)) # render and read - entry = NETWORK_CONFIGS['small'] - found = self._render_and_read(network_config=yaml.load(entry['yaml']), - dir=render_dir) + entry = NETWORK_CONFIGS["small"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml"]), dir=render_dir + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) # check ifcfg-rh is in the 'plugins' list config = sysconfig.ConfigObj(nm_cfg) - self.assertIn('ifcfg-rh', config['main']['plugins']) + self.assertIn("ifcfg-rh", config["main"]["plugins"]) def test_check_ifcfg_rh_plugins_string(self): """ifcfg-rh plugin is append when plugins is a string.""" @@ -3488,22 +4061,23 @@ USERCTL=no util.ensure_dir(os.path.dirname(nm_cfg)) # write a template nm.conf, note plugins is a value here - util.write_file(nm_cfg, '# test_check_ifcfg_rh\n[main]\nplugins=foo\n') + util.write_file(nm_cfg, "# test_check_ifcfg_rh\n[main]\nplugins=foo\n") # render and read - entry = NETWORK_CONFIGS['small'] - found = self._render_and_read(network_config=yaml.load(entry['yaml']), - dir=render_dir) + entry = NETWORK_CONFIGS["small"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml"]), dir=render_dir + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) # check raw content has plugin nm_file_content = util.load_file(nm_cfg) - self.assertIn('ifcfg-rh', nm_file_content) + self.assertIn("ifcfg-rh", nm_file_content) # check ifcfg-rh is in the 'plugins' list config = sysconfig.ConfigObj(nm_cfg) - self.assertIn('ifcfg-rh', config['main']['plugins']) + self.assertIn("ifcfg-rh", config["main"]["plugins"]) def test_check_ifcfg_rh_plugins_no_plugins(self): """enable_ifcfg_plugin creates plugins value if missing.""" @@ -3513,28 +4087,32 @@ USERCTL=no util.ensure_dir(os.path.dirname(nm_cfg)) # write a template nm.conf, note plugins is missing - util.write_file(nm_cfg, '# test_check_ifcfg_rh\n[main]\n') + util.write_file(nm_cfg, "# test_check_ifcfg_rh\n[main]\n") self.assertTrue(os.path.exists(nm_cfg)) # render and read - entry = NETWORK_CONFIGS['small'] - found = self._render_and_read(network_config=yaml.load(entry['yaml']), - dir=render_dir) + entry = NETWORK_CONFIGS["small"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml"]), dir=render_dir + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) # check ifcfg-rh is in the 'plugins' list config = sysconfig.ConfigObj(nm_cfg) - self.assertIn('ifcfg-rh', config['main']['plugins']) + self.assertIn("ifcfg-rh", config["main"]["plugins"]) def test_netplan_dhcp_false_disable_dhcp_in_state(self): """netplan config with dhcp[46]: False should not add dhcp in state""" net_config = yaml.load(NETPLAN_DHCP_FALSE) - ns = network_state.parse_net_config_data(net_config, - skip_broken=False) + ns = network_state.parse_net_config_data(net_config, skip_broken=False) - dhcp_found = [snet for iface in ns.iter_interfaces() - for snet in iface['subnets'] if 'dhcp' in snet['type']] + dhcp_found = [ + snet + for iface in ns.iter_interfaces() + for snet in iface["subnets"] + if "dhcp" in snet["type"] + ] self.assertEqual([], dhcp_found) @@ -3542,9 +4120,10 @@ USERCTL=no """netplan cfg with dhcp[46]: False should not have bootproto=dhcp""" entry = { - 'yaml': NETPLAN_DHCP_FALSE, - 'expected_sysconfig': { - 'ifcfg-ens3': textwrap.dedent("""\ + "yaml": NETPLAN_DHCP_FALSE, + "expected_sysconfig": { + "ifcfg-ens3": textwrap.dedent( + """\ BOOTPROTO=none DEFROUTE=yes DEVICE=ens3 @@ -3564,33 +4143,42 @@ USERCTL=no ONBOOT=yes TYPE=Ethernet USERCTL=no - """), - } + """ + ), + }, } - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) - self._compare_files_to_expected(entry['expected_sysconfig'], found) + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) + self._compare_files_to_expected(entry["expected_sysconfig"], found) self._assert_headers(found) def test_from_v2_vlan_mtu(self): """verify mtu gets rendered on bond when source is netplan.""" v2data = { - 'version': 2, - 'ethernets': {'eno1': {}}, - 'vlans': { - 'eno1.1000': { - 'addresses': ["192.6.1.9/24"], - 'id': 1000, 'link': 'eno1', 'mtu': 1495}}} + "version": 2, + "ethernets": {"eno1": {}}, + "vlans": { + "eno1.1000": { + "addresses": ["192.6.1.9/24"], + "id": 1000, + "link": "eno1", + "mtu": 1495, + } + }, + } expected = { - 'ifcfg-eno1': textwrap.dedent("""\ + "ifcfg-eno1": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eno1 NM_CONTROLLED=no ONBOOT=yes TYPE=Ethernet USERCTL=no - """), - 'ifcfg-eno1.1000': textwrap.dedent("""\ + """ + ), + "ifcfg-eno1.1000": textwrap.dedent( + """\ BOOTPROTO=none DEVICE=eno1.1000 IPADDR=192.6.1.9 @@ -3601,23 +4189,29 @@ USERCTL=no PHYSDEV=eno1 USERCTL=no VLAN=yes - """) + """ + ), } self._compare_files_to_expected( - expected, self._render_and_read(network_config=v2data)) + expected, self._render_and_read(network_config=v2data) + ) def test_from_v2_bond_mtu(self): """verify mtu gets rendered on bond when source is netplan.""" v2data = { - 'version': 2, - 'bonds': { - 'bond0': {'addresses': ['10.101.8.65/26'], - 'interfaces': ['enp0s0', 'enp0s1'], - 'mtu': 1334, - 'parameters': {}}} + "version": 2, + "bonds": { + "bond0": { + "addresses": ["10.101.8.65/26"], + "interfaces": ["enp0s0", "enp0s1"], + "mtu": 1334, + "parameters": {}, + } + }, } expected = { - 'ifcfg-bond0': textwrap.dedent("""\ + "ifcfg-bond0": textwrap.dedent( + """\ BONDING_MASTER=yes BONDING_SLAVE0=enp0s0 BONDING_SLAVE1=enp0s1 @@ -3630,8 +4224,10 @@ USERCTL=no ONBOOT=yes TYPE=Bond USERCTL=no - """), - 'ifcfg-enp0s0': textwrap.dedent("""\ + """ + ), + "ifcfg-enp0s0": textwrap.dedent( + """\ BONDING_MASTER=yes BOOTPROTO=none DEVICE=enp0s0 @@ -3641,8 +4237,10 @@ USERCTL=no SLAVE=yes TYPE=Bond USERCTL=no - """), - 'ifcfg-enp0s1': textwrap.dedent("""\ + """ + ), + "ifcfg-enp0s1": textwrap.dedent( + """\ BONDING_MASTER=yes BOOTPROTO=none DEVICE=enp0s1 @@ -3652,21 +4250,28 @@ USERCTL=no SLAVE=yes TYPE=Bond USERCTL=no - """) + """ + ), } self._compare_files_to_expected( - expected, self._render_and_read(network_config=v2data)) + expected, self._render_and_read(network_config=v2data) + ) def test_from_v2_route_metric(self): """verify route-metric gets rendered on nic when source is netplan.""" - overrides = {'route-metric': 100} + overrides = {"route-metric": 100} v2base = { - 'version': 2, - 'ethernets': { - 'eno1': {'dhcp4': True, - 'match': {'macaddress': '07-1c-c6-75-a4-be'}}}} + "version": 2, + "ethernets": { + "eno1": { + "dhcp4": True, + "match": {"macaddress": "07-1c-c6-75-a4-be"}, + } + }, + } expected = { - 'ifcfg-eno1': textwrap.dedent("""\ + "ifcfg-eno1": textwrap.dedent( + """\ BOOTPROTO=dhcp DEVICE=eno1 HWADDR=07-1c-c6-75-a4-be @@ -3675,36 +4280,42 @@ USERCTL=no ONBOOT=yes TYPE=Ethernet USERCTL=no - """), + """ + ), } - for dhcp_ver in ('dhcp4', 'dhcp6'): + for dhcp_ver in ("dhcp4", "dhcp6"): v2data = copy.deepcopy(v2base) - if dhcp_ver == 'dhcp6': - expected['ifcfg-eno1'] += "IPV6INIT=yes\nDHCPV6C=yes\n" - v2data['ethernets']['eno1'].update( - {dhcp_ver: True, '{0}-overrides'.format(dhcp_ver): overrides}) + if dhcp_ver == "dhcp6": + expected["ifcfg-eno1"] += "IPV6INIT=yes\nDHCPV6C=yes\n" + v2data["ethernets"]["eno1"].update( + {dhcp_ver: True, "{0}-overrides".format(dhcp_ver): overrides} + ) self._compare_files_to_expected( - expected, self._render_and_read(network_config=v2data)) + expected, self._render_and_read(network_config=v2data) + ) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", - mock.Mock(return_value=False) + mock.Mock(return_value=False), ) class TestOpenSuseSysConfigRendering(CiTestCase): with_logs = True - scripts_dir = '/etc/sysconfig/network' - header = ('# Created by cloud-init on instance boot automatically, ' - 'do not edit.\n#\n') + scripts_dir = "/etc/sysconfig/network" + header = ( + "# Created by cloud-init on instance boot automatically, " + "do not edit.\n#\n" + ) - expected_name = 'expected_sysconfig_opensuse' + expected_name = "expected_sysconfig_opensuse" def _get_renderer(self): - distro_cls = distros.fetch('opensuse') + distro_cls = distros.fetch("opensuse") return sysconfig.Renderer( - config=distro_cls.renderer_configs.get('sysconfig')) + config=distro_cls.renderer_configs.get("sysconfig") + ) def _render_and_read(self, network_config=None, state=None, dir=None): if dir is None: @@ -3725,12 +4336,15 @@ class TestOpenSuseSysConfigRendering(CiTestCase): orig_maxdiff = self.maxDiff expected_d = dict( (os.path.join(self.scripts_dir, k), util.load_shell_content(v)) - for k, v in expected.items()) + for k, v in expected.items() + ) # only compare the files in scripts_dir scripts_found = dict( - (k, util.load_shell_content(v)) for k, v in found.items() - if k.startswith(self.scripts_dir)) + (k, util.load_shell_content(v)) + for k, v in found.items() + if k.startswith(self.scripts_dir) + ) try: self.maxDiff = None self.assertEqual(expected_d, scripts_found) @@ -3738,9 +4352,14 @@ class TestOpenSuseSysConfigRendering(CiTestCase): self.maxDiff = orig_maxdiff def _assert_headers(self, found): - missing = [f for f in found - if (f.startswith(self.scripts_dir) and - not found[f].startswith(self.header))] + missing = [ + f + for f in found + if ( + f.startswith(self.scripts_dir) + and not found[f].startswith(self.header) + ) + ] if missing: raise AssertionError("Missing headers in: %s" % missing) @@ -3748,16 +4367,22 @@ class TestOpenSuseSysConfigRendering(CiTestCase): @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_default_generation(self, mock_get_devicelist, - mock_read_sys_net, - mock_sys_dev_path, m_get_cmdline): + def test_default_generation( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + m_get_cmdline, + ): tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path) + _setup_test( + tmp_dir, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ) network_cfg = net.generate_fallback_config() - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) @@ -3765,7 +4390,7 @@ class TestOpenSuseSysConfigRendering(CiTestCase): renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) - render_file = 'etc/sysconfig/network/ifcfg-eth1000' + render_file = "etc/sysconfig/network/ifcfg-eth1000" with open(os.path.join(render_dir, render_file)) as fh: content = fh.read() expected_content = """ @@ -3779,98 +4404,101 @@ STARTMODE=auto # TODO(rjschwei): re-enable test once route writing is implemented # for SUSE distros -# def test_multiple_ipv4_default_gateways(self): -# """ValueError is raised when duplicate ipv4 gateways exist.""" -# net_json = { -# "services": [{"type": "dns", "address": "172.19.0.12"}], -# "networks": [{ -# "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", -# "type": "ipv4", "netmask": "255.255.252.0", -# "link": "tap1a81968a-79", -# "routes": [{ -# "netmask": "0.0.0.0", -# "network": "0.0.0.0", -# "gateway": "172.19.3.254", -# }, { -# "netmask": "0.0.0.0", # A second default gateway -# "network": "0.0.0.0", -# "gateway": "172.20.3.254", -# }], -# "ip_address": "172.19.1.34", "id": "network0" -# }], -# "links": [ -# { -# "ethernet_mac_address": "fa:16:3e:ed:9a:59", -# "mtu": None, "type": "bridge", "id": -# "tap1a81968a-79", -# "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" -# }, -# ], -# } -# macs = {'fa:16:3e:ed:9a:59': 'eth0'} -# render_dir = self.tmp_dir() -# network_cfg = openstack.convert_net_json(net_json, known_macs=macs) -# ns = network_state.parse_net_config_data(network_cfg, -# skip_broken=False) -# renderer = self._get_renderer() -# with self.assertRaises(ValueError): -# renderer.render_network_state(ns, target=render_dir) -# self.assertEqual([], os.listdir(render_dir)) -# -# def test_multiple_ipv6_default_gateways(self): -# """ValueError is raised when duplicate ipv6 gateways exist.""" -# net_json = { -# "services": [{"type": "dns", "address": "172.19.0.12"}], -# "networks": [{ -# "network_id": "public-ipv6", -# "type": "ipv6", "netmask": "", -# "link": "tap1a81968a-79", -# "routes": [{ -# "gateway": "2001:DB8::1", -# "netmask": "::", -# "network": "::" -# }, { -# "gateway": "2001:DB9::1", -# "netmask": "::", -# "network": "::" -# }], -# "ip_address": "2001:DB8::10", "id": "network1" -# }], -# "links": [ -# { -# "ethernet_mac_address": "fa:16:3e:ed:9a:59", -# "mtu": None, "type": "bridge", "id": -# "tap1a81968a-79", -# "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" -# }, -# ], -# } -# macs = {'fa:16:3e:ed:9a:59': 'eth0'} -# render_dir = self.tmp_dir() -# network_cfg = openstack.convert_net_json(net_json, known_macs=macs) -# ns = network_state.parse_net_config_data(network_cfg, -# skip_broken=False) -# renderer = self._get_renderer() -# with self.assertRaises(ValueError): -# renderer.render_network_state(ns, target=render_dir) -# self.assertEqual([], os.listdir(render_dir)) + # def test_multiple_ipv4_default_gateways(self): + # """ValueError is raised when duplicate ipv4 gateways exist.""" + # net_json = { + # "services": [{"type": "dns", "address": "172.19.0.12"}], + # "networks": [{ + # "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4", + # "type": "ipv4", "netmask": "255.255.252.0", + # "link": "tap1a81968a-79", + # "routes": [{ + # "netmask": "0.0.0.0", + # "network": "0.0.0.0", + # "gateway": "172.19.3.254", + # }, { + # "netmask": "0.0.0.0", # A second default gateway + # "network": "0.0.0.0", + # "gateway": "172.20.3.254", + # }], + # "ip_address": "172.19.1.34", "id": "network0" + # }], + # "links": [ + # { + # "ethernet_mac_address": "fa:16:3e:ed:9a:59", + # "mtu": None, "type": "bridge", "id": + # "tap1a81968a-79", + # "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + # }, + # ], + # } + # macs = {'fa:16:3e:ed:9a:59': 'eth0'} + # render_dir = self.tmp_dir() + # network_cfg = openstack.convert_net_json(net_json, known_macs=macs) # noqa: E501 + # ns = network_state.parse_net_config_data(network_cfg, + # skip_broken=False) + # renderer = self._get_renderer() + # with self.assertRaises(ValueError): + # renderer.render_network_state(ns, target=render_dir) + # self.assertEqual([], os.listdir(render_dir)) + # + # def test_multiple_ipv6_default_gateways(self): + # """ValueError is raised when duplicate ipv6 gateways exist.""" + # net_json = { + # "services": [{"type": "dns", "address": "172.19.0.12"}], + # "networks": [{ + # "network_id": "public-ipv6", + # "type": "ipv6", "netmask": "", + # "link": "tap1a81968a-79", + # "routes": [{ + # "gateway": "2001:DB8::1", + # "netmask": "::", + # "network": "::" + # }, { + # "gateway": "2001:DB9::1", + # "netmask": "::", + # "network": "::" + # }], + # "ip_address": "2001:DB8::10", "id": "network1" + # }], + # "links": [ + # { + # "ethernet_mac_address": "fa:16:3e:ed:9a:59", + # "mtu": None, "type": "bridge", "id": + # "tap1a81968a-79", + # "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f" + # }, + # ], + # } + # macs = {'fa:16:3e:ed:9a:59': 'eth0'} + # render_dir = self.tmp_dir() + # network_cfg = openstack.convert_net_json(net_json, known_macs=macs) # noqa: E501 + # ns = network_state.parse_net_config_data(network_cfg, + # skip_broken=False) + # renderer = self._get_renderer() + # with self.assertRaises(ValueError): + # renderer.render_network_state(ns, target=render_dir) + # self.assertEqual([], os.listdir(render_dir)) def test_openstack_rendering_samples(self): for os_sample in OS_SAMPLES: render_dir = self.tmp_dir() - ex_input = os_sample['in_data'] - ex_mac_addrs = os_sample['in_macs'] + ex_input = os_sample["in_data"] + ex_mac_addrs = os_sample["in_macs"] network_cfg = openstack.convert_net_json( - ex_input, known_macs=ex_mac_addrs) - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ex_input, known_macs=ex_mac_addrs + ) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) renderer = self._get_renderer() # render a multiple times to simulate reboots renderer.render_network_state(ns, target=render_dir) renderer.render_network_state(ns, target=render_dir) renderer.render_network_state(ns, target=render_dir) - for fn, expected_content in os_sample.get('out_sysconfig_opensuse', - []): + for fn, expected_content in os_sample.get( + "out_sysconfig_opensuse", [] + ): with open(os.path.join(render_dir, fn)) as fh: self.assertEqual(expected_content, fh.read()) @@ -3881,8 +4509,8 @@ STARTMODE=auto renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) found = dir2dict(render_dir) - nspath = '/etc/sysconfig/network/' - self.assertNotIn(nspath + 'ifcfg-lo', found.keys()) + nspath = "/etc/sysconfig/network/" + self.assertNotIn(nspath + "ifcfg-lo", found.keys()) expected = """\ # Created by cloud-init on instance boot automatically, do not edit. # @@ -3892,10 +4520,10 @@ LLADDR=52:54:00:12:34:00 NETMASK=255.255.255.0 STARTMODE=auto """ - self.assertEqual(expected, found[nspath + 'ifcfg-interface0']) + self.assertEqual(expected, found[nspath + "ifcfg-interface0"]) # The configuration has no nameserver information make sure we # do not write the resolv.conf file - respath = '/etc/resolv.conf' + respath = "/etc/resolv.conf" self.assertNotIn(respath, found.keys()) def test_config_with_explicit_loopback(self): @@ -3903,33 +4531,33 @@ STARTMODE=auto render_dir = self.tmp_path("render") os.makedirs(render_dir) # write an etc/resolv.conf and expect it to not be modified - resolvconf = os.path.join(render_dir, 'etc/resolv.conf') + resolvconf = os.path.join(render_dir, "etc/resolv.conf") resolvconf_content = "# Original Content" util.write_file(resolvconf, resolvconf_content) renderer = self._get_renderer() renderer.render_network_state(ns, target=render_dir) found = dir2dict(render_dir) - nspath = '/etc/sysconfig/network/' - self.assertNotIn(nspath + 'ifcfg-lo', found.keys()) + nspath = "/etc/sysconfig/network/" + self.assertNotIn(nspath + "ifcfg-lo", found.keys()) expected = """\ # Created by cloud-init on instance boot automatically, do not edit. # BOOTPROTO=dhcp STARTMODE=auto """ - self.assertEqual(expected, found[nspath + 'ifcfg-eth0']) + self.assertEqual(expected, found[nspath + "ifcfg-eth0"]) # a dhcp only config should not modify resolv.conf - self.assertEqual(resolvconf_content, found['/etc/resolv.conf']) + self.assertEqual(resolvconf_content, found["/etc/resolv.conf"]) def test_bond_config(self): - expected_name = 'expected_sysconfig_opensuse' - entry = NETWORK_CONFIGS['bond'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + expected_name = "expected_sysconfig_opensuse" + entry = NETWORK_CONFIGS["bond"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) for fname, contents in entry[expected_name].items(): print(fname) print(contents) print() - print('-- expected ^ | v rendered --') + print("-- expected ^ | v rendered --") for fname, contents in found.items(): print(fname) print(contents) @@ -3938,120 +4566,129 @@ STARTMODE=auto self._assert_headers(found) def test_vlan_config(self): - entry = NETWORK_CONFIGS['vlan'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["vlan"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_bridge_config(self): - entry = NETWORK_CONFIGS['bridge'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["bridge"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_manual_config(self): - entry = NETWORK_CONFIGS['manual'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["manual"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_all_config(self): - entry = NETWORK_CONFIGS['all'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["all"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) self.assertNotIn( - 'WARNING: Network config: ignoring eth0.101 device-level mtu', - self.logs.getvalue()) + "WARNING: Network config: ignoring eth0.101 device-level mtu", + self.logs.getvalue(), + ) def test_small_config(self): - entry = NETWORK_CONFIGS['small'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["small"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_v4_and_v6_static_config(self): - entry = NETWORK_CONFIGS['v4_and_v6_static'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6_static"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) expected_msg = ( - 'WARNING: Network config: ignoring iface0 device-level mtu:8999' - ' because ipv4 subnet-level mtu:9000 provided.') + "WARNING: Network config: ignoring iface0 device-level mtu:8999" + " because ipv4 subnet-level mtu:9000 provided." + ) self.assertIn(expected_msg, self.logs.getvalue()) def test_dhcpv6_only_config(self): - entry = NETWORK_CONFIGS['dhcpv6_only'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_only"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_simple_render_ipv6_slaac(self): - entry = NETWORK_CONFIGS['ipv6_slaac'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["ipv6_slaac"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_dhcpv6_stateless_config(self): - entry = NETWORK_CONFIGS['dhcpv6_stateless'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateless"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_wakeonlan_disabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_disabled'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_disabled"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_wakeonlan_enabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_enabled'] - found = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_enabled"] + found = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_render_v4_and_v6(self): - entry = NETWORK_CONFIGS['v4_and_v6'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) def test_render_v6_and_v4(self): - entry = NETWORK_CONFIGS['v6_and_v4'] - found = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v6_and_v4"] + found = self._render_and_read(network_config=yaml.load(entry["yaml"])) self._compare_files_to_expected(entry[self.expected_name], found) self._assert_headers(found) class TestEniNetRendering(CiTestCase): - @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot") @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_default_generation(self, mock_get_devicelist, - mock_read_sys_net, - mock_sys_dev_path, m_get_cmdline): + def test_default_generation( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + m_get_cmdline, + ): tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path) + _setup_test( + tmp_dir, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ) network_cfg = net.generate_fallback_config() - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) renderer = eni.Renderer( - {'eni_path': 'interfaces', 'netrules_path': None}) + {"eni_path": "interfaces", "netrules_path": None} + ) renderer.render_network_state(ns, target=render_dir) - self.assertTrue(os.path.exists(os.path.join(render_dir, - 'interfaces'))) - with open(os.path.join(render_dir, 'interfaces')) as fh: + self.assertTrue(os.path.exists(os.path.join(render_dir, "interfaces"))) + with open(os.path.join(render_dir, "interfaces")) as fh: contents = fh.read() expected = """ @@ -4076,62 +4713,74 @@ auto eth0 iface eth0 inet dhcp """ self.assertEqual( - expected, dir2dict(tmp_dir)['/etc/network/interfaces']) + expected, dir2dict(tmp_dir)["/etc/network/interfaces"] + ) def test_v2_route_metric_to_eni(self): """Network v2 route-metric overrides are preserved in eni output""" tmp_dir = self.tmp_dir() renderer = eni.Renderer() - expected_tmpl = textwrap.dedent("""\ + expected_tmpl = textwrap.dedent( + """\ auto lo iface lo inet loopback auto eth0 iface eth0 inet{suffix} dhcp metric 100 - """) - for dhcp_ver in ('dhcp4', 'dhcp6'): - suffix = '6' if dhcp_ver == 'dhcp6' else '' + """ + ) + for dhcp_ver in ("dhcp4", "dhcp6"): + suffix = "6" if dhcp_ver == "dhcp6" else "" dhcp_cfg = { dhcp_ver: True, - '{ver}-overrides'.format(ver=dhcp_ver): {'route-metric': 100}} - v2_input = {'version': 2, 'ethernets': {'eth0': dhcp_cfg}} + "{ver}-overrides".format(ver=dhcp_ver): {"route-metric": 100}, + } + v2_input = {"version": 2, "ethernets": {"eth0": dhcp_cfg}} ns = network_state.parse_net_config_data(v2_input) renderer.render_network_state(ns, target=tmp_dir) self.assertEqual( expected_tmpl.format(suffix=suffix), - dir2dict(tmp_dir)['/etc/network/interfaces']) + dir2dict(tmp_dir)["/etc/network/interfaces"], + ) class TestNetplanNetRendering(CiTestCase): - @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot") @mock.patch("cloudinit.net.netplan._clean_default") @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_default_generation(self, mock_get_devicelist, - mock_read_sys_net, - mock_sys_dev_path, - mock_clean_default, m_get_cmdline): + def test_default_generation( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + mock_clean_default, + m_get_cmdline, + ): tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path) + _setup_test( + tmp_dir, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ) network_cfg = net.generate_fallback_config() - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) - render_target = 'netplan.yaml' + render_target = "netplan.yaml" renderer = netplan.Renderer( - {'netplan_path': render_target, 'postcmds': False}) + {"netplan_path": render_target, "postcmds": False} + ) renderer.render_network_state(ns, target=render_dir) - self.assertTrue(os.path.exists(os.path.join(render_dir, - render_target))) + self.assertTrue( + os.path.exists(os.path.join(render_dir, render_target)) + ) with open(os.path.join(render_dir, render_target)) as fh: contents = fh.read() print(contents) @@ -4151,8 +4800,9 @@ network: class TestNetplanCleanDefault(CiTestCase): - snapd_known_path = 'etc/netplan/00-snapd-config.yaml' - snapd_known_content = textwrap.dedent("""\ + snapd_known_path = "etc/netplan/00-snapd-config.yaml" + snapd_known_content = textwrap.dedent( + """\ # This is the initial network config. # It can be overwritten by cloud-init or console-conf. network: @@ -4166,15 +4816,18 @@ class TestNetplanCleanDefault(CiTestCase): match: name: "eth*" dhcp4: true - """) + """ + ) stub_known = { - 'run/systemd/network/10-netplan-all-en.network': 'foo-en', - 'run/systemd/network/10-netplan-all-eth.network': 'foo-eth', - 'run/systemd/generator/netplan.stamp': 'stamp', + "run/systemd/network/10-netplan-all-en.network": "foo-en", + "run/systemd/network/10-netplan-all-eth.network": "foo-eth", + "run/systemd/generator/netplan.stamp": "stamp", } def test_clean_known_config_cleaned(self): - content = {self.snapd_known_path: self.snapd_known_content, } + content = { + self.snapd_known_path: self.snapd_known_content, + } content.update(self.stub_known) tmpd = self.tmp_dir() files = sorted(populate_dir(tmpd, content)) @@ -4183,7 +4836,9 @@ class TestNetplanCleanDefault(CiTestCase): self.assertEqual([], found) def test_clean_unknown_config_not_cleaned(self): - content = {self.snapd_known_path: self.snapd_known_content, } + content = { + self.snapd_known_path: self.snapd_known_content, + } content.update(self.stub_known) content[self.snapd_known_path] += "# user put a comment\n" tmpd = self.tmp_dir() @@ -4214,78 +4869,100 @@ class TestNetplanCleanDefault(CiTestCase): class TestNetplanPostcommands(CiTestCase): mycfg = { - 'config': [{"type": "physical", "name": "eth0", - "mac_address": "c0:d6:9f:2c:e8:80", - "subnets": [{"type": "dhcp"}]}], - 'version': 1} - - @mock.patch.object(netplan.Renderer, '_netplan_generate') - @mock.patch.object(netplan.Renderer, '_net_setup_link') - @mock.patch('cloudinit.subp.subp') - def test_netplan_render_calls_postcmds(self, mock_subp, - mock_netplan_generate, - mock_net_setup_link): + "config": [ + { + "type": "physical", + "name": "eth0", + "mac_address": "c0:d6:9f:2c:e8:80", + "subnets": [{"type": "dhcp"}], + } + ], + "version": 1, + } + + @mock.patch.object(netplan.Renderer, "_netplan_generate") + @mock.patch.object(netplan.Renderer, "_net_setup_link") + @mock.patch("cloudinit.subp.subp") + def test_netplan_render_calls_postcmds( + self, mock_subp, mock_netplan_generate, mock_net_setup_link + ): tmp_dir = self.tmp_dir() - ns = network_state.parse_net_config_data(self.mycfg, - skip_broken=False) + ns = network_state.parse_net_config_data(self.mycfg, skip_broken=False) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) - render_target = 'netplan.yaml' + render_target = "netplan.yaml" renderer = netplan.Renderer( - {'netplan_path': render_target, 'postcmds': True}) + {"netplan_path": render_target, "postcmds": True} + ) mock_subp.side_effect = iter([subp.ProcessExecutionError]) renderer.render_network_state(ns, target=render_dir) mock_netplan_generate.assert_called_with(run=True) mock_net_setup_link.assert_called_with(run=True) - @mock.patch('cloudinit.util.SeLinuxGuard') + @mock.patch("cloudinit.util.SeLinuxGuard") @mock.patch.object(netplan, "get_devicelist") - @mock.patch('cloudinit.subp.subp') + @mock.patch("cloudinit.subp.subp") def test_netplan_postcmds(self, mock_subp, mock_devlist, mock_sel): mock_sel.__enter__ = mock.Mock(return_value=False) mock_sel.__exit__ = mock.Mock() - mock_devlist.side_effect = [['lo']] + mock_devlist.side_effect = [["lo"]] tmp_dir = self.tmp_dir() - ns = network_state.parse_net_config_data(self.mycfg, - skip_broken=False) + ns = network_state.parse_net_config_data(self.mycfg, skip_broken=False) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) - render_target = 'netplan.yaml' + render_target = "netplan.yaml" renderer = netplan.Renderer( - {'netplan_path': render_target, 'postcmds': True}) - mock_subp.side_effect = iter([ - subp.ProcessExecutionError, - ('', ''), - ('', ''), - ]) + {"netplan_path": render_target, "postcmds": True} + ) + mock_subp.side_effect = iter( + [ + subp.ProcessExecutionError, + ("", ""), + ("", ""), + ] + ) expected = [ - mock.call(['netplan', 'info'], capture=True), - mock.call(['netplan', 'generate'], capture=True), - mock.call(['udevadm', 'test-builtin', 'net_setup_link', - '/sys/class/net/lo'], capture=True), + mock.call(["netplan", "info"], capture=True), + mock.call(["netplan", "generate"], capture=True), + mock.call( + [ + "udevadm", + "test-builtin", + "net_setup_link", + "/sys/class/net/lo", + ], + capture=True, + ), ] - with mock.patch.object(os.path, 'islink', return_value=True): + with mock.patch.object(os.path, "islink", return_value=True): renderer.render_network_state(ns, target=render_dir) mock_subp.assert_has_calls(expected) class TestEniNetworkStateToEni(CiTestCase): mycfg = { - 'config': [{"type": "physical", "name": "eth0", - "mac_address": "c0:d6:9f:2c:e8:80", - "subnets": [{"type": "dhcp"}]}], - 'version': 1} - my_mac = 'c0:d6:9f:2c:e8:80' + "config": [ + { + "type": "physical", + "name": "eth0", + "mac_address": "c0:d6:9f:2c:e8:80", + "subnets": [{"type": "dhcp"}], + } + ], + "version": 1, + } + my_mac = "c0:d6:9f:2c:e8:80" def test_no_header(self): rendered = eni.network_state_to_eni( network_state=network_state.parse_net_config_data(self.mycfg), - render_hwaddress=True) + render_hwaddress=True, + ) self.assertIn(self.my_mac, rendered) self.assertIn("hwaddress", rendered) @@ -4293,14 +4970,17 @@ class TestEniNetworkStateToEni(CiTestCase): header = "# hello world\n" rendered = eni.network_state_to_eni( network_state=network_state.parse_net_config_data(self.mycfg), - header=header, render_hwaddress=True) + header=header, + render_hwaddress=True, + ) self.assertIn(header, rendered) self.assertIn(self.my_mac, rendered) def test_no_hwaddress(self): rendered = eni.network_state_to_eni( network_state=network_state.parse_net_config_data(self.mycfg), - render_hwaddress=False) + render_hwaddress=False, + ) self.assertNotIn(self.my_mac, rendered) self.assertNotIn("hwaddress", rendered) @@ -4309,189 +4989,218 @@ class TestCmdlineConfigParsing(CiTestCase): with_logs = True simple_cfg = { - 'config': [{"type": "physical", "name": "eth0", - "mac_address": "c0:d6:9f:2c:e8:80", - "subnets": [{"type": "dhcp"}]}]} + "config": [ + { + "type": "physical", + "name": "eth0", + "mac_address": "c0:d6:9f:2c:e8:80", + "subnets": [{"type": "dhcp"}], + } + ] + } def test_cmdline_convert_dhcp(self): found = cmdline._klibc_to_config_entry(DHCP_CONTENT_1) - self.assertEqual(found, ('eth0', DHCP_EXPECTED_1)) + self.assertEqual(found, ("eth0", DHCP_EXPECTED_1)) def test_cmdline_convert_dhcp6(self): found = cmdline._klibc_to_config_entry(DHCP6_CONTENT_1) - self.assertEqual(found, ('eno1', DHCP6_EXPECTED_1)) + self.assertEqual(found, ("eno1", DHCP6_EXPECTED_1)) def test_cmdline_convert_static(self): found = cmdline._klibc_to_config_entry(STATIC_CONTENT_1) - self.assertEqual(found, ('eth1', STATIC_EXPECTED_1)) + self.assertEqual(found, ("eth1", STATIC_EXPECTED_1)) def test_config_from_cmdline_net_cfg(self): files = [] - pairs = (('net-eth0.cfg', DHCP_CONTENT_1), - ('net-eth1.cfg', STATIC_CONTENT_1)) + pairs = ( + ("net-eth0.cfg", DHCP_CONTENT_1), + ("net-eth1.cfg", STATIC_CONTENT_1), + ) - macs = {'eth1': 'b8:ae:ed:75:ff:2b', - 'eth0': 'b8:ae:ed:75:ff:2a'} + macs = {"eth1": "b8:ae:ed:75:ff:2b", "eth0": "b8:ae:ed:75:ff:2a"} dhcp = copy.deepcopy(DHCP_EXPECTED_1) - dhcp['mac_address'] = macs['eth0'] + dhcp["mac_address"] = macs["eth0"] static = copy.deepcopy(STATIC_EXPECTED_1) - static['mac_address'] = macs['eth1'] + static["mac_address"] = macs["eth1"] - expected = {'version': 1, 'config': [dhcp, static]} + expected = {"version": 1, "config": [dhcp, static]} with temp_utils.tempdir() as tmpd: for fname, content in pairs: fp = os.path.join(tmpd, fname) files.append(fp) util.write_file(fp, content) - found = cmdline.config_from_klibc_net_cfg(files=files, - mac_addrs=macs) + found = cmdline.config_from_klibc_net_cfg( + files=files, mac_addrs=macs + ) self.assertEqual(found, expected) def test_cmdline_with_b64(self): data = base64.b64encode(json.dumps(self.simple_cfg).encode()) encoded_text = data.decode() - raw_cmdline = 'ro network-config=' + encoded_text + ' root=foo' + raw_cmdline = "ro network-config=" + encoded_text + " root=foo" found = cmdline.read_kernel_cmdline_config(cmdline=raw_cmdline) self.assertEqual(found, self.simple_cfg) def test_cmdline_with_net_config_disabled(self): - raw_cmdline = 'ro network-config=disabled root=foo' + raw_cmdline = "ro network-config=disabled root=foo" found = cmdline.read_kernel_cmdline_config(cmdline=raw_cmdline) - self.assertEqual(found, {'config': 'disabled'}) + self.assertEqual(found, {"config": "disabled"}) def test_cmdline_with_net_config_unencoded_logs_error(self): """network-config cannot be unencoded besides 'disabled'.""" - raw_cmdline = 'ro network-config={config:disabled} root=foo' + raw_cmdline = "ro network-config={config:disabled} root=foo" found = cmdline.read_kernel_cmdline_config(cmdline=raw_cmdline) self.assertIsNone(found) expected_log = ( - 'ERROR: Expected base64 encoded kernel commandline parameter' - ' network-config. Ignoring network-config={config:disabled}.') + "ERROR: Expected base64 encoded kernel commandline parameter" + " network-config. Ignoring network-config={config:disabled}." + ) self.assertIn(expected_log, self.logs.getvalue()) def test_cmdline_with_b64_gz(self): data = _gzip_data(json.dumps(self.simple_cfg).encode()) encoded_text = base64.b64encode(data).decode() - raw_cmdline = 'ro network-config=' + encoded_text + ' root=foo' + raw_cmdline = "ro network-config=" + encoded_text + " root=foo" found = cmdline.read_kernel_cmdline_config(cmdline=raw_cmdline) self.assertEqual(found, self.simple_cfg) class TestCmdlineKlibcNetworkConfigSource(FilesystemMockingTestCase): macs = { - 'eth0': '14:02:ec:42:48:00', - 'eno1': '14:02:ec:42:48:01', + "eth0": "14:02:ec:42:48:00", + "eno1": "14:02:ec:42:48:01", } def test_without_ip(self): - content = {'/run/net-eth0.conf': DHCP_CONTENT_1, - cmdline._OPEN_ISCSI_INTERFACE_FILE: "eth0\n"} + content = { + "/run/net-eth0.conf": DHCP_CONTENT_1, + cmdline._OPEN_ISCSI_INTERFACE_FILE: "eth0\n", + } exp1 = copy.deepcopy(DHCP_EXPECTED_1) - exp1['mac_address'] = self.macs['eth0'] + exp1["mac_address"] = self.macs["eth0"] root = self.tmp_dir() populate_dir(root, content) self.reRoot(root) src = cmdline.KlibcNetworkConfigSource( - _cmdline='foo root=/root/bar', _mac_addrs=self.macs, + _cmdline="foo root=/root/bar", + _mac_addrs=self.macs, ) self.assertTrue(src.is_applicable()) found = src.render_config() - self.assertEqual(found['version'], 1) - self.assertEqual(found['config'], [exp1]) + self.assertEqual(found["version"], 1) + self.assertEqual(found["config"], [exp1]) def test_with_ip(self): - content = {'/run/net-eth0.conf': DHCP_CONTENT_1} + content = {"/run/net-eth0.conf": DHCP_CONTENT_1} exp1 = copy.deepcopy(DHCP_EXPECTED_1) - exp1['mac_address'] = self.macs['eth0'] + exp1["mac_address"] = self.macs["eth0"] root = self.tmp_dir() populate_dir(root, content) self.reRoot(root) src = cmdline.KlibcNetworkConfigSource( - _cmdline='foo ip=dhcp', _mac_addrs=self.macs, + _cmdline="foo ip=dhcp", + _mac_addrs=self.macs, ) self.assertTrue(src.is_applicable()) found = src.render_config() - self.assertEqual(found['version'], 1) - self.assertEqual(found['config'], [exp1]) + self.assertEqual(found["version"], 1) + self.assertEqual(found["config"], [exp1]) def test_with_ip6(self): - content = {'/run/net6-eno1.conf': DHCP6_CONTENT_1} + content = {"/run/net6-eno1.conf": DHCP6_CONTENT_1} root = self.tmp_dir() populate_dir(root, content) self.reRoot(root) src = cmdline.KlibcNetworkConfigSource( - _cmdline='foo ip6=dhcp root=/dev/sda', _mac_addrs=self.macs, + _cmdline="foo ip6=dhcp root=/dev/sda", + _mac_addrs=self.macs, ) self.assertTrue(src.is_applicable()) found = src.render_config() self.assertEqual( found, - {'version': 1, 'config': [ - {'type': 'physical', 'name': 'eno1', - 'mac_address': self.macs['eno1'], - 'subnets': [ - {'dns_nameservers': ['2001:67c:1562:8010::2:1'], - 'control': 'manual', 'type': 'dhcp6', 'netmask': '64'}]}]}) + { + "version": 1, + "config": [ + { + "type": "physical", + "name": "eno1", + "mac_address": self.macs["eno1"], + "subnets": [ + { + "dns_nameservers": ["2001:67c:1562:8010::2:1"], + "control": "manual", + "type": "dhcp6", + "netmask": "64", + } + ], + } + ], + }, + ) def test_with_no_ip_or_ip6(self): # if there is no ip= or ip6= on cmdline, return value should be None - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( - _files=files, _cmdline='foo root=/dev/sda', _mac_addrs=self.macs, + _files=files, + _cmdline="foo root=/dev/sda", + _mac_addrs=self.macs, ) self.assertFalse(src.is_applicable()) def test_with_faux_ip(self): - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( _files=files, - _cmdline='foo iscsi_target_ip=root=/dev/sda', + _cmdline="foo iscsi_target_ip=root=/dev/sda", _mac_addrs=self.macs, ) self.assertFalse(src.is_applicable()) def test_empty_cmdline(self): - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( _files=files, - _cmdline='', + _cmdline="", _mac_addrs=self.macs, ) self.assertFalse(src.is_applicable()) def test_whitespace_cmdline(self): - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( _files=files, - _cmdline=' ', + _cmdline=" ", _mac_addrs=self.macs, ) self.assertFalse(src.is_applicable()) def test_cmdline_no_lhand(self): - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( _files=files, - _cmdline='=wut', + _cmdline="=wut", _mac_addrs=self.macs, ) self.assertFalse(src.is_applicable()) def test_cmdline_embedded_ip(self): - content = {'net6-eno1.conf': DHCP6_CONTENT_1} + content = {"net6-eno1.conf": DHCP6_CONTENT_1} files = sorted(populate_dir(self.tmp_dir(), content)) src = cmdline.KlibcNetworkConfigSource( _files=files, @@ -4502,13 +5211,19 @@ class TestCmdlineKlibcNetworkConfigSource(FilesystemMockingTestCase): def test_with_both_ip_ip6(self): content = { - '/run/net-eth0.conf': DHCP_CONTENT_1, - '/run/net6-eth0.conf': DHCP6_CONTENT_1.replace('eno1', 'eth0')} + "/run/net-eth0.conf": DHCP_CONTENT_1, + "/run/net6-eth0.conf": DHCP6_CONTENT_1.replace("eno1", "eth0"), + } eth0 = copy.deepcopy(DHCP_EXPECTED_1) - eth0['mac_address'] = self.macs['eth0'] - eth0['subnets'].append( - {'control': 'manual', 'type': 'dhcp6', - 'netmask': '64', 'dns_nameservers': ['2001:67c:1562:8010::2:1']}) + eth0["mac_address"] = self.macs["eth0"] + eth0["subnets"].append( + { + "control": "manual", + "type": "dhcp6", + "netmask": "64", + "dns_nameservers": ["2001:67c:1562:8010::2:1"], + } + ) expected = [eth0] root = self.tmp_dir() @@ -4516,17 +5231,17 @@ class TestCmdlineKlibcNetworkConfigSource(FilesystemMockingTestCase): self.reRoot(root) src = cmdline.KlibcNetworkConfigSource( - _cmdline='foo ip=dhcp ip6=dhcp', _mac_addrs=self.macs, + _cmdline="foo ip=dhcp ip6=dhcp", + _mac_addrs=self.macs, ) self.assertTrue(src.is_applicable()) found = src.render_config() - self.assertEqual(found['version'], 1) - self.assertEqual(found['config'], expected) + self.assertEqual(found["version"], 1) + self.assertEqual(found["config"], expected) class TestReadInitramfsConfig(CiTestCase): - def _config_source_cls_mock(self, is_applicable, render_config=None): return lambda: mock.Mock( is_applicable=lambda: is_applicable, @@ -4534,7 +5249,7 @@ class TestReadInitramfsConfig(CiTestCase): ) def test_no_sources(self): - with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', []): + with mock.patch("cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES", []): self.assertIsNone(cmdline.read_initramfs_config()) def test_no_applicable_sources(self): @@ -4543,19 +5258,22 @@ class TestReadInitramfsConfig(CiTestCase): self._config_source_cls_mock(is_applicable=False), self._config_source_cls_mock(is_applicable=False), ] - with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', - sources): + with mock.patch( + "cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES", sources + ): self.assertIsNone(cmdline.read_initramfs_config()) def test_one_applicable_source(self): expected_config = object() sources = [ self._config_source_cls_mock( - is_applicable=True, render_config=expected_config, + is_applicable=True, + render_config=expected_config, ), ] - with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', - sources): + with mock.patch( + "cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES", sources + ): self.assertEqual(expected_config, cmdline.read_initramfs_config()) def test_one_applicable_source_after_inapplicable_sources(self): @@ -4564,45 +5282,53 @@ class TestReadInitramfsConfig(CiTestCase): self._config_source_cls_mock(is_applicable=False), self._config_source_cls_mock(is_applicable=False), self._config_source_cls_mock( - is_applicable=True, render_config=expected_config, + is_applicable=True, + render_config=expected_config, ), ] - with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', - sources): + with mock.patch( + "cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES", sources + ): self.assertEqual(expected_config, cmdline.read_initramfs_config()) def test_first_applicable_source_is_used(self): first_config, second_config = object(), object() sources = [ self._config_source_cls_mock( - is_applicable=True, render_config=first_config, + is_applicable=True, + render_config=first_config, ), self._config_source_cls_mock( - is_applicable=True, render_config=second_config, + is_applicable=True, + render_config=second_config, ), ] - with mock.patch('cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES', - sources): + with mock.patch( + "cloudinit.net.cmdline._INITRAMFS_CONFIG_SOURCES", sources + ): self.assertEqual(first_config, cmdline.read_initramfs_config()) class TestNetplanRoundTrip(CiTestCase): - NETPLAN_INFO_OUT = textwrap.dedent(""" + NETPLAN_INFO_OUT = textwrap.dedent( + """ netplan.io: features: - dhcp-use-domains - ipv6-mtu website: https://netplan.io/ - """) + """ + ) def setUp(self): super(TestNetplanRoundTrip, self).setUp() - self.add_patch('cloudinit.net.netplan.subp.subp', 'm_subp') - self.m_subp.return_value = (self.NETPLAN_INFO_OUT, '') + self.add_patch("cloudinit.net.netplan.subp.subp", "m_subp") + self.m_subp.return_value = (self.NETPLAN_INFO_OUT, "") - def _render_and_read(self, network_config=None, state=None, - netplan_path=None, target=None): + def _render_and_read( + self, network_config=None, state=None, netplan_path=None, target=None + ): if target is None: target = self.tmp_dir() @@ -4614,188 +5340,212 @@ class TestNetplanRoundTrip(CiTestCase): raise ValueError("Expected data or state, got neither") if netplan_path is None: - netplan_path = 'etc/netplan/50-cloud-init.yaml' + netplan_path = "etc/netplan/50-cloud-init.yaml" - renderer = netplan.Renderer( - config={'netplan_path': netplan_path}) + renderer = netplan.Renderer(config={"netplan_path": netplan_path}) renderer.render_network_state(ns, target=target) return dir2dict(target) def testsimple_render_bond_netplan(self): - entry = NETWORK_CONFIGS['bond'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) - print(entry['expected_netplan']) - print('-- expected ^ | v rendered --') - print(files['/etc/netplan/50-cloud-init.yaml']) + entry = NETWORK_CONFIGS["bond"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) + print(entry["expected_netplan"]) + print("-- expected ^ | v rendered --") + print(files["/etc/netplan/50-cloud-init.yaml"]) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_bond_v2_input_netplan(self): - entry = NETWORK_CONFIGS['bond'] + entry = NETWORK_CONFIGS["bond"] files = self._render_and_read( - network_config=yaml.load(entry['yaml-v2'])) - print(entry['expected_netplan-v2']) - print('-- expected ^ | v rendered --') - print(files['/etc/netplan/50-cloud-init.yaml']) + network_config=yaml.load(entry["yaml-v2"]) + ) + print(entry["expected_netplan-v2"]) + print("-- expected ^ | v rendered --") + print(files["/etc/netplan/50-cloud-init.yaml"]) self.assertEqual( - entry['expected_netplan-v2'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan-v2"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_small_netplan(self): - entry = NETWORK_CONFIGS['small'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["small"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_v4_and_v6(self): - entry = NETWORK_CONFIGS['v4_and_v6'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_v4_and_v6_static(self): - entry = NETWORK_CONFIGS['v4_and_v6_static'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6_static"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_dhcpv6_only(self): - entry = NETWORK_CONFIGS['dhcpv6_only'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_only"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_dhcpv6_accept_ra(self): - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_dhcpv6_reject_ra(self): - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_ipv6_slaac(self): - entry = NETWORK_CONFIGS['ipv6_slaac'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml'])) + entry = NETWORK_CONFIGS["ipv6_slaac"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_dhcpv6_stateless(self): - entry = NETWORK_CONFIGS['dhcpv6_stateless'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateless"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_dhcpv6_stateful(self): - entry = NETWORK_CONFIGS['dhcpv6_stateful'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateful"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_wakeonlan_disabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_disabled'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_disabled"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_wakeonlan_enabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_enabled'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_enabled"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_all(self): - entry = NETWORK_CONFIGS['all'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) - print(entry['expected_netplan']) - print('-- expected ^ | v rendered --') - print(files['/etc/netplan/50-cloud-init.yaml']) + entry = NETWORK_CONFIGS["all"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) + print(entry["expected_netplan"]) + print("-- expected ^ | v rendered --") + print(files["/etc/netplan/50-cloud-init.yaml"]) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def testsimple_render_manual(self): - entry = NETWORK_CONFIGS['manual'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["manual"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def test_render_output_has_yaml_no_aliases(self): entry = { - 'yaml': V1_NAMESERVER_ALIAS, - 'expected_netplan': NETPLAN_NO_ALIAS, + "yaml": V1_NAMESERVER_ALIAS, + "expected_netplan": NETPLAN_NO_ALIAS, } - network_config = yaml.load(entry['yaml']) + network_config = yaml.load(entry["yaml"]) ns = network_state.parse_net_config_data(network_config) files = self._render_and_read(state=ns) # check for alias - content = files['/etc/netplan/50-cloud-init.yaml'] + content = files["/etc/netplan/50-cloud-init.yaml"] # test load the yaml to ensure we don't render something not loadable # this allows single aliases, but not duplicate ones - parsed = yaml.load(files['/etc/netplan/50-cloud-init.yaml']) + parsed = yaml.load(files["/etc/netplan/50-cloud-init.yaml"]) self.assertNotEqual(None, parsed) # now look for any alias, avoid rendering them entirely # generate the first anchor string using the template # as of this writing, looks like "&id001" - anchor = r'&' + Serializer.ANCHOR_TEMPLATE % 1 + anchor = r"&" + Serializer.ANCHOR_TEMPLATE % 1 found_alias = re.search(anchor, content, re.MULTILINE) if found_alias: msg = "Error at: %s\nContent:\n%s" % (found_alias, content) - raise ValueError('Found yaml alias in rendered netplan: ' + msg) + raise ValueError("Found yaml alias in rendered netplan: " + msg) - print(entry['expected_netplan']) - print('-- expected ^ | v rendered --') - print(files['/etc/netplan/50-cloud-init.yaml']) + print(entry["expected_netplan"]) + print("-- expected ^ | v rendered --") + print(files["/etc/netplan/50-cloud-init.yaml"]) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) def test_render_output_supports_both_grat_arp_spelling(self): entry = { - 'yaml': NETPLAN_BOND_GRAT_ARP, - 'expected_netplan': NETPLAN_BOND_GRAT_ARP.replace('gratuitous', - 'gratuitious'), + "yaml": NETPLAN_BOND_GRAT_ARP, + "expected_netplan": NETPLAN_BOND_GRAT_ARP.replace( + "gratuitous", "gratuitious" + ), } - network_config = yaml.load(entry['yaml']).get('network') + network_config = yaml.load(entry["yaml"]).get("network") files = self._render_and_read(network_config=network_config) - print(entry['expected_netplan']) - print('-- expected ^ | v rendered --') - print(files['/etc/netplan/50-cloud-init.yaml']) + print(entry["expected_netplan"]) + print("-- expected ^ | v rendered --") + print(files["/etc/netplan/50-cloud-init.yaml"]) self.assertEqual( - entry['expected_netplan'].splitlines(), - files['/etc/netplan/50-cloud-init.yaml'].splitlines()) + entry["expected_netplan"].splitlines(), + files["/etc/netplan/50-cloud-init.yaml"].splitlines(), + ) class TestEniRoundTrip(CiTestCase): - - def _render_and_read(self, network_config=None, state=None, eni_path=None, - netrules_path=None, dir=None): + def _render_and_read( + self, + network_config=None, + state=None, + eni_path=None, + netrules_path=None, + dir=None, + ): if dir is None: dir = self.tmp_dir() @@ -4807,10 +5557,11 @@ class TestEniRoundTrip(CiTestCase): raise ValueError("Expected data or state, got neither") if eni_path is None: - eni_path = 'etc/network/interfaces' + eni_path = "etc/network/interfaces" renderer = eni.Renderer( - config={'eni_path': eni_path, 'netrules_path': netrules_path}) + config={"eni_path": eni_path, "netrules_path": netrules_path} + ) renderer.render_network_state(ns, target=dir) return dir2dict(dir) @@ -4820,95 +5571,112 @@ class TestEniRoundTrip(CiTestCase): files = self._render_and_read(network_config=network_config) self.assertEqual( RENDERED_ENI.splitlines(), - files['/etc/network/interfaces'].splitlines()) + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_all(self): - entry = NETWORK_CONFIGS['all'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["all"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_small(self): - entry = NETWORK_CONFIGS['small'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["small"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_v4_and_v6(self): - entry = NETWORK_CONFIGS['v4_and_v6'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_dhcpv6_only(self): - entry = NETWORK_CONFIGS['dhcpv6_only'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_only"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_v4_and_v6_static(self): - entry = NETWORK_CONFIGS['v4_and_v6_static'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["v4_and_v6_static"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_dhcpv6_stateless(self): - entry = NETWORK_CONFIGS['dhcpv6_stateless'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateless"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_ipv6_slaac(self): - entry = NETWORK_CONFIGS['ipv6_slaac'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["ipv6_slaac"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_dhcpv6_stateful(self): - entry = NETWORK_CONFIGS['dhcpv6_stateless'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["dhcpv6_stateless"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_dhcpv6_accept_ra(self): - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_dhcpv6_reject_ra(self): - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_wakeonlan_disabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_disabled'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_disabled"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_wakeonlan_enabled_config_v2(self): - entry = NETWORK_CONFIGS['wakeonlan_enabled'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + entry = NETWORK_CONFIGS["wakeonlan_enabled"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def testsimple_render_manual(self): """Test rendering of 'manual' for 'type' and 'control'. @@ -4918,145 +5686,179 @@ class TestEniRoundTrip(CiTestCase): if there were no addresses to configure. Also strange is the fact that in order to apply that MTU the ifupdown device must be set to 'auto', or the MTU would not be set.""" - entry = NETWORK_CONFIGS['manual'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["manual"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) def test_routes_rendered(self): # as reported in bug 1649652 conf = [ - {'name': 'eth0', 'type': 'physical', - 'subnets': [{ - 'address': '172.23.31.42/26', - 'dns_nameservers': [], 'gateway': '172.23.31.2', - 'type': 'static'}]}, - {'type': 'route', 'id': 4, - 'metric': 0, 'destination': '10.0.0.0/12', - 'gateway': '172.23.31.1'}, - {'type': 'route', 'id': 5, - 'metric': 0, 'destination': '192.168.2.0/16', - 'gateway': '172.23.31.1'}, - {'type': 'route', 'id': 6, - 'metric': 1, 'destination': '10.0.200.0/16', - 'gateway': '172.23.31.1'}, - {'type': 'route', 'id': 7, - 'metric': 1, 'destination': '10.0.0.100/32', - 'gateway': '172.23.31.1'}, + { + "name": "eth0", + "type": "physical", + "subnets": [ + { + "address": "172.23.31.42/26", + "dns_nameservers": [], + "gateway": "172.23.31.2", + "type": "static", + } + ], + }, + { + "type": "route", + "id": 4, + "metric": 0, + "destination": "10.0.0.0/12", + "gateway": "172.23.31.1", + }, + { + "type": "route", + "id": 5, + "metric": 0, + "destination": "192.168.2.0/16", + "gateway": "172.23.31.1", + }, + { + "type": "route", + "id": 6, + "metric": 1, + "destination": "10.0.200.0/16", + "gateway": "172.23.31.1", + }, + { + "type": "route", + "id": 7, + "metric": 1, + "destination": "10.0.0.100/32", + "gateway": "172.23.31.1", + }, ] files = self._render_and_read( - network_config={'config': conf, 'version': 1}) + network_config={"config": conf, "version": 1} + ) expected = [ - 'auto lo', - 'iface lo inet loopback', - 'auto eth0', - 'iface eth0 inet static', - ' address 172.23.31.42/26', - ' gateway 172.23.31.2', - ('post-up route add -net 10.0.0.0/12 gw ' - '172.23.31.1 metric 0 || true'), - ('pre-down route del -net 10.0.0.0/12 gw ' - '172.23.31.1 metric 0 || true'), - ('post-up route add -net 192.168.2.0/16 gw ' - '172.23.31.1 metric 0 || true'), - ('pre-down route del -net 192.168.2.0/16 gw ' - '172.23.31.1 metric 0 || true'), - ('post-up route add -net 10.0.200.0/16 gw ' - '172.23.31.1 metric 1 || true'), - ('pre-down route del -net 10.0.200.0/16 gw ' - '172.23.31.1 metric 1 || true'), - ('post-up route add -host 10.0.0.100/32 gw ' - '172.23.31.1 metric 1 || true'), - ('pre-down route del -host 10.0.0.100/32 gw ' - '172.23.31.1 metric 1 || true'), + "auto lo", + "iface lo inet loopback", + "auto eth0", + "iface eth0 inet static", + " address 172.23.31.42/26", + " gateway 172.23.31.2", + "post-up route add -net 10.0.0.0/12 gw " + "172.23.31.1 metric 0 || true", + "pre-down route del -net 10.0.0.0/12 gw " + "172.23.31.1 metric 0 || true", + "post-up route add -net 192.168.2.0/16 gw " + "172.23.31.1 metric 0 || true", + "pre-down route del -net 192.168.2.0/16 gw " + "172.23.31.1 metric 0 || true", + "post-up route add -net 10.0.200.0/16 gw " + "172.23.31.1 metric 1 || true", + "pre-down route del -net 10.0.200.0/16 gw " + "172.23.31.1 metric 1 || true", + "post-up route add -host 10.0.0.100/32 gw " + "172.23.31.1 metric 1 || true", + "pre-down route del -host 10.0.0.100/32 gw " + "172.23.31.1 metric 1 || true", ] - found = files['/etc/network/interfaces'].splitlines() + found = files["/etc/network/interfaces"].splitlines() - self.assertEqual( - expected, [line for line in found if line]) + self.assertEqual(expected, [line for line in found if line]) def test_ipv6_static_routes(self): # as reported in bug 1818669 conf = [ - {'name': 'eno3', 'type': 'physical', - 'subnets': [{ - 'address': 'fd00::12/64', - 'dns_nameservers': ['fd00:2::15'], - 'gateway': 'fd00::1', - 'ipv6': True, - 'type': 'static', - 'routes': [{'netmask': '32', - 'network': 'fd00:12::', - 'gateway': 'fd00::2'}, - {'network': 'fd00:14::', - 'gateway': 'fd00::3'}, - {'destination': 'fe00:14::/48', - 'gateway': 'fe00::4', - 'metric': 500}, - {'gateway': '192.168.23.1', - 'metric': 999, - 'netmask': 24, - 'network': '192.168.23.0'}, - {'destination': '10.23.23.0/24', - 'gateway': '10.23.23.2', - 'metric': 300}]}]}, + { + "name": "eno3", + "type": "physical", + "subnets": [ + { + "address": "fd00::12/64", + "dns_nameservers": ["fd00:2::15"], + "gateway": "fd00::1", + "ipv6": True, + "type": "static", + "routes": [ + { + "netmask": "32", + "network": "fd00:12::", + "gateway": "fd00::2", + }, + {"network": "fd00:14::", "gateway": "fd00::3"}, + { + "destination": "fe00:14::/48", + "gateway": "fe00::4", + "metric": 500, + }, + { + "gateway": "192.168.23.1", + "metric": 999, + "netmask": 24, + "network": "192.168.23.0", + }, + { + "destination": "10.23.23.0/24", + "gateway": "10.23.23.2", + "metric": 300, + }, + ], + } + ], + }, ] files = self._render_and_read( - network_config={'config': conf, 'version': 1}) + network_config={"config": conf, "version": 1} + ) expected = [ - 'auto lo', - 'iface lo inet loopback', - 'auto eno3', - 'iface eno3 inet6 static', - ' address fd00::12/64', - ' dns-nameservers fd00:2::15', - ' gateway fd00::1', - (' post-up route add -A inet6 fd00:12::/32 gw ' - 'fd00::2 || true'), - (' pre-down route del -A inet6 fd00:12::/32 gw ' - 'fd00::2 || true'), - (' post-up route add -A inet6 fd00:14::/64 gw ' - 'fd00::3 || true'), - (' pre-down route del -A inet6 fd00:14::/64 gw ' - 'fd00::3 || true'), - (' post-up route add -A inet6 fe00:14::/48 gw ' - 'fe00::4 metric 500 || true'), - (' pre-down route del -A inet6 fe00:14::/48 gw ' - 'fe00::4 metric 500 || true'), - (' post-up route add -net 192.168.23.0/24 gw ' - '192.168.23.1 metric 999 || true'), - (' pre-down route del -net 192.168.23.0/24 gw ' - '192.168.23.1 metric 999 || true'), - (' post-up route add -net 10.23.23.0/24 gw ' - '10.23.23.2 metric 300 || true'), - (' pre-down route del -net 10.23.23.0/24 gw ' - '10.23.23.2 metric 300 || true'), - + "auto lo", + "iface lo inet loopback", + "auto eno3", + "iface eno3 inet6 static", + " address fd00::12/64", + " dns-nameservers fd00:2::15", + " gateway fd00::1", + " post-up route add -A inet6 fd00:12::/32 gw fd00::2 || true", + " pre-down route del -A inet6 fd00:12::/32 gw fd00::2 || true", + " post-up route add -A inet6 fd00:14::/64 gw fd00::3 || true", + " pre-down route del -A inet6 fd00:14::/64 gw fd00::3 || true", + " post-up route add -A inet6 fe00:14::/48 gw " + "fe00::4 metric 500 || true", + " pre-down route del -A inet6 fe00:14::/48 gw " + "fe00::4 metric 500 || true", + " post-up route add -net 192.168.23.0/24 gw " + "192.168.23.1 metric 999 || true", + " pre-down route del -net 192.168.23.0/24 gw " + "192.168.23.1 metric 999 || true", + " post-up route add -net 10.23.23.0/24 gw " + "10.23.23.2 metric 300 || true", + " pre-down route del -net 10.23.23.0/24 gw " + "10.23.23.2 metric 300 || true", ] - found = files['/etc/network/interfaces'].splitlines() + found = files["/etc/network/interfaces"].splitlines() - self.assertEqual( - expected, [line for line in found if line]) + self.assertEqual(expected, [line for line in found if line]) def testsimple_render_bond(self): - entry = NETWORK_CONFIGS['bond'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + entry = NETWORK_CONFIGS["bond"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) self.assertEqual( - entry['expected_eni'].splitlines(), - files['/etc/network/interfaces'].splitlines()) + entry["expected_eni"].splitlines(), + files["/etc/network/interfaces"].splitlines(), + ) class TestNetworkdNetRendering(CiTestCase): - def create_conf_dict(self, contents): content_dict = {} for line in contents: if line: line = line.strip() - if line and re.search(r'^\[(.+)\]$', line): + if line and re.search(r"^\[(.+)\]$", line): content_dict[line] = [] key = line elif line: @@ -5073,40 +5875,48 @@ class TestNetworkdNetRendering(CiTestCase): @mock.patch("cloudinit.net.sys_dev_path") @mock.patch("cloudinit.net.read_sys_net") @mock.patch("cloudinit.net.get_devicelist") - def test_networkd_default_generation(self, mock_get_devicelist, - mock_read_sys_net, - mock_sys_dev_path, - m_get_cmdline, - m_chown): + def test_networkd_default_generation( + self, + mock_get_devicelist, + mock_read_sys_net, + mock_sys_dev_path, + m_get_cmdline, + m_chown, + ): tmp_dir = self.tmp_dir() - _setup_test(tmp_dir, mock_get_devicelist, - mock_read_sys_net, mock_sys_dev_path) + _setup_test( + tmp_dir, mock_get_devicelist, mock_read_sys_net, mock_sys_dev_path + ) network_cfg = net.generate_fallback_config() - ns = network_state.parse_net_config_data(network_cfg, - skip_broken=False) + ns = network_state.parse_net_config_data( + network_cfg, skip_broken=False + ) render_dir = os.path.join(tmp_dir, "render") os.makedirs(render_dir) - render_target = 'etc/systemd/network/10-cloud-init-eth1000.network' + render_target = "etc/systemd/network/10-cloud-init-eth1000.network" renderer = networkd.Renderer({}) renderer.render_network_state(ns, target=render_dir) - self.assertTrue(os.path.exists(os.path.join(render_dir, - render_target))) + self.assertTrue( + os.path.exists(os.path.join(render_dir, render_target)) + ) with open(os.path.join(render_dir, render_target)) as fh: contents = fh.readlines() actual = self.create_conf_dict(contents) print(actual) - expected = textwrap.dedent("""\ + expected = textwrap.dedent( + """\ [Match] Name=eth1000 MACAddress=07-1c-c6-75-a4-be [Network] - DHCP=ipv4""").rstrip(' ') + DHCP=ipv4""" + ).rstrip(" ") expected = self.create_conf_dict(expected.splitlines()) @@ -5114,13 +5924,12 @@ class TestNetworkdNetRendering(CiTestCase): class TestNetworkdRoundTrip(CiTestCase): - def create_conf_dict(self, contents): content_dict = {} for line in contents: if line: line = line.strip() - if line and re.search(r'^\[(.+)\]$', line): + if line and re.search(r"^\[(.+)\]$", line): content_dict[line] = [] key = line elif line: @@ -5132,8 +5941,9 @@ class TestNetworkdRoundTrip(CiTestCase): for k, v in actual.items(): self.assertEqual(sorted(expected[k]), sorted(v)) - def _render_and_read(self, network_config=None, state=None, nwkd_path=None, - dir=None): + def _render_and_read( + self, network_config=None, state=None, nwkd_path=None, dir=None + ): if dir is None: dir = self.tmp_dir() @@ -5145,24 +5955,24 @@ class TestNetworkdRoundTrip(CiTestCase): raise ValueError("Expected data or state, got neither") if not nwkd_path: - nwkd_path = '/etc/systemd/network/' + nwkd_path = "/etc/systemd/network/" - renderer = networkd.Renderer(config={'network_conf_dir': nwkd_path}) + renderer = networkd.Renderer(config={"network_conf_dir": nwkd_path}) renderer.render_network_state(ns, target=dir) return dir2dict(dir) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def testsimple_render_small_networkd(self, m_chown): - nwk_fn1 = '/etc/systemd/network/10-cloud-init-eth99.network' - nwk_fn2 = '/etc/systemd/network/10-cloud-init-eth1.network' - entry = NETWORK_CONFIGS['small'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + nwk_fn1 = "/etc/systemd/network/10-cloud-init-eth99.network" + nwk_fn2 = "/etc/systemd/network/10-cloud-init-eth1.network" + entry = NETWORK_CONFIGS["small"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) actual = files[nwk_fn1].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd_eth99'].splitlines() + expected = entry["expected_networkd_eth99"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @@ -5170,135 +5980,146 @@ class TestNetworkdRoundTrip(CiTestCase): actual = files[nwk_fn2].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd_eth1'].splitlines() + expected = entry["expected_networkd_eth1"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def testsimple_render_v4_and_v6(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['v4_and_v6'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["v4_and_v6"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def testsimple_render_v4_and_v6_static(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['v4_and_v6_static'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["v4_and_v6_static"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def testsimple_render_dhcpv6_only(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['dhcpv6_only'] - files = self._render_and_read(network_config=yaml.load(entry['yaml'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["dhcpv6_only"] + files = self._render_and_read(network_config=yaml.load(entry["yaml"])) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def test_dhcpv6_accept_ra_config_v1(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def test_dhcpv6_accept_ra_config_v2(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['dhcpv6_accept_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["dhcpv6_accept_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def test_dhcpv6_reject_ra_config_v1(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v1'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v1"]) + ) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) @mock.patch("cloudinit.net.util.chownbyname", return_value=True) def test_dhcpv6_reject_ra_config_v2(self, m_chown): - nwk_fn = '/etc/systemd/network/10-cloud-init-iface0.network' - entry = NETWORK_CONFIGS['dhcpv6_reject_ra'] - files = self._render_and_read(network_config=yaml.load( - entry['yaml_v2'])) + nwk_fn = "/etc/systemd/network/10-cloud-init-iface0.network" + entry = NETWORK_CONFIGS["dhcpv6_reject_ra"] + files = self._render_and_read( + network_config=yaml.load(entry["yaml_v2"]) + ) actual = files[nwk_fn].splitlines() actual = self.create_conf_dict(actual) - expected = entry['expected_networkd'].splitlines() + expected = entry["expected_networkd"].splitlines() expected = self.create_conf_dict(expected) self.compare_dicts(actual, expected) class TestRenderersSelect: - @pytest.mark.parametrize( - 'renderer_selected,netplan,eni,nm,scfg,sys,networkd', ( + "renderer_selected,netplan,eni,nm,scfg,sys,networkd", + ( # -netplan -ifupdown -nm -scfg -sys raises error - (net.RendererNotFoundError, False, False, False, False, False, - False), + ( + net.RendererNotFoundError, + False, + False, + False, + False, + False, + False, + ), # -netplan +ifupdown -nm -scfg -sys selects eni - ('eni', False, True, False, False, False, False), + ("eni", False, True, False, False, False, False), # +netplan +ifupdown -nm -scfg -sys selects eni - ('eni', True, True, False, False, False, False), + ("eni", True, True, False, False, False, False), # +netplan -ifupdown -nm -scfg -sys selects netplan - ('netplan', True, False, False, False, False, False), + ("netplan", True, False, False, False, False, False), # Ubuntu with Network-Manager installed # +netplan -ifupdown +nm -scfg -sys selects netplan - ('netplan', True, False, True, False, False, False), + ("netplan", True, False, True, False, False, False), # Centos/OpenSuse with Network-Manager installed selects sysconfig # -netplan -ifupdown +nm -scfg +sys selects netplan - ('sysconfig', False, False, True, False, True, False), + ("sysconfig", False, False, True, False, True, False), # -netplan -ifupdown -nm -scfg -sys +networkd selects networkd - ('networkd', False, False, False, False, False, True), + ("networkd", False, False, False, False, False, True), ), ) @mock.patch("cloudinit.net.renderers.networkd.available") @@ -5308,15 +6129,26 @@ class TestRenderersSelect: @mock.patch("cloudinit.net.renderers.sysconfig.available_nm") @mock.patch("cloudinit.net.renderers.eni.available") def test_valid_renderer_from_defaults_depending_on_availability( - self, m_eni_avail, m_nm_avail, m_scfg_avail, m_sys_avail, - m_netplan_avail, m_networkd_avail, renderer_selected, - netplan, eni, nm, scfg, sys, networkd + self, + m_eni_avail, + m_nm_avail, + m_scfg_avail, + m_sys_avail, + m_netplan_avail, + m_networkd_avail, + renderer_selected, + netplan, + eni, + nm, + scfg, + sys, + networkd, ): """Assert proper renderer per DEFAULT_PRIORITY given availability.""" - m_eni_avail.return_value = eni # ifupdown pkg presence - m_nm_avail.return_value = nm # network-manager presence - m_scfg_avail.return_value = scfg # sysconfig presence - m_sys_avail.return_value = sys # sysconfig/ifup/down presence + m_eni_avail.return_value = eni # ifupdown pkg presence + m_nm_avail.return_value = nm # network-manager presence + m_scfg_avail.return_value = scfg # sysconfig presence + m_sys_avail.return_value = sys # sysconfig/ifup/down presence m_netplan_avail.return_value = netplan # netplan presence m_networkd_avail.return_value = networkd # networkd presence if isinstance(renderer_selected, str): @@ -5335,14 +6167,14 @@ class TestNetRenderers(CiTestCase): def test_eni_and_sysconfig_available(self, m_eni_avail, m_sysc_avail): m_eni_avail.return_value = True m_sysc_avail.return_value = True - found = renderers.search(priority=['sysconfig', 'eni'], first=False) + found = renderers.search(priority=["sysconfig", "eni"], first=False) names = [f[0] for f in found] - self.assertEqual(['sysconfig', 'eni'], names) + self.assertEqual(["sysconfig", "eni"], names) @mock.patch("cloudinit.net.renderers.eni.available") def test_search_returns_empty_on_none(self, m_eni_avail): m_eni_avail.return_value = False - found = renderers.search(priority=['eni'], first=False) + found = renderers.search(priority=["eni"], first=False) self.assertEqual([], found) @mock.patch("cloudinit.net.renderers.sysconfig.available") @@ -5351,16 +6183,16 @@ class TestNetRenderers(CiTestCase): # available should only be called until one is found. m_eni_avail.return_value = True m_sysc_avail.side_effect = Exception("Should not call me") - found = renderers.search(priority=['eni', 'sysconfig'], first=True)[0] - self.assertEqual(['eni'], [found[0]]) + found = renderers.search(priority=["eni", "sysconfig"], first=True)[0] + self.assertEqual(["eni"], [found[0]]) @mock.patch("cloudinit.net.renderers.sysconfig.available") @mock.patch("cloudinit.net.renderers.eni.available") def test_select_positive(self, m_eni_avail, m_sysc_avail): m_eni_avail.return_value = True m_sysc_avail.return_value = False - found = renderers.select(priority=['sysconfig', 'eni']) - self.assertEqual('eni', found[0]) + found = renderers.select(priority=["sysconfig", "eni"]) + self.assertEqual("eni", found[0]) @mock.patch("cloudinit.net.renderers.sysconfig.available") @mock.patch("cloudinit.net.renderers.eni.available") @@ -5369,24 +6201,25 @@ class TestNetRenderers(CiTestCase): m_eni_avail.return_value = False m_sysc_avail.return_value = False - self.assertRaises(net.RendererNotFoundError, renderers.select, - priority=['sysconfig', 'eni']) + self.assertRaises( + net.RendererNotFoundError, + renderers.select, + priority=["sysconfig", "eni"], + ) @mock.patch("cloudinit.net.sysconfig.available_sysconfig") @mock.patch("cloudinit.util.system_info") def test_sysconfig_available_uses_variant_mapping(self, m_info, m_avail): m_avail.return_value = True variants = [ - 'suse', - 'centos', - 'eurolinux', - 'fedora', - 'rhel', + "suse", + "centos", + "eurolinux", + "fedora", + "rhel", ] for distro_name in variants: - m_info.return_value = { - "variant": distro_name - } + m_info.return_value = {"variant": distro_name} if hasattr(util.system_info, "cache_clear"): util.system_info.cache_clear() result = sysconfig.available() @@ -5395,73 +6228,93 @@ class TestNetRenderers(CiTestCase): @mock.patch("cloudinit.net.renderers.networkd.available") def test_networkd_available(self, m_nwkd_avail): m_nwkd_avail.return_value = True - found = renderers.search(priority=['networkd'], first=False) - self.assertEqual('networkd', found[0][0]) + found = renderers.search(priority=["networkd"], first=False) + self.assertEqual("networkd", found[0][0]) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", - mock.Mock(return_value=False) + mock.Mock(return_value=False), ) class TestGetInterfaces(CiTestCase): - _data = {'bonds': ['bond1'], - 'bridges': ['bridge1'], - 'vlans': ['bond1.101'], - 'own_macs': ['enp0s1', 'enp0s2', 'bridge1-nic', 'bridge1', - 'bond1.101', 'lo', 'eth1'], - 'macs': {'enp0s1': 'aa:aa:aa:aa:aa:01', - 'enp0s2': 'aa:aa:aa:aa:aa:02', - 'bond1': 'aa:aa:aa:aa:aa:01', - 'bond1.101': 'aa:aa:aa:aa:aa:01', - 'bridge1': 'aa:aa:aa:aa:aa:03', - 'bridge1-nic': 'aa:aa:aa:aa:aa:03', - 'lo': '00:00:00:00:00:00', - 'greptap0': '00:00:00:00:00:00', - 'eth1': 'aa:aa:aa:aa:aa:01', - 'tun0': None}, - 'drivers': {'enp0s1': 'virtio_net', - 'enp0s2': 'e1000', - 'bond1': None, - 'bond1.101': None, - 'bridge1': None, - 'bridge1-nic': None, - 'lo': None, - 'greptap0': None, - 'eth1': 'mlx4_core', - 'tun0': None}} + _data = { + "bonds": ["bond1"], + "bridges": ["bridge1"], + "vlans": ["bond1.101"], + "own_macs": [ + "enp0s1", + "enp0s2", + "bridge1-nic", + "bridge1", + "bond1.101", + "lo", + "eth1", + ], + "macs": { + "enp0s1": "aa:aa:aa:aa:aa:01", + "enp0s2": "aa:aa:aa:aa:aa:02", + "bond1": "aa:aa:aa:aa:aa:01", + "bond1.101": "aa:aa:aa:aa:aa:01", + "bridge1": "aa:aa:aa:aa:aa:03", + "bridge1-nic": "aa:aa:aa:aa:aa:03", + "lo": "00:00:00:00:00:00", + "greptap0": "00:00:00:00:00:00", + "eth1": "aa:aa:aa:aa:aa:01", + "tun0": None, + }, + "drivers": { + "enp0s1": "virtio_net", + "enp0s2": "e1000", + "bond1": None, + "bond1.101": None, + "bridge1": None, + "bridge1-nic": None, + "lo": None, + "greptap0": None, + "eth1": "mlx4_core", + "tun0": None, + }, + } data = {} def _se_get_devicelist(self): - return list(self.data['devices']) + return list(self.data["devices"]) def _se_device_driver(self, name): - return self.data['drivers'][name] + return self.data["drivers"][name] def _se_device_devid(self, name): - return '0x%s' % sorted(list(self.data['drivers'].keys())).index(name) + return "0x%s" % sorted(list(self.data["drivers"].keys())).index(name) def _se_get_interface_mac(self, name): - return self.data['macs'][name] + return self.data["macs"][name] def _se_is_bridge(self, name): - return name in self.data['bridges'] + return name in self.data["bridges"] def _se_is_vlan(self, name): - return name in self.data['vlans'] + return name in self.data["vlans"] def _se_interface_has_own_mac(self, name): - return name in self.data['own_macs'] + return name in self.data["own_macs"] def _mock_setup(self): self.data = copy.deepcopy(self._data) - self.data['devices'] = set(list(self.data['macs'].keys())) - mocks = ('get_devicelist', 'get_interface_mac', 'is_bridge', - 'interface_has_own_mac', 'is_vlan', 'device_driver', - 'device_devid') + self.data["devices"] = set(list(self.data["macs"].keys())) + mocks = ( + "get_devicelist", + "get_interface_mac", + "is_bridge", + "interface_has_own_mac", + "is_vlan", + "device_driver", + "device_devid", + ) self.mocks = {} for n in mocks: - m = mock.patch('cloudinit.net.' + n, - side_effect=getattr(self, '_se_' + n)) + m = mock.patch( + "cloudinit.net." + n, side_effect=getattr(self, "_se_" + n) + ) self.addCleanup(m.stop) self.mocks[n] = m.start() @@ -5469,30 +6322,31 @@ class TestGetInterfaces(CiTestCase): self._mock_setup() ret = net.get_interfaces() - self.assertIn('enp0s1', self._se_get_devicelist()) - self.assertIn('eth1', self._se_get_devicelist()) - found = [ent for ent in ret if 'aa:aa:aa:aa:aa:01' in ent] + self.assertIn("enp0s1", self._se_get_devicelist()) + self.assertIn("eth1", self._se_get_devicelist()) + found = [ent for ent in ret if "aa:aa:aa:aa:aa:01" in ent] self.assertEqual(len(found), 2) def test_gi_excludes_any_without_mac_address(self): self._mock_setup() ret = net.get_interfaces() - self.assertIn('tun0', self._se_get_devicelist()) - found = [ent for ent in ret if 'tun0' in ent] + self.assertIn("tun0", self._se_get_devicelist()) + found = [ent for ent in ret if "tun0" in ent] self.assertEqual(len(found), 0) def test_gi_excludes_stolen_macs(self): self._mock_setup() ret = net.get_interfaces() - self.mocks['interface_has_own_mac'].assert_has_calls( - [mock.call('enp0s1'), mock.call('bond1')], any_order=True) + self.mocks["interface_has_own_mac"].assert_has_calls( + [mock.call("enp0s1"), mock.call("bond1")], any_order=True + ) expected = [ - ('enp0s2', 'aa:aa:aa:aa:aa:02', 'e1000', '0x5'), - ('enp0s1', 'aa:aa:aa:aa:aa:01', 'virtio_net', '0x4'), - ('eth1', 'aa:aa:aa:aa:aa:01', 'mlx4_core', '0x6'), - ('lo', '00:00:00:00:00:00', None, '0x8'), - ('bridge1-nic', 'aa:aa:aa:aa:aa:03', None, '0x3'), + ("enp0s2", "aa:aa:aa:aa:aa:02", "e1000", "0x5"), + ("enp0s1", "aa:aa:aa:aa:aa:01", "virtio_net", "0x4"), + ("eth1", "aa:aa:aa:aa:aa:01", "mlx4_core", "0x6"), + ("lo", "00:00:00:00:00:00", None, "0x8"), + ("bridge1-nic", "aa:aa:aa:aa:aa:03", None, "0x3"), ] self.assertEqual(sorted(expected), sorted(ret)) @@ -5501,24 +6355,29 @@ class TestGetInterfaces(CiTestCase): # add a device 'b1', make all return they have their "own mac", # set everything other than 'b1' to be a bridge. # then expect b1 is the only thing left. - self.data['macs']['b1'] = 'aa:aa:aa:aa:aa:b1' - self.data['drivers']['b1'] = None - self.data['devices'].add('b1') - self.data['bonds'] = [] - self.data['own_macs'] = self.data['devices'] - self.data['bridges'] = [f for f in self.data['devices'] if f != "b1"] + self.data["macs"]["b1"] = "aa:aa:aa:aa:aa:b1" + self.data["drivers"]["b1"] = None + self.data["devices"].add("b1") + self.data["bonds"] = [] + self.data["own_macs"] = self.data["devices"] + self.data["bridges"] = [f for f in self.data["devices"] if f != "b1"] ret = net.get_interfaces() - self.assertEqual([('b1', 'aa:aa:aa:aa:aa:b1', None, '0x0')], ret) - self.mocks['is_bridge'].assert_has_calls( - [mock.call('bridge1'), mock.call('enp0s1'), mock.call('bond1'), - mock.call('b1')], - any_order=True) + self.assertEqual([("b1", "aa:aa:aa:aa:aa:b1", None, "0x0")], ret) + self.mocks["is_bridge"].assert_has_calls( + [ + mock.call("bridge1"), + mock.call("enp0s1"), + mock.call("bond1"), + mock.call("b1"), + ], + any_order=True, + ) class TestInterfaceHasOwnMac(CiTestCase): """Test interface_has_own_mac. This is admittedly a bit whitebox.""" - @mock.patch('cloudinit.net.read_sys_net_int', return_value=None) + @mock.patch("cloudinit.net.read_sys_net_int", return_value=None) def test_non_strict_with_no_addr_assign_type(self, m_read_sys_net_int): """If nic does not have addr_assign_type, it is not "stolen". @@ -5535,18 +6394,19 @@ class TestInterfaceHasOwnMac(CiTestCase): """ self.assertTrue(interface_has_own_mac("eth0")) - @mock.patch('cloudinit.net.read_sys_net_int', return_value=None) + @mock.patch("cloudinit.net.read_sys_net_int", return_value=None) def test_strict_with_no_addr_assign_type_raises(self, m_read_sys_net_int): with self.assertRaises(ValueError): interface_has_own_mac("eth0", True) - @mock.patch('cloudinit.net.read_sys_net_int') + @mock.patch("cloudinit.net.read_sys_net_int") def test_expected_values(self, m_read_sys_net_int): msg = "address_assign_type=%d said to not have own mac" for address_assign_type in (0, 1, 3): m_read_sys_net_int.return_value = address_assign_type self.assertTrue( - interface_has_own_mac("eth0", msg % address_assign_type)) + interface_has_own_mac("eth0", msg % address_assign_type) + ) m_read_sys_net_int.return_value = 2 self.assertFalse(interface_has_own_mac("eth0")) @@ -5554,218 +6414,281 @@ class TestInterfaceHasOwnMac(CiTestCase): @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", - mock.Mock(return_value=False) + mock.Mock(return_value=False), ) class TestGetInterfacesByMac(CiTestCase): - _data = {'bonds': ['bond1'], - 'bridges': ['bridge1'], - 'vlans': ['bond1.101'], - 'own_macs': ['enp0s1', 'enp0s2', 'bridge1-nic', 'bridge1', - 'bond1.101', 'lo'], - 'macs': {'enp0s1': 'aa:aa:aa:aa:aa:01', - 'enp0s2': 'aa:aa:aa:aa:aa:02', - 'bond1': 'aa:aa:aa:aa:aa:01', - 'bond1.101': 'aa:aa:aa:aa:aa:01', - 'bridge1': 'aa:aa:aa:aa:aa:03', - 'bridge1-nic': 'aa:aa:aa:aa:aa:03', - 'lo': '00:00:00:00:00:00', - 'greptap0': '00:00:00:00:00:00', - 'tun0': None}} + _data = { + "bonds": ["bond1"], + "bridges": ["bridge1"], + "vlans": ["bond1.101"], + "own_macs": [ + "enp0s1", + "enp0s2", + "bridge1-nic", + "bridge1", + "bond1.101", + "lo", + ], + "macs": { + "enp0s1": "aa:aa:aa:aa:aa:01", + "enp0s2": "aa:aa:aa:aa:aa:02", + "bond1": "aa:aa:aa:aa:aa:01", + "bond1.101": "aa:aa:aa:aa:aa:01", + "bridge1": "aa:aa:aa:aa:aa:03", + "bridge1-nic": "aa:aa:aa:aa:aa:03", + "lo": "00:00:00:00:00:00", + "greptap0": "00:00:00:00:00:00", + "tun0": None, + }, + } data = {} def _se_get_devicelist(self): - return list(self.data['devices']) + return list(self.data["devices"]) def _se_get_interface_mac(self, name): - return self.data['macs'][name] + return self.data["macs"][name] def _se_is_bridge(self, name): - return name in self.data['bridges'] + return name in self.data["bridges"] def _se_is_vlan(self, name): - return name in self.data['vlans'] + return name in self.data["vlans"] def _se_interface_has_own_mac(self, name): - return name in self.data['own_macs'] + return name in self.data["own_macs"] def _se_get_ib_interface_hwaddr(self, name, ethernet_format): - ib_hwaddr = self.data.get('ib_hwaddr', {}) + ib_hwaddr = self.data.get("ib_hwaddr", {}) return ib_hwaddr.get(name, {}).get(ethernet_format) def _mock_setup(self): self.data = copy.deepcopy(self._data) - self.data['devices'] = set(list(self.data['macs'].keys())) - mocks = ('get_devicelist', 'get_interface_mac', 'is_bridge', - 'interface_has_own_mac', 'is_vlan', 'get_ib_interface_hwaddr') + self.data["devices"] = set(list(self.data["macs"].keys())) + mocks = ( + "get_devicelist", + "get_interface_mac", + "is_bridge", + "interface_has_own_mac", + "is_vlan", + "get_ib_interface_hwaddr", + ) self.mocks = {} for n in mocks: - m = mock.patch('cloudinit.net.' + n, - side_effect=getattr(self, '_se_' + n)) + m = mock.patch( + "cloudinit.net." + n, side_effect=getattr(self, "_se_" + n) + ) self.addCleanup(m.stop) self.mocks[n] = m.start() def test_raise_exception_on_duplicate_macs(self): self._mock_setup() - self.data['macs']['bridge1-nic'] = self.data['macs']['enp0s1'] + self.data["macs"]["bridge1-nic"] = self.data["macs"]["enp0s1"] self.assertRaises(RuntimeError, net.get_interfaces_by_mac) def test_excludes_any_without_mac_address(self): self._mock_setup() ret = net.get_interfaces_by_mac() - self.assertIn('tun0', self._se_get_devicelist()) - self.assertNotIn('tun0', ret.values()) + self.assertIn("tun0", self._se_get_devicelist()) + self.assertNotIn("tun0", ret.values()) def test_excludes_stolen_macs(self): self._mock_setup() ret = net.get_interfaces_by_mac() - self.mocks['interface_has_own_mac'].assert_has_calls( - [mock.call('enp0s1'), mock.call('bond1')], any_order=True) + self.mocks["interface_has_own_mac"].assert_has_calls( + [mock.call("enp0s1"), mock.call("bond1")], any_order=True + ) self.assertEqual( - {'aa:aa:aa:aa:aa:01': 'enp0s1', 'aa:aa:aa:aa:aa:02': 'enp0s2', - 'aa:aa:aa:aa:aa:03': 'bridge1-nic', '00:00:00:00:00:00': 'lo'}, - ret) + { + "aa:aa:aa:aa:aa:01": "enp0s1", + "aa:aa:aa:aa:aa:02": "enp0s2", + "aa:aa:aa:aa:aa:03": "bridge1-nic", + "00:00:00:00:00:00": "lo", + }, + ret, + ) def test_excludes_bridges(self): self._mock_setup() # add a device 'b1', make all return they have their "own mac", # set everything other than 'b1' to be a bridge. # then expect b1 is the only thing left. - self.data['macs']['b1'] = 'aa:aa:aa:aa:aa:b1' - self.data['devices'].add('b1') - self.data['bonds'] = [] - self.data['own_macs'] = self.data['devices'] - self.data['bridges'] = [f for f in self.data['devices'] if f != "b1"] + self.data["macs"]["b1"] = "aa:aa:aa:aa:aa:b1" + self.data["devices"].add("b1") + self.data["bonds"] = [] + self.data["own_macs"] = self.data["devices"] + self.data["bridges"] = [f for f in self.data["devices"] if f != "b1"] ret = net.get_interfaces_by_mac() - self.assertEqual({'aa:aa:aa:aa:aa:b1': 'b1'}, ret) - self.mocks['is_bridge'].assert_has_calls( - [mock.call('bridge1'), mock.call('enp0s1'), mock.call('bond1'), - mock.call('b1')], - any_order=True) + self.assertEqual({"aa:aa:aa:aa:aa:b1": "b1"}, ret) + self.mocks["is_bridge"].assert_has_calls( + [ + mock.call("bridge1"), + mock.call("enp0s1"), + mock.call("bond1"), + mock.call("b1"), + ], + any_order=True, + ) def test_excludes_vlans(self): self._mock_setup() # add a device 'b1', make all return they have their "own mac", # set everything other than 'b1' to be a vlan. # then expect b1 is the only thing left. - self.data['macs']['b1'] = 'aa:aa:aa:aa:aa:b1' - self.data['devices'].add('b1') - self.data['bonds'] = [] - self.data['bridges'] = [] - self.data['own_macs'] = self.data['devices'] - self.data['vlans'] = [f for f in self.data['devices'] if f != "b1"] + self.data["macs"]["b1"] = "aa:aa:aa:aa:aa:b1" + self.data["devices"].add("b1") + self.data["bonds"] = [] + self.data["bridges"] = [] + self.data["own_macs"] = self.data["devices"] + self.data["vlans"] = [f for f in self.data["devices"] if f != "b1"] ret = net.get_interfaces_by_mac() - self.assertEqual({'aa:aa:aa:aa:aa:b1': 'b1'}, ret) - self.mocks['is_vlan'].assert_has_calls( - [mock.call('bridge1'), mock.call('enp0s1'), mock.call('bond1'), - mock.call('b1')], - any_order=True) + self.assertEqual({"aa:aa:aa:aa:aa:b1": "b1"}, ret) + self.mocks["is_vlan"].assert_has_calls( + [ + mock.call("bridge1"), + mock.call("enp0s1"), + mock.call("bond1"), + mock.call("b1"), + ], + any_order=True, + ) def test_duplicates_of_empty_mac_are_ok(self): """Duplicate macs of 00:00:00:00:00:00 should be skipped.""" self._mock_setup() empty_mac = "00:00:00:00:00:00" - addnics = ('greptap1', 'lo', 'greptap2') - self.data['macs'].update(dict((k, empty_mac) for k in addnics)) - self.data['devices'].update(set(addnics)) - self.data['own_macs'].extend(list(addnics)) + addnics = ("greptap1", "lo", "greptap2") + self.data["macs"].update(dict((k, empty_mac) for k in addnics)) + self.data["devices"].update(set(addnics)) + self.data["own_macs"].extend(list(addnics)) ret = net.get_interfaces_by_mac() - self.assertEqual('lo', ret[empty_mac]) + self.assertEqual("lo", ret[empty_mac]) def test_skip_all_zeros(self): """Any mac of 00:... should be skipped.""" self._mock_setup() emac1, emac2, emac4, emac6 = ( - '00', '00:00', '00:00:00:00', '00:00:00:00:00:00') - addnics = {'empty1': emac1, 'emac2a': emac2, 'emac2b': emac2, - 'emac4': emac4, 'emac6': emac6} - self.data['macs'].update(addnics) - self.data['devices'].update(set(addnics)) - self.data['own_macs'].extend(addnics.keys()) + "00", + "00:00", + "00:00:00:00", + "00:00:00:00:00:00", + ) + addnics = { + "empty1": emac1, + "emac2a": emac2, + "emac2b": emac2, + "emac4": emac4, + "emac6": emac6, + } + self.data["macs"].update(addnics) + self.data["devices"].update(set(addnics)) + self.data["own_macs"].extend(addnics.keys()) ret = net.get_interfaces_by_mac() - self.assertEqual('lo', ret['00:00:00:00:00:00']) + self.assertEqual("lo", ret["00:00:00:00:00:00"]) def test_ib(self): - ib_addr = '80:00:00:28:fe:80:00:00:00:00:00:00:00:11:22:03:00:33:44:56' - ib_addr_eth_format = '00:11:22:33:44:56' + ib_addr = "80:00:00:28:fe:80:00:00:00:00:00:00:00:11:22:03:00:33:44:56" + ib_addr_eth_format = "00:11:22:33:44:56" self._mock_setup() - self.data['devices'] = ['enp0s1', 'ib0'] - self.data['own_macs'].append('ib0') - self.data['macs']['ib0'] = ib_addr - self.data['ib_hwaddr'] = {'ib0': {True: ib_addr_eth_format, - False: ib_addr}} + self.data["devices"] = ["enp0s1", "ib0"] + self.data["own_macs"].append("ib0") + self.data["macs"]["ib0"] = ib_addr + self.data["ib_hwaddr"] = { + "ib0": {True: ib_addr_eth_format, False: ib_addr} + } result = net.get_interfaces_by_mac() - expected = {'aa:aa:aa:aa:aa:01': 'enp0s1', - ib_addr_eth_format: 'ib0', ib_addr: 'ib0'} + expected = { + "aa:aa:aa:aa:aa:01": "enp0s1", + ib_addr_eth_format: "ib0", + ib_addr: "ib0", + } self.assertEqual(expected, result) class TestInterfacesSorting(CiTestCase): - def test_natural_order(self): - data = ['ens5', 'ens6', 'ens3', 'ens20', 'ens13', 'ens2'] + data = ["ens5", "ens6", "ens3", "ens20", "ens13", "ens2"] self.assertEqual( sorted(data, key=natural_sort_key), - ['ens2', 'ens3', 'ens5', 'ens6', 'ens13', 'ens20']) - data2 = ['enp2s0', 'enp2s3', 'enp0s3', 'enp0s13', 'enp0s8', 'enp1s2'] + ["ens2", "ens3", "ens5", "ens6", "ens13", "ens20"], + ) + data2 = ["enp2s0", "enp2s3", "enp0s3", "enp0s13", "enp0s8", "enp1s2"] self.assertEqual( sorted(data2, key=natural_sort_key), - ['enp0s3', 'enp0s8', 'enp0s13', 'enp1s2', 'enp2s0', 'enp2s3']) + ["enp0s3", "enp0s8", "enp0s13", "enp1s2", "enp2s0", "enp2s3"], + ) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", - mock.Mock(return_value=False) + mock.Mock(return_value=False), ) class TestGetIBHwaddrsByInterface(CiTestCase): - _ib_addr = '80:00:00:28:fe:80:00:00:00:00:00:00:00:11:22:03:00:33:44:56' - _ib_addr_eth_format = '00:11:22:33:44:56' - _data = {'devices': ['enp0s1', 'enp0s2', 'bond1', 'bridge1', - 'bridge1-nic', 'tun0', 'ib0'], - 'bonds': ['bond1'], - 'bridges': ['bridge1'], - 'own_macs': ['enp0s1', 'enp0s2', 'bridge1-nic', 'bridge1', 'ib0'], - 'macs': {'enp0s1': 'aa:aa:aa:aa:aa:01', - 'enp0s2': 'aa:aa:aa:aa:aa:02', - 'bond1': 'aa:aa:aa:aa:aa:01', - 'bridge1': 'aa:aa:aa:aa:aa:03', - 'bridge1-nic': 'aa:aa:aa:aa:aa:03', - 'tun0': None, - 'ib0': _ib_addr}, - 'ib_hwaddr': {'ib0': {True: _ib_addr_eth_format, - False: _ib_addr}}} + _ib_addr = "80:00:00:28:fe:80:00:00:00:00:00:00:00:11:22:03:00:33:44:56" + _ib_addr_eth_format = "00:11:22:33:44:56" + _data = { + "devices": [ + "enp0s1", + "enp0s2", + "bond1", + "bridge1", + "bridge1-nic", + "tun0", + "ib0", + ], + "bonds": ["bond1"], + "bridges": ["bridge1"], + "own_macs": ["enp0s1", "enp0s2", "bridge1-nic", "bridge1", "ib0"], + "macs": { + "enp0s1": "aa:aa:aa:aa:aa:01", + "enp0s2": "aa:aa:aa:aa:aa:02", + "bond1": "aa:aa:aa:aa:aa:01", + "bridge1": "aa:aa:aa:aa:aa:03", + "bridge1-nic": "aa:aa:aa:aa:aa:03", + "tun0": None, + "ib0": _ib_addr, + }, + "ib_hwaddr": {"ib0": {True: _ib_addr_eth_format, False: _ib_addr}}, + } data = {} def _mock_setup(self): self.data = copy.deepcopy(self._data) - mocks = ('get_devicelist', 'get_interface_mac', 'is_bridge', - 'interface_has_own_mac', 'get_ib_interface_hwaddr') + mocks = ( + "get_devicelist", + "get_interface_mac", + "is_bridge", + "interface_has_own_mac", + "get_ib_interface_hwaddr", + ) self.mocks = {} for n in mocks: - m = mock.patch('cloudinit.net.' + n, - side_effect=getattr(self, '_se_' + n)) + m = mock.patch( + "cloudinit.net." + n, side_effect=getattr(self, "_se_" + n) + ) self.addCleanup(m.stop) self.mocks[n] = m.start() def _se_get_devicelist(self): - return self.data['devices'] + return self.data["devices"] def _se_get_interface_mac(self, name): - return self.data['macs'][name] + return self.data["macs"][name] def _se_is_bridge(self, name): - return name in self.data['bridges'] + return name in self.data["bridges"] def _se_interface_has_own_mac(self, name): - return name in self.data['own_macs'] + return name in self.data["own_macs"] def _se_get_ib_interface_hwaddr(self, name, ethernet_format): - ib_hwaddr = self.data.get('ib_hwaddr', {}) + ib_hwaddr = self.data.get("ib_hwaddr", {}) return ib_hwaddr.get(name, {}).get(ethernet_format) def test_ethernet(self): self._mock_setup() - self.data['devices'].remove('ib0') + self.data["devices"].remove("ib0") result = net.get_ib_hwaddrs_by_interface() expected = {} self.assertEqual(expected, result) @@ -5773,7 +6696,7 @@ class TestGetIBHwaddrsByInterface(CiTestCase): def test_ib(self): self._mock_setup() result = net.get_ib_hwaddrs_by_interface() - expected = {'ib0': self._ib_addr} + expected = {"ib0": self._ib_addr} self.assertEqual(expected, result) @@ -5786,239 +6709,305 @@ def _gzip_data(data): class TestRenameInterfaces(CiTestCase): - - @mock.patch('cloudinit.subp.subp') + @mock.patch("cloudinit.subp.subp") def test_rename_all(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'interface0', 'virtio_net', '0x3'), - ('00:11:22:33:44:aa', 'interface2', 'virtio_net', '0x5'), + ("00:11:22:33:44:55", "interface0", "virtio_net", "0x3"), + ("00:11:22:33:44:aa", "interface2", "virtio_net", "0x5"), ] current_info = { - 'ens3': { - 'downable': True, - 'device_id': '0x3', - 'driver': 'virtio_net', - 'mac': '00:11:22:33:44:55', - 'name': 'ens3', - 'up': False}, - 'ens5': { - 'downable': True, - 'device_id': '0x5', - 'driver': 'virtio_net', - 'mac': '00:11:22:33:44:aa', - 'name': 'ens5', - 'up': False}, + "ens3": { + "downable": True, + "device_id": "0x3", + "driver": "virtio_net", + "mac": "00:11:22:33:44:55", + "name": "ens3", + "up": False, + }, + "ens5": { + "downable": True, + "device_id": "0x5", + "driver": "virtio_net", + "mac": "00:11:22:33:44:aa", + "name": "ens5", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'ens3', 'name', 'interface0'], - capture=True), - mock.call(['ip', 'link', 'set', 'ens5', 'name', 'interface2'], - capture=True), - ]) - - @mock.patch('cloudinit.subp.subp') + mock_subp.assert_has_calls( + [ + mock.call( + ["ip", "link", "set", "ens3", "name", "interface0"], + capture=True, + ), + mock.call( + ["ip", "link", "set", "ens5", "name", "interface2"], + capture=True, + ), + ] + ) + + @mock.patch("cloudinit.subp.subp") def test_rename_no_driver_no_device_id(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'interface0', None, None), - ('00:11:22:33:44:aa', 'interface1', None, None), + ("00:11:22:33:44:55", "interface0", None, None), + ("00:11:22:33:44:aa", "interface1", None, None), ] current_info = { - 'eth0': { - 'downable': True, - 'device_id': None, - 'driver': None, - 'mac': '00:11:22:33:44:55', - 'name': 'eth0', - 'up': False}, - 'eth1': { - 'downable': True, - 'device_id': None, - 'driver': None, - 'mac': '00:11:22:33:44:aa', - 'name': 'eth1', - 'up': False}, + "eth0": { + "downable": True, + "device_id": None, + "driver": None, + "mac": "00:11:22:33:44:55", + "name": "eth0", + "up": False, + }, + "eth1": { + "downable": True, + "device_id": None, + "driver": None, + "mac": "00:11:22:33:44:aa", + "name": "eth1", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'eth0', 'name', 'interface0'], - capture=True), - mock.call(['ip', 'link', 'set', 'eth1', 'name', 'interface1'], - capture=True), - ]) - - @mock.patch('cloudinit.subp.subp') + mock_subp.assert_has_calls( + [ + mock.call( + ["ip", "link", "set", "eth0", "name", "interface0"], + capture=True, + ), + mock.call( + ["ip", "link", "set", "eth1", "name", "interface1"], + capture=True, + ), + ] + ) + + @mock.patch("cloudinit.subp.subp") def test_rename_all_bounce(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'interface0', 'virtio_net', '0x3'), - ('00:11:22:33:44:aa', 'interface2', 'virtio_net', '0x5'), + ("00:11:22:33:44:55", "interface0", "virtio_net", "0x3"), + ("00:11:22:33:44:aa", "interface2", "virtio_net", "0x5"), ] current_info = { - 'ens3': { - 'downable': True, - 'device_id': '0x3', - 'driver': 'virtio_net', - 'mac': '00:11:22:33:44:55', - 'name': 'ens3', - 'up': True}, - 'ens5': { - 'downable': True, - 'device_id': '0x5', - 'driver': 'virtio_net', - 'mac': '00:11:22:33:44:aa', - 'name': 'ens5', - 'up': True}, + "ens3": { + "downable": True, + "device_id": "0x3", + "driver": "virtio_net", + "mac": "00:11:22:33:44:55", + "name": "ens3", + "up": True, + }, + "ens5": { + "downable": True, + "device_id": "0x5", + "driver": "virtio_net", + "mac": "00:11:22:33:44:aa", + "name": "ens5", + "up": True, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'ens3', 'down'], capture=True), - mock.call(['ip', 'link', 'set', 'ens3', 'name', 'interface0'], - capture=True), - mock.call(['ip', 'link', 'set', 'ens5', 'down'], capture=True), - mock.call(['ip', 'link', 'set', 'ens5', 'name', 'interface2'], - capture=True), - mock.call(['ip', 'link', 'set', 'interface0', 'up'], capture=True), - mock.call(['ip', 'link', 'set', 'interface2', 'up'], capture=True) - ]) - - @mock.patch('cloudinit.subp.subp') + mock_subp.assert_has_calls( + [ + mock.call(["ip", "link", "set", "ens3", "down"], capture=True), + mock.call( + ["ip", "link", "set", "ens3", "name", "interface0"], + capture=True, + ), + mock.call(["ip", "link", "set", "ens5", "down"], capture=True), + mock.call( + ["ip", "link", "set", "ens5", "name", "interface2"], + capture=True, + ), + mock.call( + ["ip", "link", "set", "interface0", "up"], capture=True + ), + mock.call( + ["ip", "link", "set", "interface2", "up"], capture=True + ), + ] + ) + + @mock.patch("cloudinit.subp.subp") def test_rename_duplicate_macs(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', '0x3'), - ('00:11:22:33:44:55', 'vf1', 'mlx4_core', '0x5'), + ("00:11:22:33:44:55", "eth0", "hv_netsvc", "0x3"), + ("00:11:22:33:44:55", "vf1", "mlx4_core", "0x5"), ] current_info = { - 'eth0': { - 'downable': True, - 'device_id': '0x3', - 'driver': 'hv_netsvc', - 'mac': '00:11:22:33:44:55', - 'name': 'eth0', - 'up': False}, - 'eth1': { - 'downable': True, - 'device_id': '0x5', - 'driver': 'mlx4_core', - 'mac': '00:11:22:33:44:55', - 'name': 'eth1', - 'up': False}, + "eth0": { + "downable": True, + "device_id": "0x3", + "driver": "hv_netsvc", + "mac": "00:11:22:33:44:55", + "name": "eth0", + "up": False, + }, + "eth1": { + "downable": True, + "device_id": "0x5", + "driver": "mlx4_core", + "mac": "00:11:22:33:44:55", + "name": "eth1", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'], - capture=True), - ]) + mock_subp.assert_has_calls( + [ + mock.call( + ["ip", "link", "set", "eth1", "name", "vf1"], capture=True + ), + ] + ) - @mock.patch('cloudinit.subp.subp') + @mock.patch("cloudinit.subp.subp") def test_rename_duplicate_macs_driver_no_devid(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', None), - ('00:11:22:33:44:55', 'vf1', 'mlx4_core', None), + ("00:11:22:33:44:55", "eth0", "hv_netsvc", None), + ("00:11:22:33:44:55", "vf1", "mlx4_core", None), ] current_info = { - 'eth0': { - 'downable': True, - 'device_id': '0x3', - 'driver': 'hv_netsvc', - 'mac': '00:11:22:33:44:55', - 'name': 'eth0', - 'up': False}, - 'eth1': { - 'downable': True, - 'device_id': '0x5', - 'driver': 'mlx4_core', - 'mac': '00:11:22:33:44:55', - 'name': 'eth1', - 'up': False}, + "eth0": { + "downable": True, + "device_id": "0x3", + "driver": "hv_netsvc", + "mac": "00:11:22:33:44:55", + "name": "eth0", + "up": False, + }, + "eth1": { + "downable": True, + "device_id": "0x5", + "driver": "mlx4_core", + "mac": "00:11:22:33:44:55", + "name": "eth1", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'], - capture=True), - ]) + mock_subp.assert_has_calls( + [ + mock.call( + ["ip", "link", "set", "eth1", "name", "vf1"], capture=True + ), + ] + ) - @mock.patch('cloudinit.subp.subp') + @mock.patch("cloudinit.subp.subp") def test_rename_multi_mac_dups(self, mock_subp): renames = [ - ('00:11:22:33:44:55', 'eth0', 'hv_netsvc', '0x3'), - ('00:11:22:33:44:55', 'vf1', 'mlx4_core', '0x5'), - ('00:11:22:33:44:55', 'vf2', 'mlx4_core', '0x7'), + ("00:11:22:33:44:55", "eth0", "hv_netsvc", "0x3"), + ("00:11:22:33:44:55", "vf1", "mlx4_core", "0x5"), + ("00:11:22:33:44:55", "vf2", "mlx4_core", "0x7"), ] current_info = { - 'eth0': { - 'downable': True, - 'device_id': '0x3', - 'driver': 'hv_netsvc', - 'mac': '00:11:22:33:44:55', - 'name': 'eth0', - 'up': False}, - 'eth1': { - 'downable': True, - 'device_id': '0x5', - 'driver': 'mlx4_core', - 'mac': '00:11:22:33:44:55', - 'name': 'eth1', - 'up': False}, - 'eth2': { - 'downable': True, - 'device_id': '0x7', - 'driver': 'mlx4_core', - 'mac': '00:11:22:33:44:55', - 'name': 'eth2', - 'up': False}, + "eth0": { + "downable": True, + "device_id": "0x3", + "driver": "hv_netsvc", + "mac": "00:11:22:33:44:55", + "name": "eth0", + "up": False, + }, + "eth1": { + "downable": True, + "device_id": "0x5", + "driver": "mlx4_core", + "mac": "00:11:22:33:44:55", + "name": "eth1", + "up": False, + }, + "eth2": { + "downable": True, + "device_id": "0x7", + "driver": "mlx4_core", + "mac": "00:11:22:33:44:55", + "name": "eth2", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) print(mock_subp.call_args_list) - mock_subp.assert_has_calls([ - mock.call(['ip', 'link', 'set', 'eth1', 'name', 'vf1'], - capture=True), - mock.call(['ip', 'link', 'set', 'eth2', 'name', 'vf2'], - capture=True), - ]) - - @mock.patch('cloudinit.subp.subp') + mock_subp.assert_has_calls( + [ + mock.call( + ["ip", "link", "set", "eth1", "name", "vf1"], capture=True + ), + mock.call( + ["ip", "link", "set", "eth2", "name", "vf2"], capture=True + ), + ] + ) + + @mock.patch("cloudinit.subp.subp") def test_rename_macs_case_insensitive(self, mock_subp): """_rename_interfaces must support upper or lower case macs.""" renames = [ - ('aa:aa:aa:aa:aa:aa', 'en0', None, None), - ('BB:BB:BB:BB:BB:BB', 'en1', None, None), - ('cc:cc:cc:cc:cc:cc', 'en2', None, None), - ('DD:DD:DD:DD:DD:DD', 'en3', None, None), + ("aa:aa:aa:aa:aa:aa", "en0", None, None), + ("BB:BB:BB:BB:BB:BB", "en1", None, None), + ("cc:cc:cc:cc:cc:cc", "en2", None, None), + ("DD:DD:DD:DD:DD:DD", "en3", None, None), ] current_info = { - 'eth0': {'downable': True, 'mac': 'AA:AA:AA:AA:AA:AA', - 'name': 'eth0', 'up': False}, - 'eth1': {'downable': True, 'mac': 'bb:bb:bb:bb:bb:bb', - 'name': 'eth1', 'up': False}, - 'eth2': {'downable': True, 'mac': 'cc:cc:cc:cc:cc:cc', - 'name': 'eth2', 'up': False}, - 'eth3': {'downable': True, 'mac': 'DD:DD:DD:DD:DD:DD', - 'name': 'eth3', 'up': False}, + "eth0": { + "downable": True, + "mac": "AA:AA:AA:AA:AA:AA", + "name": "eth0", + "up": False, + }, + "eth1": { + "downable": True, + "mac": "bb:bb:bb:bb:bb:bb", + "name": "eth1", + "up": False, + }, + "eth2": { + "downable": True, + "mac": "cc:cc:cc:cc:cc:cc", + "name": "eth2", + "up": False, + }, + "eth3": { + "downable": True, + "mac": "DD:DD:DD:DD:DD:DD", + "name": "eth3", + "up": False, + }, } net._rename_interfaces(renames, current_info=current_info) expected = [ - mock.call(['ip', 'link', 'set', 'eth%d' % i, 'name', 'en%d' % i], - capture=True) - for i in range(len(renames))] + mock.call( + ["ip", "link", "set", "eth%d" % i, "name", "en%d" % i], + capture=True, + ) + for i in range(len(renames)) + ] mock_subp.assert_has_calls(expected) class TestNetworkState(CiTestCase): - def test_bcast_addr(self): """Test mask_and_ipv4_to_bcast_addr proper execution.""" bcast_addr = network_state.mask_and_ipv4_to_bcast_addr - self.assertEqual("192.168.1.255", - bcast_addr("255.255.255.0", "192.168.1.1")) - self.assertEqual("128.42.7.255", - bcast_addr("255.255.248.0", "128.42.5.4")) - self.assertEqual("10.1.21.255", - bcast_addr("255.255.255.0", "10.1.21.4")) + self.assertEqual( + "192.168.1.255", bcast_addr("255.255.255.0", "192.168.1.1") + ) + self.assertEqual( + "128.42.7.255", bcast_addr("255.255.248.0", "128.42.5.4") + ) + self.assertEqual( + "10.1.21.255", bcast_addr("255.255.255.0", "10.1.21.4") + ) + # vi: ts=4 expandtab |