summaryrefslogtreecommitdiff
path: root/tests/unittests/net/test_init.py
diff options
context:
space:
mode:
authorJames Falcon <james.falcon@canonical.com>2021-12-15 20:16:38 -0600
committerGitHub <noreply@github.com>2021-12-15 19:16:38 -0700
commitbae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf (patch)
tree1fbb3269fc87e39832e3286ef42eefd2b23fcd44 /tests/unittests/net/test_init.py
parent2bcf4fa972fde686c2e3141c58e640640b44dd00 (diff)
downloadvyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.tar.gz
vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.zip
Adopt Black and isort (SC-700) (#1157)
Applied Black and isort, fixed any linting issues, updated tox.ini and CI.
Diffstat (limited to 'tests/unittests/net/test_init.py')
-rw-r--r--tests/unittests/net/test_init.py1368
1 files changed, 849 insertions, 519 deletions
diff --git a/tests/unittests/net/test_init.py b/tests/unittests/net/test_init.py
index 82854ab3..b245da94 100644
--- a/tests/unittests/net/test_init.py
+++ b/tests/unittests/net/test_init.py
@@ -13,24 +13,23 @@ import requests
import cloudinit.net as net
from cloudinit import safeyaml as yaml
-from tests.unittests.helpers import CiTestCase, HttprettyTestCase
from cloudinit.subp import ProcessExecutionError
from cloudinit.util import ensure_file, write_file
+from tests.unittests.helpers import CiTestCase, HttprettyTestCase
class TestSysDevPath(CiTestCase):
-
def test_sys_dev_path(self):
"""sys_dev_path returns a path under SYS_CLASS_NET for a device."""
- dev = 'something'
- path = 'attribute'
- expected = net.SYS_CLASS_NET + dev + '/' + path
+ dev = "something"
+ path = "attribute"
+ expected = net.SYS_CLASS_NET + dev + "/" + path
self.assertEqual(expected, net.sys_dev_path(dev, path))
def test_sys_dev_path_without_path(self):
"""When path param isn't provided it defaults to empty string."""
- dev = 'something'
- expected = net.SYS_CLASS_NET + dev + '/'
+ dev = "something"
+ expected = net.SYS_CLASS_NET + dev + "/"
self.assertEqual(expected, net.sys_dev_path(dev))
@@ -39,25 +38,25 @@ class TestReadSysNet(CiTestCase):
def setUp(self):
super(TestReadSysNet, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
def test_read_sys_net_strips_contents_of_sys_path(self):
"""read_sys_net strips whitespace from the contents of a sys file."""
- content = 'some stuff with trailing whitespace\t\r\n'
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
- self.assertEqual(content.strip(), net.read_sys_net('dev', 'attr'))
+ content = "some stuff with trailing whitespace\t\r\n"
+ write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+ self.assertEqual(content.strip(), net.read_sys_net("dev", "attr"))
def test_read_sys_net_reraises_oserror(self):
"""read_sys_net raises OSError/IOError when file doesn't exist."""
# Non-specific Exception because versions of python OSError vs IOError.
with self.assertRaises(Exception) as context_manager: # noqa: H202
- net.read_sys_net('dev', 'attr')
+ net.read_sys_net("dev", "attr")
error = context_manager.exception
- self.assertIn('No such file or directory', str(error))
+ self.assertIn("No such file or directory", str(error))
def test_read_sys_net_handles_error_with_on_enoent(self):
"""read_sys_net handles OSError/IOError with on_enoent if provided."""
@@ -66,43 +65,44 @@ class TestReadSysNet(CiTestCase):
def on_enoent(e):
handled_errors.append(e)
- net.read_sys_net('dev', 'attr', on_enoent=on_enoent)
+ net.read_sys_net("dev", "attr", on_enoent=on_enoent)
error = handled_errors[0]
self.assertIsInstance(error, Exception)
- self.assertIn('No such file or directory', str(error))
+ self.assertIn("No such file or directory", str(error))
def test_read_sys_net_translates_content(self):
"""read_sys_net translates content when translate dict is provided."""
content = "you're welcome\n"
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
- translate = {"you're welcome": 'de nada'}
+ write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+ translate = {"you're welcome": "de nada"}
self.assertEqual(
- 'de nada',
- net.read_sys_net('dev', 'attr', translate=translate))
+ "de nada", net.read_sys_net("dev", "attr", translate=translate)
+ )
def test_read_sys_net_errors_on_translation_failures(self):
"""read_sys_net raises a KeyError and logs details on failure."""
content = "you're welcome\n"
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
+ write_file(os.path.join(self.sysdir, "dev", "attr"), content)
with self.assertRaises(KeyError) as context_manager:
- net.read_sys_net('dev', 'attr', translate={})
+ net.read_sys_net("dev", "attr", translate={})
error = context_manager.exception
self.assertEqual('"you\'re welcome"', str(error))
self.assertIn(
"Found unexpected (not translatable) value 'you're welcome' in "
"'{0}dev/attr".format(self.sysdir),
- self.logs.getvalue())
+ self.logs.getvalue(),
+ )
def test_read_sys_net_handles_handles_with_onkeyerror(self):
"""read_sys_net handles translation errors calling on_keyerror."""
content = "you're welcome\n"
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
+ write_file(os.path.join(self.sysdir, "dev", "attr"), content)
handled_errors = []
def on_keyerror(e):
handled_errors.append(e)
- net.read_sys_net('dev', 'attr', translate={}, on_keyerror=on_keyerror)
+ net.read_sys_net("dev", "attr", translate={}, on_keyerror=on_keyerror)
error = handled_errors[0]
self.assertIsInstance(error, KeyError)
self.assertEqual('"you\'re welcome"', str(error))
@@ -110,274 +110,308 @@ class TestReadSysNet(CiTestCase):
def test_read_sys_net_safe_false_on_translate_failure(self):
"""read_sys_net_safe returns False on translation failures."""
content = "you're welcome\n"
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
- self.assertFalse(net.read_sys_net_safe('dev', 'attr', translate={}))
+ write_file(os.path.join(self.sysdir, "dev", "attr"), content)
+ self.assertFalse(net.read_sys_net_safe("dev", "attr", translate={}))
def test_read_sys_net_safe_returns_false_on_noent_failure(self):
"""read_sys_net_safe returns False on file not found failures."""
- self.assertFalse(net.read_sys_net_safe('dev', 'attr'))
+ self.assertFalse(net.read_sys_net_safe("dev", "attr"))
def test_read_sys_net_int_returns_none_on_error(self):
"""read_sys_net_safe returns None on failures."""
- self.assertFalse(net.read_sys_net_int('dev', 'attr'))
+ self.assertFalse(net.read_sys_net_int("dev", "attr"))
def test_read_sys_net_int_returns_none_on_valueerror(self):
"""read_sys_net_safe returns None when content is not an int."""
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), 'NOTINT\n')
- self.assertFalse(net.read_sys_net_int('dev', 'attr'))
+ write_file(os.path.join(self.sysdir, "dev", "attr"), "NOTINT\n")
+ self.assertFalse(net.read_sys_net_int("dev", "attr"))
def test_read_sys_net_int_returns_integer_from_content(self):
"""read_sys_net_safe returns None on failures."""
- write_file(os.path.join(self.sysdir, 'dev', 'attr'), '1\n')
- self.assertEqual(1, net.read_sys_net_int('dev', 'attr'))
+ write_file(os.path.join(self.sysdir, "dev", "attr"), "1\n")
+ self.assertEqual(1, net.read_sys_net_int("dev", "attr"))
def test_is_up_true(self):
"""is_up is True if sys/net/devname/operstate is 'up' or 'unknown'."""
- for state in ['up', 'unknown']:
- write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
- self.assertTrue(net.is_up('eth0'))
+ for state in ["up", "unknown"]:
+ write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
+ self.assertTrue(net.is_up("eth0"))
def test_is_up_false(self):
"""is_up is False if sys/net/devname/operstate is 'down' or invalid."""
- for state in ['down', 'incomprehensible']:
- write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
- self.assertFalse(net.is_up('eth0'))
+ for state in ["down", "incomprehensible"]:
+ write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
+ self.assertFalse(net.is_up("eth0"))
def test_is_bridge(self):
"""is_bridge is True when /sys/net/devname/bridge exists."""
- self.assertFalse(net.is_bridge('eth0'))
- ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
- self.assertTrue(net.is_bridge('eth0'))
+ self.assertFalse(net.is_bridge("eth0"))
+ ensure_file(os.path.join(self.sysdir, "eth0", "bridge"))
+ self.assertTrue(net.is_bridge("eth0"))
def test_is_bond(self):
"""is_bond is True when /sys/net/devname/bonding exists."""
- self.assertFalse(net.is_bond('eth0'))
- ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
- self.assertTrue(net.is_bond('eth0'))
+ self.assertFalse(net.is_bond("eth0"))
+ ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
+ self.assertTrue(net.is_bond("eth0"))
def test_get_master(self):
"""get_master returns the path when /sys/net/devname/master exists."""
- self.assertIsNone(net.get_master('enP1s1'))
- master_path = os.path.join(self.sysdir, 'enP1s1', 'master')
+ self.assertIsNone(net.get_master("enP1s1"))
+ master_path = os.path.join(self.sysdir, "enP1s1", "master")
ensure_file(master_path)
- self.assertEqual(master_path, net.get_master('enP1s1'))
+ self.assertEqual(master_path, net.get_master("enP1s1"))
def test_master_is_bridge_or_bond(self):
- bridge_mac = 'aa:bb:cc:aa:bb:cc'
- bond_mac = 'cc:bb:aa:cc:bb:aa'
+ bridge_mac = "aa:bb:cc:aa:bb:cc"
+ bond_mac = "cc:bb:aa:cc:bb:aa"
# No master => False
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
- write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
+ write_file(os.path.join(self.sysdir, "eth1", "address"), bridge_mac)
+ write_file(os.path.join(self.sysdir, "eth2", "address"), bond_mac)
- self.assertFalse(net.master_is_bridge_or_bond('eth1'))
- self.assertFalse(net.master_is_bridge_or_bond('eth2'))
+ self.assertFalse(net.master_is_bridge_or_bond("eth1"))
+ self.assertFalse(net.master_is_bridge_or_bond("eth2"))
# masters without bridge/bonding => False
- write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
- write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
+ write_file(os.path.join(self.sysdir, "br0", "address"), bridge_mac)
+ write_file(os.path.join(self.sysdir, "bond0", "address"), bond_mac)
- os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
- os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
+ os.symlink("../br0", os.path.join(self.sysdir, "eth1", "master"))
+ os.symlink("../bond0", os.path.join(self.sysdir, "eth2", "master"))
- self.assertFalse(net.master_is_bridge_or_bond('eth1'))
- self.assertFalse(net.master_is_bridge_or_bond('eth2'))
+ self.assertFalse(net.master_is_bridge_or_bond("eth1"))
+ self.assertFalse(net.master_is_bridge_or_bond("eth2"))
# masters with bridge/bonding => True
- write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
- write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
+ write_file(os.path.join(self.sysdir, "br0", "bridge"), "")
+ write_file(os.path.join(self.sysdir, "bond0", "bonding"), "")
- self.assertTrue(net.master_is_bridge_or_bond('eth1'))
- self.assertTrue(net.master_is_bridge_or_bond('eth2'))
+ self.assertTrue(net.master_is_bridge_or_bond("eth1"))
+ self.assertTrue(net.master_is_bridge_or_bond("eth2"))
def test_master_is_openvswitch(self):
- ovs_mac = 'bb:cc:aa:bb:cc:aa'
+ ovs_mac = "bb:cc:aa:bb:cc:aa"
# No master => False
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), ovs_mac)
+ write_file(os.path.join(self.sysdir, "eth1", "address"), ovs_mac)
- self.assertFalse(net.master_is_bridge_or_bond('eth1'))
+ self.assertFalse(net.master_is_bridge_or_bond("eth1"))
# masters without ovs-system => False
- write_file(os.path.join(self.sysdir, 'ovs-system', 'address'), ovs_mac)
+ write_file(os.path.join(self.sysdir, "ovs-system", "address"), ovs_mac)
- os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth1',
- 'master'))
+ os.symlink(
+ "../ovs-system", os.path.join(self.sysdir, "eth1", "master")
+ )
- self.assertFalse(net.master_is_openvswitch('eth1'))
+ self.assertFalse(net.master_is_openvswitch("eth1"))
# masters with ovs-system => True
- os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth1',
- 'upper_ovs-system'))
+ os.symlink(
+ "../ovs-system",
+ os.path.join(self.sysdir, "eth1", "upper_ovs-system"),
+ )
- self.assertTrue(net.master_is_openvswitch('eth1'))
+ self.assertTrue(net.master_is_openvswitch("eth1"))
def test_is_vlan(self):
"""is_vlan is True when /sys/net/devname/uevent has DEVTYPE=vlan."""
- ensure_file(os.path.join(self.sysdir, 'eth0', 'uevent'))
- self.assertFalse(net.is_vlan('eth0'))
- content = 'junk\nDEVTYPE=vlan\njunk\n'
- write_file(os.path.join(self.sysdir, 'eth0', 'uevent'), content)
- self.assertTrue(net.is_vlan('eth0'))
+ ensure_file(os.path.join(self.sysdir, "eth0", "uevent"))
+ self.assertFalse(net.is_vlan("eth0"))
+ content = "junk\nDEVTYPE=vlan\njunk\n"
+ write_file(os.path.join(self.sysdir, "eth0", "uevent"), content)
+ self.assertTrue(net.is_vlan("eth0"))
class TestGenerateFallbackConfig(CiTestCase):
-
def setUp(self):
super(TestGenerateFallbackConfig, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
- self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
- return_value=False)
- self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
- self.add_patch('cloudinit.net.is_netfailover', 'm_netfail',
- return_value=False)
- self.add_patch('cloudinit.net.is_netfail_master', 'm_netfail_master',
- return_value=False)
+ self.add_patch(
+ "cloudinit.net.util.is_container",
+ "m_is_container",
+ return_value=False,
+ )
+ self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle")
+ self.add_patch(
+ "cloudinit.net.is_netfailover", "m_netfail", return_value=False
+ )
+ self.add_patch(
+ "cloudinit.net.is_netfail_master",
+ "m_netfail_master",
+ return_value=False,
+ )
def test_generate_fallback_finds_connected_eth_with_mac(self):
"""generate_fallback_config finds any connected device with a mac."""
- write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
- write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
+ write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+ write_file(os.path.join(self.sysdir, "eth1", "carrier"), "1")
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
expected = {
- 'ethernets': {'eth1': {'match': {'macaddress': mac},
- 'dhcp4': True, 'set-name': 'eth1'}},
- 'version': 2}
+ "ethernets": {
+ "eth1": {
+ "match": {"macaddress": mac},
+ "dhcp4": True,
+ "set-name": "eth1",
+ }
+ },
+ "version": 2,
+ }
self.assertEqual(expected, net.generate_fallback_config())
def test_generate_fallback_finds_dormant_eth_with_mac(self):
"""generate_fallback_config finds any dormant device with a mac."""
- write_file(os.path.join(self.sysdir, 'eth0', 'dormant'), '1')
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
+ write_file(os.path.join(self.sysdir, "eth0", "dormant"), "1")
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
expected = {
- 'ethernets': {'eth0': {'match': {'macaddress': mac}, 'dhcp4': True,
- 'set-name': 'eth0'}},
- 'version': 2}
+ "ethernets": {
+ "eth0": {
+ "match": {"macaddress": mac},
+ "dhcp4": True,
+ "set-name": "eth0",
+ }
+ },
+ "version": 2,
+ }
self.assertEqual(expected, net.generate_fallback_config())
def test_generate_fallback_finds_eth_by_operstate(self):
"""generate_fallback_config finds any dormant device with a mac."""
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
expected = {
- 'ethernets': {
- 'eth0': {'dhcp4': True, 'match': {'macaddress': mac},
- 'set-name': 'eth0'}},
- 'version': 2}
- valid_operstates = ['dormant', 'down', 'lowerlayerdown', 'unknown']
+ "ethernets": {
+ "eth0": {
+ "dhcp4": True,
+ "match": {"macaddress": mac},
+ "set-name": "eth0",
+ }
+ },
+ "version": 2,
+ }
+ valid_operstates = ["dormant", "down", "lowerlayerdown", "unknown"]
for state in valid_operstates:
- write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
+ write_file(os.path.join(self.sysdir, "eth0", "operstate"), state)
self.assertEqual(expected, net.generate_fallback_config())
- write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'noworky')
+ write_file(os.path.join(self.sysdir, "eth0", "operstate"), "noworky")
self.assertIsNone(net.generate_fallback_config())
def test_generate_fallback_config_skips_veth(self):
"""generate_fallback_config will skip any veth interfaces."""
# A connected veth which gets ignored
- write_file(os.path.join(self.sysdir, 'veth0', 'carrier'), '1')
+ write_file(os.path.join(self.sysdir, "veth0", "carrier"), "1")
self.assertIsNone(net.generate_fallback_config())
def test_generate_fallback_config_skips_bridges(self):
"""generate_fallback_config will skip any bridges interfaces."""
# A connected veth which gets ignored
- write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
- ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
+ write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+ ensure_file(os.path.join(self.sysdir, "eth0", "bridge"))
self.assertIsNone(net.generate_fallback_config())
def test_generate_fallback_config_skips_bonds(self):
"""generate_fallback_config will skip any bonded interfaces."""
# A connected veth which gets ignored
- write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
- ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
+ write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth0", "address"), mac)
+ ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
self.assertIsNone(net.generate_fallback_config())
def test_generate_fallback_config_skips_netfail_devs(self):
"""gen_fallback_config ignores netfail primary,sby no mac on master."""
- mac = 'aa:bb:cc:aa:bb:cc' # netfailover devs share the same mac
- for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
- write_file(os.path.join(self.sysdir, iface, 'carrier'), '1')
- write_file(
- os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
+ mac = "aa:bb:cc:aa:bb:cc" # netfailover devs share the same mac
+ for iface in ["ens3", "ens3sby", "enP0s1f3"]:
+ write_file(os.path.join(self.sysdir, iface, "carrier"), "1")
write_file(
- os.path.join(self.sysdir, iface, 'address'), mac)
+ os.path.join(self.sysdir, iface, "addr_assign_type"), "0"
+ )
+ write_file(os.path.join(self.sysdir, iface, "address"), mac)
def is_netfail(iface, _driver=None):
# ens3 is the master
- if iface == 'ens3':
+ if iface == "ens3":
return False
return True
+
self.m_netfail.side_effect = is_netfail
def is_netfail_master(iface, _driver=None):
# ens3 is the master
- if iface == 'ens3':
+ if iface == "ens3":
return True
return False
+
self.m_netfail_master.side_effect = is_netfail_master
expected = {
- 'ethernets': {
- 'ens3': {'dhcp4': True, 'match': {'name': 'ens3'},
- 'set-name': 'ens3'}},
- 'version': 2}
+ "ethernets": {
+ "ens3": {
+ "dhcp4": True,
+ "match": {"name": "ens3"},
+ "set-name": "ens3",
+ }
+ },
+ "version": 2,
+ }
result = net.generate_fallback_config()
self.assertEqual(expected, result)
class TestNetFindFallBackNic(CiTestCase):
-
def setUp(self):
super(TestNetFindFallBackNic, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
- self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
- return_value=False)
- self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
+ self.add_patch(
+ "cloudinit.net.util.is_container",
+ "m_is_container",
+ return_value=False,
+ )
+ self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle")
def test_generate_fallback_finds_first_connected_eth_with_mac(self):
"""find_fallback_nic finds any connected device with a mac."""
- write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
- write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
- self.assertEqual('eth1', net.find_fallback_nic())
+ write_file(os.path.join(self.sysdir, "eth0", "carrier"), "1")
+ write_file(os.path.join(self.sysdir, "eth1", "carrier"), "1")
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
+ self.assertEqual("eth1", net.find_fallback_nic())
class TestGetDeviceList(CiTestCase):
-
def setUp(self):
super(TestGetDeviceList, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
def test_get_devicelist_raise_oserror(self):
"""get_devicelist raise any non-ENOENT OSerror."""
- error = OSError('Can not do it')
+ error = OSError("Can not do it")
error.errno = errno.EPERM # Set non-ENOENT
self.m_sys_path.side_effect = error
with self.assertRaises(OSError) as context_manager:
net.get_devicelist()
exception = context_manager.exception
- self.assertEqual('Can not do it', str(exception))
+ self.assertEqual("Can not do it", str(exception))
def test_get_devicelist_empty_without_sys_net(self):
"""get_devicelist returns empty list when missing SYS_CLASS_NET."""
- self.m_sys_path.return_value = 'idontexist'
+ self.m_sys_path.return_value = "idontexist"
self.assertEqual([], net.get_devicelist())
def test_get_devicelist_empty_with_no_devices_in_sys_net(self):
@@ -386,9 +420,9 @@ class TestGetDeviceList(CiTestCase):
def test_get_devicelist_lists_any_subdirectories_in_sys_net(self):
"""get_devicelist returns a directory listing for SYS_CLASS_NET."""
- write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'up')
- write_file(os.path.join(self.sysdir, 'eth1', 'operstate'), 'up')
- self.assertCountEqual(['eth0', 'eth1'], net.get_devicelist())
+ write_file(os.path.join(self.sysdir, "eth0", "operstate"), "up")
+ write_file(os.path.join(self.sysdir, "eth1", "operstate"), "up")
+ self.assertCountEqual(["eth0", "eth1"], net.get_devicelist())
@mock.patch(
@@ -396,239 +430,288 @@ class TestGetDeviceList(CiTestCase):
mock.Mock(return_value=False),
)
class TestGetInterfaceMAC(CiTestCase):
-
def setUp(self):
super(TestGetInterfaceMAC, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
def test_get_interface_mac_false_with_no_mac(self):
"""get_device_list returns False when no mac is reported."""
- ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
- mac_path = os.path.join(self.sysdir, 'eth0', 'address')
+ ensure_file(os.path.join(self.sysdir, "eth0", "bonding"))
+ mac_path = os.path.join(self.sysdir, "eth0", "address")
self.assertFalse(os.path.exists(mac_path))
- self.assertFalse(net.get_interface_mac('eth0'))
+ self.assertFalse(net.get_interface_mac("eth0"))
def test_get_interface_mac(self):
"""get_interfaces returns the mac from SYS_CLASS_NET/dev/address."""
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
- self.assertEqual(mac, net.get_interface_mac('eth1'))
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth1", "address"), mac)
+ self.assertEqual(mac, net.get_interface_mac("eth1"))
def test_get_interface_mac_grabs_bonding_address(self):
"""get_interfaces returns the source device mac for bonded devices."""
- source_dev_mac = 'aa:bb:cc:aa:bb:cc'
- bonded_mac = 'dd:ee:ff:dd:ee:ff'
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), bonded_mac)
+ source_dev_mac = "aa:bb:cc:aa:bb:cc"
+ bonded_mac = "dd:ee:ff:dd:ee:ff"
+ write_file(os.path.join(self.sysdir, "eth1", "address"), bonded_mac)
write_file(
- os.path.join(self.sysdir, 'eth1', 'bonding_slave', 'perm_hwaddr'),
- source_dev_mac)
- self.assertEqual(source_dev_mac, net.get_interface_mac('eth1'))
+ os.path.join(self.sysdir, "eth1", "bonding_slave", "perm_hwaddr"),
+ source_dev_mac,
+ )
+ self.assertEqual(source_dev_mac, net.get_interface_mac("eth1"))
def test_get_interfaces_empty_list_without_sys_net(self):
"""get_interfaces returns an empty list when missing SYS_CLASS_NET."""
- self.m_sys_path.return_value = 'idontexist'
+ self.m_sys_path.return_value = "idontexist"
self.assertEqual([], net.get_interfaces())
def test_get_interfaces_by_mac_skips_empty_mac(self):
"""Ignore 00:00:00:00:00:00 addresses from get_interfaces_by_mac."""
- empty_mac = '00:00:00:00:00:00'
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), empty_mac)
- write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
- write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
- write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
- expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
+ empty_mac = "00:00:00:00:00:00"
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth1", "address"), empty_mac)
+ write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+ write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+ write_file(os.path.join(self.sysdir, "eth2", "address"), mac)
+ expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)]
self.assertEqual(expected, net.get_interfaces())
def test_get_interfaces_by_mac_skips_missing_mac(self):
"""Ignore interfaces without an address from get_interfaces_by_mac."""
- write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
- address_path = os.path.join(self.sysdir, 'eth1', 'address')
+ write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+ address_path = os.path.join(self.sysdir, "eth1", "address")
self.assertFalse(os.path.exists(address_path))
- mac = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
- write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
- expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
+ mac = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+ write_file(os.path.join(self.sysdir, "eth2", "address"), mac)
+ expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)]
self.assertEqual(expected, net.get_interfaces())
def test_get_interfaces_by_mac_skips_master_devs(self):
"""Ignore interfaces with a master device which would have dup mac."""
- mac1 = mac2 = 'aa:bb:cc:aa:bb:cc'
- write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac1)
- write_file(os.path.join(self.sysdir, 'eth1', 'master'), "blah")
- write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
- write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac2)
- expected = [('eth2', mac2, None, None)]
+ mac1 = mac2 = "aa:bb:cc:aa:bb:cc"
+ write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0")
+ write_file(os.path.join(self.sysdir, "eth1", "address"), mac1)
+ write_file(os.path.join(self.sysdir, "eth1", "master"), "blah")
+ write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0")
+ write_file(os.path.join(self.sysdir, "eth2", "address"), mac2)
+ expected = [("eth2", mac2, None, None)]
self.assertEqual(expected, net.get_interfaces())
- @mock.patch('cloudinit.net.is_netfailover')
+ @mock.patch("cloudinit.net.is_netfailover")
def test_get_interfaces_by_mac_skips_netfailvoer(self, m_netfail):
"""Ignore interfaces if netfailover primary or standby."""
- mac = 'aa:bb:cc:aa:bb:cc' # netfailover devs share the same mac
- for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
+ mac = "aa:bb:cc:aa:bb:cc" # netfailover devs share the same mac
+ for iface in ["ens3", "ens3sby", "enP0s1f3"]:
write_file(
- os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
- write_file(
- os.path.join(self.sysdir, iface, 'address'), mac)
+ os.path.join(self.sysdir, iface, "addr_assign_type"), "0"
+ )
+ write_file(os.path.join(self.sysdir, iface, "address"), mac)
def is_netfail(iface, _driver=None):
# ens3 is the master
- if iface == 'ens3':
+ if iface == "ens3":
return False
else:
return True
+
m_netfail.side_effect = is_netfail
- expected = [('ens3', mac, None, None)]
+ expected = [("ens3", mac, None, None)]
self.assertEqual(expected, net.get_interfaces())
def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds(
- self
+ self,
):
- bridge_mac = 'aa:bb:cc:aa:bb:cc'
- bond_mac = 'cc:bb:aa:cc:bb:aa'
- ovs_mac = 'bb:cc:aa:bb:cc:aa'
+ bridge_mac = "aa:bb:cc:aa:bb:cc"
+ bond_mac = "cc:bb:aa:cc:bb:aa"
+ ovs_mac = "bb:cc:aa:bb:cc:aa"
- write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
- write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
+ write_file(os.path.join(self.sysdir, "br0", "address"), bridge_mac)
+ write_file(os.path.join(self.sysdir, "br0", "bridge"), "")
- write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
- write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
+ write_file(os.path.join(self.sysdir, "bond0", "address"), bond_mac)
+ write_file(os.path.join(self.sysdir, "bond0", "bonding"), "")
- write_file(os.path.join(self.sysdir, 'ovs-system', 'address'),
- ovs_mac)
+ write_file(os.path.join(self.sysdir, "ovs-system", "address"), ovs_mac)
- write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
- os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
+ write_file(os.path.join(self.sysdir, "eth1", "address"), bridge_mac)
+ os.symlink("../br0", os.path.join(self.sysdir, "eth1", "master"))
- write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
- os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
+ write_file(os.path.join(self.sysdir, "eth2", "address"), bond_mac)
+ os.symlink("../bond0", os.path.join(self.sysdir, "eth2", "master"))
- write_file(os.path.join(self.sysdir, 'eth3', 'address'), ovs_mac)
- os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth3',
- 'master'))
- os.symlink('../ovs-system', os.path.join(self.sysdir, 'eth3',
- 'upper_ovs-system'))
+ write_file(os.path.join(self.sysdir, "eth3", "address"), ovs_mac)
+ os.symlink(
+ "../ovs-system", os.path.join(self.sysdir, "eth3", "master")
+ )
+ os.symlink(
+ "../ovs-system",
+ os.path.join(self.sysdir, "eth3", "upper_ovs-system"),
+ )
interface_names = [interface[0] for interface in net.get_interfaces()]
- self.assertEqual(['eth1', 'eth2', 'eth3', 'ovs-system'],
- sorted(interface_names))
+ self.assertEqual(
+ ["eth1", "eth2", "eth3", "ovs-system"], sorted(interface_names)
+ )
class TestInterfaceHasOwnMAC(CiTestCase):
-
def setUp(self):
super(TestInterfaceHasOwnMAC, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
def test_interface_has_own_mac_false_when_stolen(self):
"""Return False from interface_has_own_mac when address is stolen."""
- write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '2')
- self.assertFalse(net.interface_has_own_mac('eth1'))
+ write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "2")
+ self.assertFalse(net.interface_has_own_mac("eth1"))
def test_interface_has_own_mac_true_when_not_stolen(self):
"""Return False from interface_has_own_mac when mac isn't stolen."""
- valid_assign_types = ['0', '1', '3']
- assign_path = os.path.join(self.sysdir, 'eth1', 'addr_assign_type')
+ valid_assign_types = ["0", "1", "3"]
+ assign_path = os.path.join(self.sysdir, "eth1", "addr_assign_type")
for _type in valid_assign_types:
write_file(assign_path, _type)
- self.assertTrue(net.interface_has_own_mac('eth1'))
+ self.assertTrue(net.interface_has_own_mac("eth1"))
def test_interface_has_own_mac_strict_errors_on_absent_assign_type(self):
"""When addr_assign_type is absent, interface_has_own_mac errors."""
with self.assertRaises(ValueError):
- net.interface_has_own_mac('eth1', strict=True)
+ net.interface_has_own_mac("eth1", strict=True)
-@mock.patch('cloudinit.net.subp.subp')
+@mock.patch("cloudinit.net.subp.subp")
class TestEphemeralIPV4Network(CiTestCase):
with_logs = True
def setUp(self):
super(TestEphemeralIPV4Network, self).setUp()
- sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
+ sys_mock = mock.patch("cloudinit.net.get_sys_class_path")
self.m_sys_path = sys_mock.start()
- self.sysdir = self.tmp_dir() + '/'
+ self.sysdir = self.tmp_dir() + "/"
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp):
"""No required params for EphemeralIPv4Network can be None."""
required_params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.0",
+ "broadcast": "192.168.2.255",
+ }
for key in required_params.keys():
params = copy.deepcopy(required_params)
params[key] = None
with self.assertRaises(ValueError) as context_manager:
net.EphemeralIPv4Network(**params)
error = context_manager.exception
- self.assertIn('Cannot init network on', str(error))
+ self.assertIn("Cannot init network on", str(error))
self.assertEqual(0, m_subp.call_count)
def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp):
"""Raise an error when prefix_or_mask is not a netmask or prefix."""
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'broadcast': '192.168.2.255'}
- invalid_masks = ('invalid', 'invalid.', '123.123.123')
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "broadcast": "192.168.2.255",
+ }
+ invalid_masks = ("invalid", "invalid.", "123.123.123")
for error_val in invalid_masks:
- params['prefix_or_mask'] = error_val
+ params["prefix_or_mask"] = error_val
with self.assertRaises(ValueError) as context_manager:
with net.EphemeralIPv4Network(**params):
pass
error = context_manager.exception
- self.assertIn('Cannot setup network: netmask', str(error))
+ self.assertIn("Cannot setup network: netmask", str(error))
self.assertEqual(0, m_subp.call_count)
def test_ephemeral_ipv4_network_performs_teardown(self, m_subp):
"""EphemeralIPv4Network performs teardown on the device if setup."""
expected_setup_calls = [
mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'}),
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/24",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ ),
mock.call(
- ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
- capture=True)]
+ ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+ capture=True,
+ ),
+ ]
expected_teardown_calls = [
mock.call(
- ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0',
- 'down'], capture=True),
+ [
+ "ip",
+ "-family",
+ "inet",
+ "link",
+ "set",
+ "dev",
+ "eth0",
+ "down",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-family', 'inet', 'addr', 'del', '192.168.2.2/24',
- 'dev', 'eth0'], capture=True)]
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "del",
+ "192.168.2.2/24",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
+ ]
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.0",
+ "broadcast": "192.168.2.255",
+ }
with net.EphemeralIPv4Network(**params):
self.assertEqual(expected_setup_calls, m_subp.call_args_list)
m_subp.assert_has_calls(expected_teardown_calls)
- @mock.patch('cloudinit.net.readurl')
+ @mock.patch("cloudinit.net.readurl")
def test_ephemeral_ipv4_no_network_if_url_connectivity(
- self, m_readurl, m_subp):
+ self, m_readurl, m_subp
+ ):
"""No network setup is performed if we can successfully connect to
connectivity_url."""
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
- 'connectivity_url_data': {'url': 'http://example.org/index.html'}
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.0",
+ "broadcast": "192.168.2.255",
+ "connectivity_url_data": {"url": "http://example.org/index.html"},
}
with net.EphemeralIPv4Network(**params):
self.assertEqual(
- [mock.call(url='http://example.org/index.html', timeout=5)],
- m_readurl.call_args_list
+ [mock.call(url="http://example.org/index.html", timeout=5)],
+ m_readurl.call_args_list,
)
# Ensure that no teardown happens:
m_subp.assert_has_calls([])
@@ -639,67 +722,173 @@ class TestEphemeralIPV4Network(CiTestCase):
It performs no cleanup as the interface was already setup.
"""
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.0",
+ "broadcast": "192.168.2.255",
+ }
m_subp.side_effect = ProcessExecutionError(
- '', 'RTNETLINK answers: File exists', 2)
+ "", "RTNETLINK answers: File exists", 2
+ )
expected_calls = [
mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'})]
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/24",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ )
+ ]
with net.EphemeralIPv4Network(**params):
pass
self.assertEqual(expected_calls, m_subp.call_args_list)
self.assertIn(
- 'Skip ephemeral network setup, eth0 already has address',
- self.logs.getvalue())
+ "Skip ephemeral network setup, eth0 already has address",
+ self.logs.getvalue(),
+ )
def test_ephemeral_ipv4_network_with_prefix(self, m_subp):
"""EphemeralIPv4Network takes a valid prefix to setup the network."""
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '24', 'broadcast': '192.168.2.255'}
- for prefix_val in ['24', 16]: # prefix can be int or string
- params['prefix_or_mask'] = prefix_val
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "24",
+ "broadcast": "192.168.2.255",
+ }
+ for prefix_val in ["24", 16]: # prefix can be int or string
+ params["prefix_or_mask"] = prefix_val
with net.EphemeralIPv4Network(**params):
pass
- m_subp.assert_has_calls([mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'})])
- m_subp.assert_has_calls([mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/16',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'})])
+ m_subp.assert_has_calls(
+ [
+ mock.call(
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/24",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ )
+ ]
+ )
+ m_subp.assert_has_calls(
+ [
+ mock.call(
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/16",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ )
+ ]
+ )
def test_ephemeral_ipv4_network_with_new_default_route(self, m_subp):
"""Add the route when router is set and no default route exists."""
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
- 'router': '192.168.2.1'}
- m_subp.return_value = '', '' # Empty response from ip route gw check
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.0",
+ "broadcast": "192.168.2.255",
+ "router": "192.168.2.1",
+ }
+ m_subp.return_value = "", "" # Empty response from ip route gw check
expected_setup_calls = [
mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'}),
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/24",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ ),
mock.call(
- ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
- capture=True),
+ ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+ capture=True,
+ ),
+ mock.call(["ip", "route", "show", "0.0.0.0/0"], capture=True),
mock.call(
- ['ip', 'route', 'show', '0.0.0.0/0'], capture=True),
- mock.call(['ip', '-4', 'route', 'add', '192.168.2.1',
- 'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
+ [
+ "ip",
+ "-4",
+ "route",
+ "add",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ "src",
+ "192.168.2.2",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'add', 'default', 'via',
- '192.168.2.1', 'dev', 'eth0'], capture=True)]
+ [
+ "ip",
+ "-4",
+ "route",
+ "add",
+ "default",
+ "via",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
+ ]
expected_teardown_calls = [
- mock.call(['ip', '-4', 'route', 'del', 'default', 'dev', 'eth0'],
- capture=True),
- mock.call(['ip', '-4', 'route', 'del', '192.168.2.1',
- 'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
+ mock.call(
+ ["ip", "-4", "route", "del", "default", "dev", "eth0"],
+ capture=True,
+ ),
+ mock.call(
+ [
+ "ip",
+ "-4",
+ "route",
+ "del",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ "src",
+ "192.168.2.2",
+ ],
+ capture=True,
+ ),
]
with net.EphemeralIPv4Network(**params):
@@ -708,45 +897,138 @@ class TestEphemeralIPV4Network(CiTestCase):
def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp):
params = {
- 'interface': 'eth0', 'ip': '192.168.2.2',
- 'prefix_or_mask': '255.255.255.255', 'broadcast': '192.168.2.255',
- 'static_routes': [('192.168.2.1/32', '0.0.0.0'),
- ('169.254.169.254/32', '192.168.2.1'),
- ('0.0.0.0/0', '192.168.2.1')],
- 'router': '192.168.2.1'}
+ "interface": "eth0",
+ "ip": "192.168.2.2",
+ "prefix_or_mask": "255.255.255.255",
+ "broadcast": "192.168.2.255",
+ "static_routes": [
+ ("192.168.2.1/32", "0.0.0.0"),
+ ("169.254.169.254/32", "192.168.2.1"),
+ ("0.0.0.0/0", "192.168.2.1"),
+ ],
+ "router": "192.168.2.1",
+ }
expected_setup_calls = [
mock.call(
- ['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/32',
- 'broadcast', '192.168.2.255', 'dev', 'eth0'],
- capture=True, update_env={'LANG': 'C'}),
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "add",
+ "192.168.2.2/32",
+ "broadcast",
+ "192.168.2.255",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ update_env={"LANG": "C"},
+ ),
mock.call(
- ['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
- capture=True),
+ ["ip", "-family", "inet", "link", "set", "dev", "eth0", "up"],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'append', '192.168.2.1/32',
- 'dev', 'eth0'], capture=True),
+ [
+ "ip",
+ "-4",
+ "route",
+ "append",
+ "192.168.2.1/32",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'append', '169.254.169.254/32',
- 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
+ [
+ "ip",
+ "-4",
+ "route",
+ "append",
+ "169.254.169.254/32",
+ "via",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'append', '0.0.0.0/0',
- 'via', '192.168.2.1', 'dev', 'eth0'], capture=True)]
+ [
+ "ip",
+ "-4",
+ "route",
+ "append",
+ "0.0.0.0/0",
+ "via",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
+ ]
expected_teardown_calls = [
mock.call(
- ['ip', '-4', 'route', 'del', '0.0.0.0/0',
- 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
+ [
+ "ip",
+ "-4",
+ "route",
+ "del",
+ "0.0.0.0/0",
+ "via",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'del', '169.254.169.254/32',
- 'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
+ [
+ "ip",
+ "-4",
+ "route",
+ "del",
+ "169.254.169.254/32",
+ "via",
+ "192.168.2.1",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-4', 'route', 'del', '192.168.2.1/32',
- 'dev', 'eth0'], capture=True),
+ ["ip", "-4", "route", "del", "192.168.2.1/32", "dev", "eth0"],
+ capture=True,
+ ),
mock.call(
- ['ip', '-family', 'inet', 'link', 'set', 'dev',
- 'eth0', 'down'], capture=True),
+ [
+ "ip",
+ "-family",
+ "inet",
+ "link",
+ "set",
+ "dev",
+ "eth0",
+ "down",
+ ],
+ capture=True,
+ ),
mock.call(
- ['ip', '-family', 'inet', 'addr', 'del',
- '192.168.2.2/32', 'dev', 'eth0'], capture=True)
+ [
+ "ip",
+ "-family",
+ "inet",
+ "addr",
+ "del",
+ "192.168.2.2/32",
+ "dev",
+ "eth0",
+ ],
+ capture=True,
+ ),
]
with net.EphemeralIPv4Network(**params):
self.assertEqual(expected_setup_calls, m_subp.call_args_list)
@@ -754,7 +1036,8 @@ class TestEphemeralIPV4Network(CiTestCase):
class TestApplyNetworkCfgNames(CiTestCase):
- V1_CONFIG = textwrap.dedent("""\
+ V1_CONFIG = textwrap.dedent(
+ """\
version: 1
config:
- type: physical
@@ -765,8 +1048,10 @@ class TestApplyNetworkCfgNames(CiTestCase):
address: 10.0.2.15
netmask: 255.255.255.0
gateway: 10.0.2.2
- """)
- V2_CONFIG = textwrap.dedent("""\
+ """
+ )
+ V2_CONFIG = textwrap.dedent(
+ """\
version: 2
ethernets:
interface0:
@@ -776,9 +1061,11 @@ class TestApplyNetworkCfgNames(CiTestCase):
- 10.0.2.15/24
gateway4: 10.0.2.2
set-name: interface0
- """)
+ """
+ )
- V2_CONFIG_NO_SETNAME = textwrap.dedent("""\
+ V2_CONFIG_NO_SETNAME = textwrap.dedent(
+ """\
version: 2
ethernets:
interface0:
@@ -787,9 +1074,11 @@ class TestApplyNetworkCfgNames(CiTestCase):
addresses:
- 10.0.2.15/24
gateway4: 10.0.2.2
- """)
+ """
+ )
- V2_CONFIG_NO_MAC = textwrap.dedent("""\
+ V2_CONFIG_NO_MAC = textwrap.dedent(
+ """\
version: 2
ethernets:
interface0:
@@ -799,40 +1088,43 @@ class TestApplyNetworkCfgNames(CiTestCase):
- 10.0.2.15/24
gateway4: 10.0.2.2
set-name: interface0
- """)
+ """
+ )
- @mock.patch('cloudinit.net.device_devid')
- @mock.patch('cloudinit.net.device_driver')
- @mock.patch('cloudinit.net._rename_interfaces')
- def test_apply_v1_renames(self, m_rename_interfaces, m_device_driver,
- m_device_devid):
- m_device_driver.return_value = 'virtio_net'
- m_device_devid.return_value = '0x15d8'
+ @mock.patch("cloudinit.net.device_devid")
+ @mock.patch("cloudinit.net.device_driver")
+ @mock.patch("cloudinit.net._rename_interfaces")
+ def test_apply_v1_renames(
+ self, m_rename_interfaces, m_device_driver, m_device_devid
+ ):
+ m_device_driver.return_value = "virtio_net"
+ m_device_devid.return_value = "0x15d8"
net.apply_network_config_names(yaml.load(self.V1_CONFIG))
- call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
+ call = ["52:54:00:12:34:00", "interface0", "virtio_net", "0x15d8"]
m_rename_interfaces.assert_called_with([call])
- @mock.patch('cloudinit.net.device_devid')
- @mock.patch('cloudinit.net.device_driver')
- @mock.patch('cloudinit.net._rename_interfaces')
- def test_apply_v2_renames(self, m_rename_interfaces, m_device_driver,
- m_device_devid):
- m_device_driver.return_value = 'virtio_net'
- m_device_devid.return_value = '0x15d8'
+ @mock.patch("cloudinit.net.device_devid")
+ @mock.patch("cloudinit.net.device_driver")
+ @mock.patch("cloudinit.net._rename_interfaces")
+ def test_apply_v2_renames(
+ self, m_rename_interfaces, m_device_driver, m_device_devid
+ ):
+ m_device_driver.return_value = "virtio_net"
+ m_device_devid.return_value = "0x15d8"
net.apply_network_config_names(yaml.load(self.V2_CONFIG))
- call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
+ call = ["52:54:00:12:34:00", "interface0", "virtio_net", "0x15d8"]
m_rename_interfaces.assert_called_with([call])
- @mock.patch('cloudinit.net._rename_interfaces')
+ @mock.patch("cloudinit.net._rename_interfaces")
def test_apply_v2_renames_skips_without_setname(self, m_rename_interfaces):
net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_SETNAME))
m_rename_interfaces.assert_called_with([])
- @mock.patch('cloudinit.net._rename_interfaces')
+ @mock.patch("cloudinit.net._rename_interfaces")
def test_apply_v2_renames_skips_without_mac(self, m_rename_interfaces):
net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_MAC))
m_rename_interfaces.assert_called_with([])
@@ -843,184 +1135,191 @@ class TestApplyNetworkCfgNames(CiTestCase):
class TestHasURLConnectivity(HttprettyTestCase):
-
def setUp(self):
super(TestHasURLConnectivity, self).setUp()
- self.url = 'http://fake/'
- self.kwargs = {'allow_redirects': True, 'timeout': 5.0}
+ self.url = "http://fake/"
+ self.kwargs = {"allow_redirects": True, "timeout": 5.0}
- @mock.patch('cloudinit.net.readurl')
+ @mock.patch("cloudinit.net.readurl")
def test_url_timeout_on_connectivity_check(self, m_readurl):
"""A timeout of 5 seconds is provided when reading a url."""
self.assertTrue(
- net.has_url_connectivity({'url': self.url}),
- 'Expected True on url connect')
+ net.has_url_connectivity({"url": self.url}),
+ "Expected True on url connect",
+ )
def test_true_on_url_connectivity_success(self):
httpretty.register_uri(httpretty.GET, self.url)
self.assertTrue(
- net.has_url_connectivity({'url': self.url}),
- 'Expected True on url connect')
+ net.has_url_connectivity({"url": self.url}),
+ "Expected True on url connect",
+ )
- @mock.patch('requests.Session.request')
+ @mock.patch("requests.Session.request")
def test_true_on_url_connectivity_timeout(self, m_request):
"""A timeout raised accessing the url will return False."""
- m_request.side_effect = requests.Timeout('Fake Connection Timeout')
+ m_request.side_effect = requests.Timeout("Fake Connection Timeout")
self.assertFalse(
- net.has_url_connectivity({'url': self.url}),
- 'Expected False on url timeout')
+ net.has_url_connectivity({"url": self.url}),
+ "Expected False on url timeout",
+ )
def test_true_on_url_connectivity_failure(self):
httpretty.register_uri(httpretty.GET, self.url, body={}, status=404)
self.assertFalse(
- net.has_url_connectivity({'url': self.url}),
- 'Expected False on url fail')
+ net.has_url_connectivity({"url": self.url}),
+ "Expected False on url fail",
+ )
def _mk_v1_phys(mac, name, driver, device_id):
- v1_cfg = {'type': 'physical', 'name': name, 'mac_address': mac}
+ v1_cfg = {"type": "physical", "name": name, "mac_address": mac}
params = {}
if driver:
- params.update({'driver': driver})
+ params.update({"driver": driver})
if device_id:
- params.update({'device_id': device_id})
+ params.update({"device_id": device_id})
if params:
- v1_cfg.update({'params': params})
+ v1_cfg.update({"params": params})
return v1_cfg
def _mk_v2_phys(mac, name, driver=None, device_id=None):
- v2_cfg = {'set-name': name, 'match': {'macaddress': mac}}
+ v2_cfg = {"set-name": name, "match": {"macaddress": mac}}
if driver:
- v2_cfg['match'].update({'driver': driver})
+ v2_cfg["match"].update({"driver": driver})
if device_id:
- v2_cfg['match'].update({'device_id': device_id})
+ v2_cfg["match"].update({"device_id": device_id})
return v2_cfg
class TestExtractPhysdevs(CiTestCase):
-
def setUp(self):
super(TestExtractPhysdevs, self).setUp()
- self.add_patch('cloudinit.net.device_driver', 'm_driver')
- self.add_patch('cloudinit.net.device_devid', 'm_devid')
+ self.add_patch("cloudinit.net.device_driver", "m_driver")
+ self.add_patch("cloudinit.net.device_devid", "m_devid")
def test_extract_physdevs_looks_up_driver_v1(self):
- driver = 'virtio'
+ driver = "virtio"
self.m_driver.return_value = driver
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
+ ["aa:bb:cc:dd:ee:ff", "eth0", None, "0x1000"],
]
netcfg = {
- 'version': 1,
- 'config': [_mk_v1_phys(*args) for args in physdevs],
+ "version": 1,
+ "config": [_mk_v1_phys(*args) for args in physdevs],
}
# insert the driver value for verification
physdevs[0][2] = driver
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
- self.m_driver.assert_called_with('eth0')
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
+ self.m_driver.assert_called_with("eth0")
def test_extract_physdevs_looks_up_driver_v2(self):
- driver = 'virtio'
+ driver = "virtio"
self.m_driver.return_value = driver
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
+ ["aa:bb:cc:dd:ee:ff", "eth0", None, "0x1000"],
]
netcfg = {
- 'version': 2,
- 'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
+ "version": 2,
+ "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
}
# insert the driver value for verification
physdevs[0][2] = driver
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
- self.m_driver.assert_called_with('eth0')
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
+ self.m_driver.assert_called_with("eth0")
def test_extract_physdevs_looks_up_devid_v1(self):
- devid = '0x1000'
+ devid = "0x1000"
self.m_devid.return_value = devid
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
+ ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", None],
]
netcfg = {
- 'version': 1,
- 'config': [_mk_v1_phys(*args) for args in physdevs],
+ "version": 1,
+ "config": [_mk_v1_phys(*args) for args in physdevs],
}
# insert the driver value for verification
physdevs[0][3] = devid
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
- self.m_devid.assert_called_with('eth0')
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
+ self.m_devid.assert_called_with("eth0")
def test_extract_physdevs_looks_up_devid_v2(self):
- devid = '0x1000'
+ devid = "0x1000"
self.m_devid.return_value = devid
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
+ ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", None],
]
netcfg = {
- 'version': 2,
- 'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
+ "version": 2,
+ "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
}
# insert the driver value for verification
physdevs[0][3] = devid
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
- self.m_devid.assert_called_with('eth0')
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
+ self.m_devid.assert_called_with("eth0")
def test_get_v1_type_physical(self):
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
- ['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
- ['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
+ ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
+ ["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
+ ["09:87:65:43:21:10", "ens0p1", "mlx4_core", "0:0:1000"],
]
netcfg = {
- 'version': 1,
- 'config': [_mk_v1_phys(*args) for args in physdevs],
+ "version": 1,
+ "config": [_mk_v1_phys(*args) for args in physdevs],
}
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
def test_get_v2_type_physical(self):
physdevs = [
- ['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
- ['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
- ['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
+ ["aa:bb:cc:dd:ee:ff", "eth0", "virtio", "0x1000"],
+ ["00:11:22:33:44:55", "ens3", "e1000", "0x1643"],
+ ["09:87:65:43:21:10", "ens0p1", "mlx4_core", "0:0:1000"],
]
netcfg = {
- 'version': 2,
- 'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
+ "version": 2,
+ "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs},
}
- self.assertEqual(sorted(physdevs),
- sorted(net.extract_physdevs(netcfg)))
+ self.assertEqual(
+ sorted(physdevs), sorted(net.extract_physdevs(netcfg))
+ )
def test_get_v2_type_physical_skips_if_no_set_name(self):
netcfg = {
- 'version': 2,
- 'ethernets': {
- 'ens3': {
- 'match': {'macaddress': '00:11:22:33:44:55'},
+ "version": 2,
+ "ethernets": {
+ "ens3": {
+ "match": {"macaddress": "00:11:22:33:44:55"},
}
- }
+ },
}
self.assertEqual([], net.extract_physdevs(netcfg))
def test_runtime_error_on_unknown_netcfg_version(self):
with self.assertRaises(RuntimeError):
- net.extract_physdevs({'version': 3, 'awesome_config': []})
+ net.extract_physdevs({"version": 3, "awesome_config": []})
class TestNetFailOver(CiTestCase):
-
def setUp(self):
super(TestNetFailOver, self).setUp()
- self.add_patch('cloudinit.net.util', 'm_util')
- self.add_patch('cloudinit.net.read_sys_net', 'm_read_sys_net')
- self.add_patch('cloudinit.net.device_driver', 'm_device_driver')
+ self.add_patch("cloudinit.net.util", "m_util")
+ self.add_patch("cloudinit.net.read_sys_net", "m_read_sys_net")
+ self.add_patch("cloudinit.net.device_driver", "m_device_driver")
def test_get_dev_features(self):
devname = self.random_string()
@@ -1029,192 +1328,210 @@ class TestNetFailOver(CiTestCase):
self.assertEqual(features, net.get_dev_features(devname))
self.assertEqual(1, self.m_read_sys_net.call_count)
- self.assertEqual(mock.call(devname, 'device/features'),
- self.m_read_sys_net.call_args_list[0])
+ self.assertEqual(
+ mock.call(devname, "device/features"),
+ self.m_read_sys_net.call_args_list[0],
+ )
def test_get_dev_features_none_returns_empty_string(self):
devname = self.random_string()
- self.m_read_sys_net.side_effect = Exception('error')
- self.assertEqual('', net.get_dev_features(devname))
+ self.m_read_sys_net.side_effect = Exception("error")
+ self.assertEqual("", net.get_dev_features(devname))
self.assertEqual(1, self.m_read_sys_net.call_count)
- self.assertEqual(mock.call(devname, 'device/features'),
- self.m_read_sys_net.call_args_list[0])
+ self.assertEqual(
+ mock.call(devname, "device/features"),
+ self.m_read_sys_net.call_args_list[0],
+ )
- @mock.patch('cloudinit.net.get_dev_features')
+ @mock.patch("cloudinit.net.get_dev_features")
def test_has_netfail_standby_feature(self, m_dev_features):
devname = self.random_string()
- standby_features = ('0' * 62) + '1' + '0'
+ standby_features = ("0" * 62) + "1" + "0"
m_dev_features.return_value = standby_features
self.assertTrue(net.has_netfail_standby_feature(devname))
- @mock.patch('cloudinit.net.get_dev_features')
+ @mock.patch("cloudinit.net.get_dev_features")
def test_has_netfail_standby_feature_short_is_false(self, m_dev_features):
devname = self.random_string()
standby_features = self.random_string()
m_dev_features.return_value = standby_features
self.assertFalse(net.has_netfail_standby_feature(devname))
- @mock.patch('cloudinit.net.get_dev_features')
- def test_has_netfail_standby_feature_not_present_is_false(self,
- m_dev_features):
+ @mock.patch("cloudinit.net.get_dev_features")
+ def test_has_netfail_standby_feature_not_present_is_false(
+ self, m_dev_features
+ ):
devname = self.random_string()
- standby_features = '0' * 64
+ standby_features = "0" * 64
m_dev_features.return_value = standby_features
self.assertFalse(net.has_netfail_standby_feature(devname))
- @mock.patch('cloudinit.net.get_dev_features')
- def test_has_netfail_standby_feature_no_features_is_false(self,
- m_dev_features):
+ @mock.patch("cloudinit.net.get_dev_features")
+ def test_has_netfail_standby_feature_no_features_is_false(
+ self, m_dev_features
+ ):
devname = self.random_string()
standby_features = None
m_dev_features.return_value = standby_features
self.assertFalse(net.has_netfail_standby_feature(devname))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_master(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = False # no master sysfs attr
m_standby.return_value = True # has standby feature flag
self.assertTrue(net.is_netfail_master(devname, driver))
- @mock.patch('cloudinit.net.sys_dev_path')
+ @mock.patch("cloudinit.net.sys_dev_path")
def test_is_netfail_master_checks_master_attr(self, m_sysdev):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_sysdev.return_value = self.random_string()
self.assertFalse(net.is_netfail_master(devname, driver))
self.assertEqual(1, m_sysdev.call_count)
- self.assertEqual(mock.call(devname, path='master'),
- m_sysdev.call_args_list[0])
+ self.assertEqual(
+ mock.call(devname, path="master"), m_sysdev.call_args_list[0]
+ )
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_master_wrong_driver(self, m_exists, m_standby):
devname = self.random_string()
driver = self.random_string()
self.assertFalse(net.is_netfail_master(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_master_has_master_attr(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = True # has master sysfs attr
self.assertFalse(net.is_netfail_master(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_master_no_standby_feat(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = False # no master sysfs attr
m_standby.return_value = False # no standby feature flag
self.assertFalse(net.is_netfail_master(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
- @mock.patch('cloudinit.net.sys_dev_path')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
+ @mock.patch("cloudinit.net.sys_dev_path")
def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby):
devname = self.random_string()
driver = self.random_string() # device not virtio_net
master_devname = self.random_string()
- m_sysdev.return_value = "%s/%s" % (self.random_string(),
- master_devname)
+ m_sysdev.return_value = "%s/%s" % (
+ self.random_string(),
+ master_devname,
+ )
m_exists.return_value = True # has master sysfs attr
- self.m_device_driver.return_value = 'virtio_net' # master virtio_net
+ self.m_device_driver.return_value = "virtio_net" # master virtio_net
m_standby.return_value = True # has standby feature flag
self.assertTrue(net.is_netfail_primary(devname, driver))
self.assertEqual(1, self.m_device_driver.call_count)
- self.assertEqual(mock.call(master_devname),
- self.m_device_driver.call_args_list[0])
+ self.assertEqual(
+ mock.call(master_devname), self.m_device_driver.call_args_list[0]
+ )
self.assertEqual(1, m_standby.call_count)
- self.assertEqual(mock.call(master_devname),
- m_standby.call_args_list[0])
-
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
- @mock.patch('cloudinit.net.sys_dev_path')
- def test_is_netfail_primary_wrong_driver(self, m_sysdev, m_exists,
- m_standby):
+ self.assertEqual(
+ mock.call(master_devname), m_standby.call_args_list[0]
+ )
+
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
+ @mock.patch("cloudinit.net.sys_dev_path")
+ def test_is_netfail_primary_wrong_driver(
+ self, m_sysdev, m_exists, m_standby
+ ):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
self.assertFalse(net.is_netfail_primary(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
- @mock.patch('cloudinit.net.sys_dev_path')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
+ @mock.patch("cloudinit.net.sys_dev_path")
def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby):
devname = self.random_string()
driver = self.random_string() # device not virtio_net
m_exists.return_value = False # no master sysfs attr
self.assertFalse(net.is_netfail_primary(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
- @mock.patch('cloudinit.net.sys_dev_path')
- def test_is_netfail_primary_bad_master(self, m_sysdev, m_exists,
- m_standby):
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
+ @mock.patch("cloudinit.net.sys_dev_path")
+ def test_is_netfail_primary_bad_master(
+ self, m_sysdev, m_exists, m_standby
+ ):
devname = self.random_string()
driver = self.random_string() # device not virtio_net
master_devname = self.random_string()
- m_sysdev.return_value = "%s/%s" % (self.random_string(),
- master_devname)
+ m_sysdev.return_value = "%s/%s" % (
+ self.random_string(),
+ master_devname,
+ )
m_exists.return_value = True # has master sysfs attr
- self.m_device_driver.return_value = 'XXXX' # master not virtio_net
+ self.m_device_driver.return_value = "XXXX" # master not virtio_net
self.assertFalse(net.is_netfail_primary(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
- @mock.patch('cloudinit.net.sys_dev_path')
- def test_is_netfail_primary_no_standby(self, m_sysdev, m_exists,
- m_standby):
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
+ @mock.patch("cloudinit.net.sys_dev_path")
+ def test_is_netfail_primary_no_standby(
+ self, m_sysdev, m_exists, m_standby
+ ):
devname = self.random_string()
driver = self.random_string() # device not virtio_net
master_devname = self.random_string()
- m_sysdev.return_value = "%s/%s" % (self.random_string(),
- master_devname)
+ m_sysdev.return_value = "%s/%s" % (
+ self.random_string(),
+ master_devname,
+ )
m_exists.return_value = True # has master sysfs attr
- self.m_device_driver.return_value = 'virtio_net' # master virtio_net
+ self.m_device_driver.return_value = "virtio_net" # master virtio_net
m_standby.return_value = False # master has no standby feature flag
self.assertFalse(net.is_netfail_primary(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_standby(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = True # has master sysfs attr
m_standby.return_value = True # has standby feature flag
self.assertTrue(net.is_netfail_standby(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_standby_wrong_driver(self, m_exists, m_standby):
devname = self.random_string()
driver = self.random_string()
self.assertFalse(net.is_netfail_standby(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_standby_no_master(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = False # has master sysfs attr
self.assertFalse(net.is_netfail_standby(devname, driver))
- @mock.patch('cloudinit.net.has_netfail_standby_feature')
- @mock.patch('cloudinit.net.os.path.exists')
+ @mock.patch("cloudinit.net.has_netfail_standby_feature")
+ @mock.patch("cloudinit.net.os.path.exists")
def test_is_netfail_standby_no_standby_feature(self, m_exists, m_standby):
devname = self.random_string()
- driver = 'virtio_net'
+ driver = "virtio_net"
m_exists.return_value = True # has master sysfs attr
m_standby.return_value = False # has standby feature flag
self.assertFalse(net.is_netfail_standby(devname, driver))
- @mock.patch('cloudinit.net.is_netfail_standby')
- @mock.patch('cloudinit.net.is_netfail_primary')
+ @mock.patch("cloudinit.net.is_netfail_standby")
+ @mock.patch("cloudinit.net.is_netfail_primary")
def test_is_netfailover_primary(self, m_primary, m_standby):
devname = self.random_string()
driver = self.random_string()
@@ -1222,8 +1539,8 @@ class TestNetFailOver(CiTestCase):
m_standby.return_value = False
self.assertTrue(net.is_netfailover(devname, driver))
- @mock.patch('cloudinit.net.is_netfail_standby')
- @mock.patch('cloudinit.net.is_netfail_primary')
+ @mock.patch("cloudinit.net.is_netfail_standby")
+ @mock.patch("cloudinit.net.is_netfail_primary")
def test_is_netfailover_standby(self, m_primary, m_standby):
devname = self.random_string()
driver = self.random_string()
@@ -1231,8 +1548,8 @@ class TestNetFailOver(CiTestCase):
m_standby.return_value = True
self.assertTrue(net.is_netfailover(devname, driver))
- @mock.patch('cloudinit.net.is_netfail_standby')
- @mock.patch('cloudinit.net.is_netfail_primary')
+ @mock.patch("cloudinit.net.is_netfail_standby")
+ @mock.patch("cloudinit.net.is_netfail_primary")
def test_is_netfailover_returns_false(self, m_primary, m_standby):
devname = self.random_string()
driver = self.random_string()
@@ -1274,6 +1591,7 @@ class TestGetOVSInternalInterfaces:
Uses the ``clear_lru_cache`` local autouse fixture to allow us to test
despite the ``lru_cache`` decorator on the unit under test.
"""
+
@pytest.fixture(autouse=True)
def clear_lru_cache(self):
net.get_ovs_internal_interfaces.cache_clear()
@@ -1364,14 +1682,19 @@ class TestIsIpAddress:
the ipaddress module correctly.
"""
- @pytest.mark.parametrize('ip_address_side_effect,expected_return', (
- (ValueError, False),
- (lambda _: ipaddress.IPv4Address('192.168.0.1'), True),
- (lambda _: ipaddress.IPv6Address('2001:db8::'), True),
- ))
+ @pytest.mark.parametrize(
+ "ip_address_side_effect,expected_return",
+ (
+ (ValueError, False),
+ (lambda _: ipaddress.IPv4Address("192.168.0.1"), True),
+ (lambda _: ipaddress.IPv6Address("2001:db8::"), True),
+ ),
+ )
def test_is_ip_address(self, ip_address_side_effect, expected_return):
- with mock.patch('cloudinit.net.ipaddress.ip_address',
- side_effect=ip_address_side_effect) as m_ip_address:
+ with mock.patch(
+ "cloudinit.net.ipaddress.ip_address",
+ side_effect=ip_address_side_effect,
+ ) as m_ip_address:
ret = net.is_ip_address(mock.sentinel.ip_address_in)
assert expected_return == ret
expected_call = mock.call(mock.sentinel.ip_address_in)
@@ -1386,13 +1709,20 @@ class TestIsIpv4Address:
the ipaddress module correctly.
"""
- @pytest.mark.parametrize('ipv4address_mock,expected_return', (
- (mock.Mock(side_effect=ValueError), False),
- (mock.Mock(return_value=ipaddress.IPv4Address('192.168.0.1')), True),
- ))
+ @pytest.mark.parametrize(
+ "ipv4address_mock,expected_return",
+ (
+ (mock.Mock(side_effect=ValueError), False),
+ (
+ mock.Mock(return_value=ipaddress.IPv4Address("192.168.0.1")),
+ True,
+ ),
+ ),
+ )
def test_is_ip_address(self, ipv4address_mock, expected_return):
- with mock.patch('cloudinit.net.ipaddress.IPv4Address',
- ipv4address_mock) as m_ipv4address:
+ with mock.patch(
+ "cloudinit.net.ipaddress.IPv4Address", ipv4address_mock
+ ) as m_ipv4address:
ret = net.is_ipv4_address(mock.sentinel.ip_address_in)
assert expected_return == ret
expected_call = mock.call(mock.sentinel.ip_address_in)