# This file is part of cloud-init. See LICENSE file for license information.
import copy
import os
import re
from io import StringIO
from textwrap import dedent
from unittest import mock
from cloudinit import distros, helpers, safeyaml, settings, subp, util
from cloudinit.distros.parsers.sys_conf import SysConf
from tests.unittests.helpers import FilesystemMockingTestCase, dir2dict
BASE_NET_CFG = """
auto lo
iface lo inet loopback
auto eth0
iface eth0 inet static
address 192.168.1.5
broadcast 192.168.1.0
gateway 192.168.1.254
netmask 255.255.255.0
network 192.168.0.0
auto eth1
iface eth1 inet dhcp
"""
BASE_NET_CFG_FROM_V2 = """
auto lo
iface lo inet loopback
auto eth0
iface eth0 inet static
address 192.168.1.5/24
gateway 192.168.1.254
auto eth1
iface eth1 inet dhcp
"""
BASE_NET_CFG_IPV6 = """
auto lo
iface lo inet loopback
auto eth0
iface eth0 inet static
address 192.168.1.5
netmask 255.255.255.0
network 192.168.0.0
broadcast 192.168.1.0
gateway 192.168.1.254
iface eth0 inet6 static
address 2607:f0d0:1002:0011::2
netmask 64
gateway 2607:f0d0:1002:0011::1
iface eth1 inet static
address 192.168.1.6
netmask 255.255.255.0
network 192.168.0.0
broadcast 192.168.1.0
gateway 192.168.1.254
iface eth1 inet6 static
address 2607:f0d0:1002:0011::3
netmask 64
gateway 2607:f0d0:1002:0011::1
"""
V1_NET_CFG = {
"config": [
{
"name": "eth0",
"subnets": [
{
"address": "192.168.1.5",
"broadcast": "192.168.1.0",
"gateway": "192.168.1.254",
"netmask": "255.255.255.0",
"type": "static",
}
],
"type": "physical",
},
{
"name": "eth1",
"subnets": [{"control": "auto", "type": "dhcp4"}],
"type": "physical",
},
],
"version": 1,
}
V1_NET_CFG_WITH_DUPS = """\
# same value in interface specific dns and global dns
# should produce single entry in network file
version: 1
config:
- type: physical
name: eth0
subnets:
- type: static
address: 192.168.0.102/24
dns_nameservers: [1.2.3.4]
dns_search: [test.com]
interface: eth0
- type: nameserver
address: [1.2.3.4]
search: [test.com]
"""
V1_NET_CFG_OUTPUT = """\
# This file is generated from information provided by the datasource. Changes
# to it will not persist across an instance reboot. To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
auto lo
iface lo inet loopback
auto eth0
iface eth0 inet static
address 192.168.1.5/24
broadcast 192.168.1.0
gateway 192.168.1.254
auto eth1
iface eth1 inet dhcp
"""
V1_NET_CFG_IPV6_OUTPUT = """\
# This file is generated from information provided by the datasource. Changes
# to it will not persist across an instance reboot. To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
auto lo
iface lo inet loopback
auto eth0
iface eth0 inet6 static
address 2607:f0d0:1002:0011::2/64
gateway 2607:f0d0:1002:0011::1
auto eth1
iface eth1 inet dhcp
"""
V1_NET_CFG_IPV6 = {
"config": [
{
"name": "eth0",
"subnets": [
{
"address": "2607:f0d0:1002:0011::2",
"gateway": "2607:f0d0:1002:0011::1",
"netmask": "64",
"type": "static6",
}
],
"type": "physical",
},
{
"name": "eth1",
"subnets": [{"control": "auto", "type": "dhcp4"}],
"type": "physical",
},
],
"version": 1,
}
V1_TO_V2_NET_CFG_OUTPUT = """\
# This file is generated from information provided by the datasource. Changes
# to it will not persist across an instance reboot. To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
network:
version: 2
ethernets:
eth0:
addresses:
- 192.168.1.5/24
gateway4: 192.168.1.254
eth1:
dhcp4: true
"""
V1_TO_V2_NET_CFG_IPV6_OUTPUT = """\
# This file is generated from information provided by the datasource. Changes
# to it will not persist across an instance reboot. To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
network:
version: 2
ethernets:
eth0:
addresses:
- 2607:f0d0:1002:0011::2/64
gateway6: 2607:f0d0:1002:0011::1
eth1:
dhcp4: true
"""
V2_NET_CFG = {
"ethernets": {
"eth7": {"addresses": ["192.168.1.5/24"], "gateway4": "192.168.1.254"},
"eth9": {"dhcp4": True},
},
"version": 2,
}
V2_TO_V2_NET_CFG_OUTPUT = """\
# This file is generated from information provided by the datasource. Changes
# to it will not persist across an instance reboot. To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
network:
ethernets:
eth7:
addresses:
- 192.168.1.5/24
gateway4: 192.168.1.254
eth9:
dhcp4: true
version: 2
"""
class WriteBuffer(object):
def __init__(self):
self.buffer = StringIO()
self.mode = None
self.omode = None
def write(self, text):
self.buffer.write(text)
def __str__(self):
return self.buffer.getvalue()
class TestNetCfgDistroBase(FilesystemMockingTestCase):
def setUp(self):
super(TestNetCfgDistroBase, self).setUp()
self.add_patch("cloudinit.util.system_is_snappy", "m_snappy")
def _get_distro(self, dname, renderers=None):
cls = distros.fetch(dname)
cfg = settings.CFG_BUILTIN
cfg["system_info"]["distro"] = dname
if renderers:
cfg["system_info"]["network"] = {"renderers": renderers}
paths = helpers.Paths({})
return cls(dname, cfg.get("system_info"), paths)
def assertCfgEquals(self, blob1, blob2):
b1 = dict(SysConf(blob1.strip().splitlines()))
b2 = dict(SysConf(blob2.strip().splitlines()))
self.assertEqual(b1, b2)
for (k, v) in b1.items():
self.assertIn(k, b2)
for (k, v) in b2.items():
self.assertIn(k, b1)
for (k, v) in b1.items():
self.assertEqual(v, b2[k])
class TestNetCfgDistroFreeBSD(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroFreeBSD, self).setUp()
self.distro = self._get_distro("freebsd", renderers=["freebsd"])
def _apply_and_verify_freebsd(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.freebsd.available") as m_avail:
m_avail.return_value = True
with self.reRooted(tmpd) as tmpd:
util.ensure_dir("/etc")
util.ensure_file("/etc/rc.conf")
util.ensure_file("/etc/resolv.conf")
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
print("----------")
print(expected)
print("^^^^ expected | rendered VVVVVVV")
print(results[cfgpath])
print("----------")
self.assertEqual(
set(expected.split("\n")), set(results[cfgpath].split("\n"))
)
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
@mock.patch("cloudinit.net.get_interfaces_by_mac")
def test_apply_network_config_freebsd_standard(self, ifaces_mac):
ifaces_mac.return_value = {
"00:15:5d:4c:73:00": "eth0",
}
rc_conf_expected = """\
defaultrouter=192.168.1.254
ifconfig_eth0='192.168.1.5 netmask 255.255.255.0'
ifconfig_eth1=DHCP
"""
expected_cfgs = {
"/etc/rc.conf": rc_conf_expected,
"/etc/resolv.conf": "",
}
self._apply_and_verify_freebsd(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
@mock.patch("cloudinit.net.get_interfaces_by_mac")
def test_apply_network_config_freebsd_ifrename(self, ifaces_mac):
ifaces_mac.return_value = {
"00:15:5d:4c:73:00": "vtnet0",
}
rc_conf_expected = """\
ifconfig_vtnet0_name=eth0
defaultrouter=192.168.1.254
ifconfig_eth0='192.168.1.5 netmask 255.255.255.0'
ifconfig_eth1=DHCP
"""
V1_NET_CFG_RENAME = copy.deepcopy(V1_NET_CFG)
V1_NET_CFG_RENAME["config"][0]["mac_address"] = "00:15:5d:4c:73:00"
expected_cfgs = {
"/etc/rc.conf": rc_conf_expected,
"/etc/resolv.conf": "",
}
self._apply_and_verify_freebsd(
self.distro.apply_network_config,
V1_NET_CFG_RENAME,
expected_cfgs=expected_cfgs.copy(),
)
@mock.patch("cloudinit.net.get_interfaces_by_mac")
def test_apply_network_config_freebsd_nameserver(self, ifaces_mac):
ifaces_mac.return_value = {
"00:15:5d:4c:73:00": "eth0",
}
V1_NET_CFG_DNS = copy.deepcopy(V1_NET_CFG)
ns = ["1.2.3.4"]
V1_NET_CFG_DNS["config"][0]["subnets"][0]["dns_nameservers"] = ns
expected_cfgs = {"/etc/resolv.conf": "nameserver 1.2.3.4\n"}
self._apply_and_verify_freebsd(
self.distro.apply_network_config,
V1_NET_CFG_DNS,
expected_cfgs=expected_cfgs.copy(),
)
class TestNetCfgDistroUbuntuEni(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroUbuntuEni, self).setUp()
self.distro = self._get_distro("ubuntu", renderers=["eni"])
def eni_path(self):
return "/etc/network/interfaces.d/50-cloud-init.cfg"
def _apply_and_verify_eni(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.eni.available") as m_avail:
m_avail.return_value = True
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
print("----------")
print(expected)
print("^^^^ expected | rendered VVVVVVV")
print(results[cfgpath])
print("----------")
self.assertEqual(expected, results[cfgpath])
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def test_apply_network_config_eni_ub(self):
expected_cfgs = {
self.eni_path(): V1_NET_CFG_OUTPUT,
}
# ub_distro.apply_network_config(V1_NET_CFG, False)
self._apply_and_verify_eni(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
def test_apply_network_config_ipv6_ub(self):
expected_cfgs = {self.eni_path(): V1_NET_CFG_IPV6_OUTPUT}
self._apply_and_verify_eni(
self.distro.apply_network_config,
V1_NET_CFG_IPV6,
expected_cfgs=expected_cfgs.copy(),
)
class TestNetCfgDistroUbuntuNetplan(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroUbuntuNetplan, self).setUp()
self.distro = self._get_distro("ubuntu", renderers=["netplan"])
self.devlist = ["eth0", "lo"]
def _apply_and_verify_netplan(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.netplan.available", return_value=True):
with mock.patch(
"cloudinit.net.netplan.get_devicelist",
return_value=self.devlist,
):
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
print("----------")
print(expected)
print("^^^^ expected | rendered VVVVVVV")
print(results[cfgpath])
print("----------")
self.assertEqual(expected, results[cfgpath])
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def netplan_path(self):
return "/etc/netplan/50-cloud-init.yaml"
def test_apply_network_config_v1_to_netplan_ub(self):
expected_cfgs = {
self.netplan_path(): V1_TO_V2_NET_CFG_OUTPUT,
}
# ub_distro.apply_network_config(V1_NET_CFG, False)
self._apply_and_verify_netplan(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
def test_apply_network_config_v1_ipv6_to_netplan_ub(self):
expected_cfgs = {
self.netplan_path(): V1_TO_V2_NET_CFG_IPV6_OUTPUT,
}
# ub_distro.apply_network_config(V1_NET_CFG_IPV6, False)
self._apply_and_verify_netplan(
self.distro.apply_network_config,
V1_NET_CFG_IPV6,
expected_cfgs=expected_cfgs.copy(),
)
def test_apply_network_config_v2_passthrough_ub(self):
expected_cfgs = {
self.netplan_path(): V2_TO_V2_NET_CFG_OUTPUT,
}
# ub_distro.apply_network_config(V2_NET_CFG, False)
self._apply_and_verify_netplan(
self.distro.apply_network_config,
V2_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
class TestNetCfgDistroRedhat(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroRedhat, self).setUp()
self.distro = self._get_distro("rhel", renderers=["sysconfig"])
def ifcfg_path(self, ifname):
return "/etc/sysconfig/network-scripts/ifcfg-%s" % ifname
def control_path(self):
return "/etc/sysconfig/network"
def _apply_and_verify(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.sysconfig.available") as m_avail:
m_avail.return_value = True
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
self.assertCfgEquals(expected, results[cfgpath])
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def test_apply_network_config_rh(self):
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=none
DEFROUTE=yes
DEVICE=eth0
GATEWAY=192.168.1.254
IPADDR=192.168.1.5
NETMASK=255.255.255.0
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.ifcfg_path("eth1"): dedent(
"""\
BOOTPROTO=dhcp
DEVICE=eth1
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.control_path(): dedent(
"""\
NETWORKING=yes
"""
),
}
# rh_distro.apply_network_config(V1_NET_CFG, False)
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
def test_apply_network_config_ipv6_rh(self):
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=none
DEFROUTE=yes
DEVICE=eth0
IPV6ADDR=2607:f0d0:1002:0011::2/64
IPV6INIT=yes
IPV6_AUTOCONF=no
IPV6_DEFAULTGW=2607:f0d0:1002:0011::1
IPV6_FORCE_ACCEPT_RA=no
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.ifcfg_path("eth1"): dedent(
"""\
BOOTPROTO=dhcp
DEVICE=eth1
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.control_path(): dedent(
"""\
NETWORKING=yes
NETWORKING_IPV6=yes
IPV6_AUTOCONF=no
"""
),
}
# rh_distro.apply_network_config(V1_NET_CFG_IPV6, False)
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG_IPV6,
expected_cfgs=expected_cfgs.copy(),
)
def test_vlan_render_unsupported(self):
"""Render officially unsupported vlan names."""
cfg = {
"version": 2,
"ethernets": {
"eth0": {
"addresses": ["192.10.1.2/24"],
"match": {"macaddress": "00:16:3e:60:7c:df"},
}
},
"vlans": {
"infra0": {
"addresses": ["10.0.1.2/16"],
"id": 1001,
"link": "eth0",
}
},
}
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=none
DEVICE=eth0
HWADDR=00:16:3e:60:7c:df
IPADDR=192.10.1.2
NETMASK=255.255.255.0
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.ifcfg_path("infra0"): dedent(
"""\
BOOTPROTO=none
DEVICE=infra0
IPADDR=10.0.1.2
NETMASK=255.255.0.0
NM_CONTROLLED=no
ONBOOT=yes
PHYSDEV=eth0
USERCTL=no
VLAN=yes
"""
),
self.control_path(): dedent(
"""\
NETWORKING=yes
"""
),
}
self._apply_and_verify(
self.distro.apply_network_config, cfg, expected_cfgs=expected_cfgs
)
def test_vlan_render(self):
cfg = {
"version": 2,
"ethernets": {"eth0": {"addresses": ["192.10.1.2/24"]}},
"vlans": {
"eth0.1001": {
"addresses": ["10.0.1.2/16"],
"id": 1001,
"link": "eth0",
}
},
}
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=none
DEVICE=eth0
IPADDR=192.10.1.2
NETMASK=255.255.255.0
NM_CONTROLLED=no
ONBOOT=yes
TYPE=Ethernet
USERCTL=no
"""
),
self.ifcfg_path("eth0.1001"): dedent(
"""\
BOOTPROTO=none
DEVICE=eth0.1001
IPADDR=10.0.1.2
NETMASK=255.255.0.0
NM_CONTROLLED=no
ONBOOT=yes
PHYSDEV=eth0
USERCTL=no
VLAN=yes
"""
),
self.control_path(): dedent(
"""\
NETWORKING=yes
"""
),
}
self._apply_and_verify(
self.distro.apply_network_config, cfg, expected_cfgs=expected_cfgs
)
class TestNetCfgDistroOpensuse(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroOpensuse, self).setUp()
self.distro = self._get_distro("opensuse", renderers=["sysconfig"])
def ifcfg_path(self, ifname):
return "/etc/sysconfig/network/ifcfg-%s" % ifname
def _apply_and_verify(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.sysconfig.available") as m_avail:
m_avail.return_value = True
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
self.assertCfgEquals(expected, results[cfgpath])
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def test_apply_network_config_opensuse(self):
"""Opensuse uses apply_network_config and renders sysconfig"""
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=static
IPADDR=192.168.1.5
NETMASK=255.255.255.0
STARTMODE=auto
"""
),
self.ifcfg_path("eth1"): dedent(
"""\
BOOTPROTO=dhcp4
STARTMODE=auto
"""
),
}
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
)
def test_apply_network_config_ipv6_opensuse(self):
"""Opensuse uses apply_network_config and renders sysconfig w/ipv6"""
expected_cfgs = {
self.ifcfg_path("eth0"): dedent(
"""\
BOOTPROTO=static
IPADDR6=2607:f0d0:1002:0011::2/64
STARTMODE=auto
"""
),
self.ifcfg_path("eth1"): dedent(
"""\
BOOTPROTO=dhcp4
STARTMODE=auto
"""
),
}
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG_IPV6,
expected_cfgs=expected_cfgs.copy(),
)
class TestNetCfgDistroArch(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroArch, self).setUp()
self.distro = self._get_distro("arch", renderers=["netplan"])
def _apply_and_verify(
self,
apply_fn,
config,
expected_cfgs=None,
bringup=False,
with_netplan=False,
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch(
"cloudinit.net.netplan.available", return_value=with_netplan
):
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
print("----------")
print(expected)
print("^^^^ expected | rendered VVVVVVV")
print(results[cfgpath])
print("----------")
self.assertEqual(expected, results[cfgpath])
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def netctl_path(self, iface):
return "/etc/netctl/%s" % iface
def netplan_path(self):
return "/etc/netplan/50-cloud-init.yaml"
def test_apply_network_config_v1_without_netplan(self):
# Note that this is in fact an invalid netctl config:
# "Address=None/None"
# But this is what the renderer has been writing out for a long time,
# and the test's purpose is to assert that the netctl renderer is
# still being used in absence of netplan, not the correctness of the
# rendered netctl config.
expected_cfgs = {
self.netctl_path("eth0"): dedent(
"""\
Address=192.168.1.5/255.255.255.0
Connection=ethernet
DNS=()
Gateway=192.168.1.254
IP=static
Interface=eth0
"""
),
self.netctl_path("eth1"): dedent(
"""\
Address=None/None
Connection=ethernet
DNS=()
Gateway=
IP=dhcp
Interface=eth1
"""
),
}
# ub_distro.apply_network_config(V1_NET_CFG, False)
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
with_netplan=False,
)
def test_apply_network_config_v1_with_netplan(self):
expected_cfgs = {
self.netplan_path(): dedent(
"""\
# generated by cloud-init
network:
version: 2
ethernets:
eth0:
addresses:
- 192.168.1.5/24
gateway4: 192.168.1.254
eth1:
dhcp4: true
"""
),
}
with mock.patch(
"cloudinit.net.netplan.get_devicelist", return_value=[]
):
self._apply_and_verify(
self.distro.apply_network_config,
V1_NET_CFG,
expected_cfgs=expected_cfgs.copy(),
with_netplan=True,
)
class TestNetCfgDistroPhoton(TestNetCfgDistroBase):
def setUp(self):
super(TestNetCfgDistroPhoton, self).setUp()
self.distro = self._get_distro("photon", renderers=["networkd"])
def create_conf_dict(self, contents):
content_dict = {}
for line in contents:
if line:
line = line.strip()
if line and re.search(r"^\[(.+)\]$", line):
content_dict[line] = []
key = line
elif line:
assert key
content_dict[key].append(line)
return content_dict
def compare_dicts(self, actual, expected):
for k, v in actual.items():
self.assertEqual(sorted(expected[k]), sorted(v))
def _apply_and_verify(
self, apply_fn, config, expected_cfgs=None, bringup=False
):
if not expected_cfgs:
raise ValueError("expected_cfg must not be None")
tmpd = None
with mock.patch("cloudinit.net.networkd.available") as m_avail:
m_avail.return_value = True
with self.reRooted(tmpd) as tmpd:
apply_fn(config, bringup)
results = dir2dict(tmpd)
for cfgpath, expected in expected_cfgs.items():
actual = self.create_conf_dict(results[cfgpath].splitlines())
self.compare_dicts(actual, expected)
self.assertEqual(0o644, get_mode(cfgpath, tmpd))
def nwk_file_path(self, ifname):
return "/etc/systemd/network/10-cloud-init-%s.network" % ifname
def net_cfg_1(self, ifname):
ret = (
"""\
[Match]
Name=%s
[Network]
DHCP=no
[Address]
Address=192.168.1.5/24
[Route]
Gateway=192.168.1.254"""
% ifname
)
return ret
def net_cfg_2(self, ifname):
ret = (
"""\
[Match]
Name=%s
[Network]
DHCP=ipv4"""
% ifname
)
return ret
def test_photon_network_config_v1(self):
tmp = self.net_cfg_1("eth0").splitlines()
expected_eth0 = self.create_conf_dict(tmp)
tmp = self.net_cfg_2("eth1").splitlines()
expected_eth1 = self.create_conf_dict(tmp)
expected_cfgs = {
self.nwk_file_path("eth0"): expected_eth0,
self.nwk_file_path("eth1"): expected_eth1,
}
self._apply_and_verify(
self.distro.apply_network_config, V1_NET_CFG, expected_cfgs.copy()
)
def test_photon_network_config_v2(self):
tmp = self.net_cfg_1("eth7").splitlines()
expected_eth7 = self.create_conf_dict(tmp)
tmp = self.net_cfg_2("eth9").splitlines()
expected_eth9 = self.create_conf_dict(tmp)
expected_cfgs = {
self.nwk_file_path("eth7"): expected_eth7,
self.nwk_file_path("eth9"): expected_eth9,
}
self._apply_and_verify(
self.distro.apply_network_config, V2_NET_CFG, expected_cfgs.copy()
)
def test_photon_network_config_v1_with_duplicates(self):
expected = """\
[Match]
Name=eth0
[Network]
DHCP=no
DNS=1.2.3.4
Domains=test.com
[Address]
Address=192.168.0.102/24"""
net_cfg = safeyaml.load(V1_NET_CFG_WITH_DUPS)
expected = self.create_conf_dict(expected.splitlines())
expected_cfgs = {
self.nwk_file_path("eth0"): expected,
}
self._apply_and_verify(
self.distro.apply_network_config, net_cfg, expected_cfgs.copy()
)
def get_mode(path, target=None):
return os.stat(subp.target_path(target, path)).st_mode & 0o777
# vi: ts=4 expandtab