|
Packit Service |
a04d08 |
# This file is part of cloud-init. See LICENSE file for license information.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
import copy
|
|
Packit Service |
a04d08 |
import errno
|
|
Packit Service |
9bfd13 |
import ipaddress
|
|
Packit Service |
b1601c |
import os
|
|
Packit Service |
b1601c |
import textwrap
|
|
Packit Service |
9bfd13 |
from unittest import mock
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
import httpretty
|
|
Packit Service |
9bfd13 |
import pytest
|
|
Packit Service |
9bfd13 |
import requests
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
import cloudinit.net as net
|
|
Packit Service |
b1601c |
from cloudinit import safeyaml as yaml
|
|
Packit Service |
9bfd13 |
from cloudinit.tests.helpers import CiTestCase, HttprettyTestCase
|
|
Packit Service |
9bfd13 |
from cloudinit.subp import ProcessExecutionError
|
|
Packit Service |
9bfd13 |
from cloudinit.util import ensure_file, write_file
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestSysDevPath(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sys_dev_path(self):
|
|
Packit Service |
a04d08 |
"""sys_dev_path returns a path under SYS_CLASS_NET for a device."""
|
|
Packit Service |
a04d08 |
dev = 'something'
|
|
Packit Service |
a04d08 |
path = 'attribute'
|
|
Packit Service |
a04d08 |
expected = net.SYS_CLASS_NET + dev + '/' + path
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.sys_dev_path(dev, path))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sys_dev_path_without_path(self):
|
|
Packit Service |
a04d08 |
"""When path param isn't provided it defaults to empty string."""
|
|
Packit Service |
a04d08 |
dev = 'something'
|
|
Packit Service |
a04d08 |
expected = net.SYS_CLASS_NET + dev + '/'
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.sys_dev_path(dev))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestReadSysNet(CiTestCase):
|
|
Packit Service |
a04d08 |
with_logs = True
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestReadSysNet, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_strips_contents_of_sys_path(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net strips whitespace from the contents of a sys file."""
|
|
Packit Service |
a04d08 |
content = 'some stuff with trailing whitespace\t\r\n'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
|
|
Packit Service |
a04d08 |
self.assertEqual(content.strip(), net.read_sys_net('dev', 'attr'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_reraises_oserror(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net raises OSError/IOError when file doesn't exist."""
|
|
Packit Service |
a04d08 |
# Non-specific Exception because versions of python OSError vs IOError.
|
|
Packit Service |
a04d08 |
with self.assertRaises(Exception) as context_manager: # noqa: H202
|
|
Packit Service |
a04d08 |
net.read_sys_net('dev', 'attr')
|
|
Packit Service |
a04d08 |
error = context_manager.exception
|
|
Packit Service |
a04d08 |
self.assertIn('No such file or directory', str(error))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_handles_error_with_on_enoent(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net handles OSError/IOError with on_enoent if provided."""
|
|
Packit Service |
a04d08 |
handled_errors = []
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def on_enoent(e):
|
|
Packit Service |
a04d08 |
handled_errors.append(e)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
net.read_sys_net('dev', 'attr', on_enoent=on_enoent)
|
|
Packit Service |
a04d08 |
error = handled_errors[0]
|
|
Packit Service |
a04d08 |
self.assertIsInstance(error, Exception)
|
|
Packit Service |
a04d08 |
self.assertIn('No such file or directory', str(error))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_translates_content(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net translates content when translate dict is provided."""
|
|
Packit Service |
a04d08 |
content = "you're welcome\n"
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
|
|
Packit Service |
a04d08 |
translate = {"you're welcome": 'de nada'}
|
|
Packit Service |
a04d08 |
self.assertEqual(
|
|
Packit Service |
a04d08 |
'de nada',
|
|
Packit Service |
a04d08 |
net.read_sys_net('dev', 'attr', translate=translate))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_errors_on_translation_failures(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net raises a KeyError and logs details on failure."""
|
|
Packit Service |
a04d08 |
content = "you're welcome\n"
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
|
|
Packit Service |
a04d08 |
with self.assertRaises(KeyError) as context_manager:
|
|
Packit Service |
a04d08 |
net.read_sys_net('dev', 'attr', translate={})
|
|
Packit Service |
a04d08 |
error = context_manager.exception
|
|
Packit Service |
a04d08 |
self.assertEqual('"you\'re welcome"', str(error))
|
|
Packit Service |
a04d08 |
self.assertIn(
|
|
Packit Service |
a04d08 |
"Found unexpected (not translatable) value 'you're welcome' in "
|
|
Packit Service |
a04d08 |
"'{0}dev/attr".format(self.sysdir),
|
|
Packit Service |
a04d08 |
self.logs.getvalue())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_handles_handles_with_onkeyerror(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net handles translation errors calling on_keyerror."""
|
|
Packit Service |
a04d08 |
content = "you're welcome\n"
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
|
|
Packit Service |
a04d08 |
handled_errors = []
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def on_keyerror(e):
|
|
Packit Service |
a04d08 |
handled_errors.append(e)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
net.read_sys_net('dev', 'attr', translate={}, on_keyerror=on_keyerror)
|
|
Packit Service |
a04d08 |
error = handled_errors[0]
|
|
Packit Service |
a04d08 |
self.assertIsInstance(error, KeyError)
|
|
Packit Service |
a04d08 |
self.assertEqual('"you\'re welcome"', str(error))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_safe_false_on_translate_failure(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net_safe returns False on translation failures."""
|
|
Packit Service |
a04d08 |
content = "you're welcome\n"
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), content)
|
|
Packit Service |
a04d08 |
self.assertFalse(net.read_sys_net_safe('dev', 'attr', translate={}))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_safe_returns_false_on_noent_failure(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net_safe returns False on file not found failures."""
|
|
Packit Service |
a04d08 |
self.assertFalse(net.read_sys_net_safe('dev', 'attr'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_int_returns_none_on_error(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net_safe returns None on failures."""
|
|
Packit Service |
a04d08 |
self.assertFalse(net.read_sys_net_int('dev', 'attr'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_int_returns_none_on_valueerror(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net_safe returns None when content is not an int."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), 'NOTINT\n')
|
|
Packit Service |
a04d08 |
self.assertFalse(net.read_sys_net_int('dev', 'attr'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_read_sys_net_int_returns_integer_from_content(self):
|
|
Packit Service |
a04d08 |
"""read_sys_net_safe returns None on failures."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'dev', 'attr'), '1\n')
|
|
Packit Service |
a04d08 |
self.assertEqual(1, net.read_sys_net_int('dev', 'attr'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_is_up_true(self):
|
|
Packit Service |
a04d08 |
"""is_up is True if sys/net/devname/operstate is 'up' or 'unknown'."""
|
|
Packit Service |
a04d08 |
for state in ['up', 'unknown']:
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_up('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_is_up_false(self):
|
|
Packit Service |
a04d08 |
"""is_up is False if sys/net/devname/operstate is 'down' or invalid."""
|
|
Packit Service |
a04d08 |
for state in ['down', 'incomprehensible']:
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_up('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_is_bridge(self):
|
|
Packit Service |
a04d08 |
"""is_bridge is True when /sys/net/devname/bridge exists."""
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_bridge('eth0'))
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_bridge('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_is_bond(self):
|
|
Packit Service |
a04d08 |
"""is_bond is True when /sys/net/devname/bonding exists."""
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_bond('eth0'))
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_bond('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_master(self):
|
|
Packit Service |
a04d08 |
"""get_master returns the path when /sys/net/devname/master exists."""
|
|
Packit Service |
a04d08 |
self.assertIsNone(net.get_master('enP1s1'))
|
|
Packit Service |
a04d08 |
master_path = os.path.join(self.sysdir, 'enP1s1', 'master')
|
|
Packit Service |
a04d08 |
ensure_file(master_path)
|
|
Packit Service |
a04d08 |
self.assertEqual(master_path, net.get_master('enP1s1'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_master_is_bridge_or_bond(self):
|
|
Packit Service |
a04d08 |
bridge_mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
bond_mac = 'cc:bb:aa:cc:bb:aa'
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# No master => False
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self.assertFalse(net.master_is_bridge_or_bond('eth1'))
|
|
Packit Service |
a04d08 |
self.assertFalse(net.master_is_bridge_or_bond('eth2'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# masters without bridge/bonding => False
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
|
|
Packit Service |
a04d08 |
os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self.assertFalse(net.master_is_bridge_or_bond('eth1'))
|
|
Packit Service |
a04d08 |
self.assertFalse(net.master_is_bridge_or_bond('eth2'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
# masters with bridge/bonding => True
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self.assertTrue(net.master_is_bridge_or_bond('eth1'))
|
|
Packit Service |
a04d08 |
self.assertTrue(net.master_is_bridge_or_bond('eth2'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_is_vlan(self):
|
|
Packit Service |
a04d08 |
"""is_vlan is True when /sys/net/devname/uevent has DEVTYPE=vlan."""
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'uevent'))
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_vlan('eth0'))
|
|
Packit Service |
a04d08 |
content = 'junk\nDEVTYPE=vlan\njunk\n'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'uevent'), content)
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_vlan('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestGenerateFallbackConfig(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestGenerateFallbackConfig, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
|
|
Packit Service |
a04d08 |
return_value=False)
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.is_netfailover', 'm_netfail',
|
|
Packit Service |
a04d08 |
return_value=False)
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.is_netfail_master', 'm_netfail_master',
|
|
Packit Service |
a04d08 |
return_value=False)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_finds_connected_eth_with_mac(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config finds any connected device with a mac."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
|
|
Packit Service |
a04d08 |
expected = {
|
|
Packit Service |
a04d08 |
'ethernets': {'eth1': {'match': {'macaddress': mac},
|
|
Packit Service |
a04d08 |
'dhcp4': True, 'set-name': 'eth1'}},
|
|
Packit Service |
a04d08 |
'version': 2}
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_finds_dormant_eth_with_mac(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config finds any dormant device with a mac."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'dormant'), '1')
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
|
|
Packit Service |
a04d08 |
expected = {
|
|
Packit Service |
a04d08 |
'ethernets': {'eth0': {'match': {'macaddress': mac}, 'dhcp4': True,
|
|
Packit Service |
a04d08 |
'set-name': 'eth0'}},
|
|
Packit Service |
a04d08 |
'version': 2}
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_finds_eth_by_operstate(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config finds any dormant device with a mac."""
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
|
|
Packit Service |
a04d08 |
expected = {
|
|
Packit Service |
a04d08 |
'ethernets': {
|
|
Packit Service |
a04d08 |
'eth0': {'dhcp4': True, 'match': {'macaddress': mac},
|
|
Packit Service |
a04d08 |
'set-name': 'eth0'}},
|
|
Packit Service |
a04d08 |
'version': 2}
|
|
Packit Service |
a04d08 |
valid_operstates = ['dormant', 'down', 'lowerlayerdown', 'unknown']
|
|
Packit Service |
a04d08 |
for state in valid_operstates:
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), state)
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'noworky')
|
|
Packit Service |
a04d08 |
self.assertIsNone(net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_config_skips_veth(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config will skip any veth interfaces."""
|
|
Packit Service |
a04d08 |
# A connected veth which gets ignored
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'veth0', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
self.assertIsNone(net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_config_skips_bridges(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config will skip any bridges interfaces."""
|
|
Packit Service |
a04d08 |
# A connected veth which gets ignored
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'bridge'))
|
|
Packit Service |
a04d08 |
self.assertIsNone(net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_config_skips_bonds(self):
|
|
Packit Service |
a04d08 |
"""generate_fallback_config will skip any bonded interfaces."""
|
|
Packit Service |
a04d08 |
# A connected veth which gets ignored
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'address'), mac)
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
|
|
Packit Service |
a04d08 |
self.assertIsNone(net.generate_fallback_config())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_config_skips_netfail_devs(self):
|
|
Packit Service |
a04d08 |
"""gen_fallback_config ignores netfail primary,sby no mac on master."""
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc' # netfailover devs share the same mac
|
|
Packit Service |
a04d08 |
for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, iface, 'carrier'), '1')
|
|
Packit Service |
a04d08 |
write_file(
|
|
Packit Service |
a04d08 |
os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(
|
|
Packit Service |
a04d08 |
os.path.join(self.sysdir, iface, 'address'), mac)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def is_netfail(iface, _driver=None):
|
|
Packit Service |
a04d08 |
# ens3 is the master
|
|
Packit Service |
a04d08 |
if iface == 'ens3':
|
|
Packit Service |
a04d08 |
return False
|
|
Packit Service |
a04d08 |
return True
|
|
Packit Service |
a04d08 |
self.m_netfail.side_effect = is_netfail
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def is_netfail_master(iface, _driver=None):
|
|
Packit Service |
a04d08 |
# ens3 is the master
|
|
Packit Service |
a04d08 |
if iface == 'ens3':
|
|
Packit Service |
a04d08 |
return True
|
|
Packit Service |
a04d08 |
return False
|
|
Packit Service |
a04d08 |
self.m_netfail_master.side_effect = is_netfail_master
|
|
Packit Service |
a04d08 |
expected = {
|
|
Packit Service |
a04d08 |
'ethernets': {
|
|
Packit Service |
a04d08 |
'ens3': {'dhcp4': True, 'match': {'name': 'ens3'},
|
|
Packit Service |
a04d08 |
'set-name': 'ens3'}},
|
|
Packit Service |
a04d08 |
'version': 2}
|
|
Packit Service |
a04d08 |
result = net.generate_fallback_config()
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, result)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestNetFindFallBackNic(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestNetFindFallBackNic, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
|
|
Packit Service |
a04d08 |
return_value=False)
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_generate_fallback_finds_first_connected_eth_with_mac(self):
|
|
Packit Service |
a04d08 |
"""find_fallback_nic finds any connected device with a mac."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'carrier'), '1')
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
|
|
Packit Service |
a04d08 |
self.assertEqual('eth1', net.find_fallback_nic())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestGetDeviceList(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestGetDeviceList, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_devicelist_raise_oserror(self):
|
|
Packit Service |
a04d08 |
"""get_devicelist raise any non-ENOENT OSerror."""
|
|
Packit Service |
a04d08 |
error = OSError('Can not do it')
|
|
Packit Service |
a04d08 |
error.errno = errno.EPERM # Set non-ENOENT
|
|
Packit Service |
a04d08 |
self.m_sys_path.side_effect = error
|
|
Packit Service |
a04d08 |
with self.assertRaises(OSError) as context_manager:
|
|
Packit Service |
a04d08 |
net.get_devicelist()
|
|
Packit Service |
a04d08 |
exception = context_manager.exception
|
|
Packit Service |
a04d08 |
self.assertEqual('Can not do it', str(exception))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_devicelist_empty_without_sys_net(self):
|
|
Packit Service |
a04d08 |
"""get_devicelist returns empty list when missing SYS_CLASS_NET."""
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = 'idontexist'
|
|
Packit Service |
a04d08 |
self.assertEqual([], net.get_devicelist())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_devicelist_empty_with_no_devices_in_sys_net(self):
|
|
Packit Service |
a04d08 |
"""get_devicelist returns empty directoty listing for SYS_CLASS_NET."""
|
|
Packit Service |
a04d08 |
self.assertEqual([], net.get_devicelist())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_devicelist_lists_any_subdirectories_in_sys_net(self):
|
|
Packit Service |
a04d08 |
"""get_devicelist returns a directory listing for SYS_CLASS_NET."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth0', 'operstate'), 'up')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'operstate'), 'up')
|
|
Packit Service |
9bfd13 |
self.assertCountEqual(['eth0', 'eth1'], net.get_devicelist())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestGetInterfaceMAC(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestGetInterfaceMAC, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interface_mac_false_with_no_mac(self):
|
|
Packit Service |
a04d08 |
"""get_device_list returns False when no mac is reported."""
|
|
Packit Service |
a04d08 |
ensure_file(os.path.join(self.sysdir, 'eth0', 'bonding'))
|
|
Packit Service |
a04d08 |
mac_path = os.path.join(self.sysdir, 'eth0', 'address')
|
|
Packit Service |
a04d08 |
self.assertFalse(os.path.exists(mac_path))
|
|
Packit Service |
a04d08 |
self.assertFalse(net.get_interface_mac('eth0'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interface_mac(self):
|
|
Packit Service |
a04d08 |
"""get_interfaces returns the mac from SYS_CLASS_NET/dev/address."""
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac)
|
|
Packit Service |
a04d08 |
self.assertEqual(mac, net.get_interface_mac('eth1'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interface_mac_grabs_bonding_address(self):
|
|
Packit Service |
a04d08 |
"""get_interfaces returns the source device mac for bonded devices."""
|
|
Packit Service |
a04d08 |
source_dev_mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
bonded_mac = 'dd:ee:ff:dd:ee:ff'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), bonded_mac)
|
|
Packit Service |
a04d08 |
write_file(
|
|
Packit Service |
a04d08 |
os.path.join(self.sysdir, 'eth1', 'bonding_slave', 'perm_hwaddr'),
|
|
Packit Service |
a04d08 |
source_dev_mac)
|
|
Packit Service |
a04d08 |
self.assertEqual(source_dev_mac, net.get_interface_mac('eth1'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interfaces_empty_list_without_sys_net(self):
|
|
Packit Service |
a04d08 |
"""get_interfaces returns an empty list when missing SYS_CLASS_NET."""
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = 'idontexist'
|
|
Packit Service |
a04d08 |
self.assertEqual([], net.get_interfaces())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interfaces_by_mac_skips_empty_mac(self):
|
|
Packit Service |
a04d08 |
"""Ignore 00:00:00:00:00:00 addresses from get_interfaces_by_mac."""
|
|
Packit Service |
a04d08 |
empty_mac = '00:00:00:00:00:00'
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), empty_mac)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
|
|
Packit Service |
a04d08 |
expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.get_interfaces())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interfaces_by_mac_skips_missing_mac(self):
|
|
Packit Service |
a04d08 |
"""Ignore interfaces without an address from get_interfaces_by_mac."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
address_path = os.path.join(self.sysdir, 'eth1', 'address')
|
|
Packit Service |
a04d08 |
self.assertFalse(os.path.exists(address_path))
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac)
|
|
Packit Service |
a04d08 |
expected = [('eth2', 'aa:bb:cc:aa:bb:cc', None, None)]
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.get_interfaces())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interfaces_by_mac_skips_master_devs(self):
|
|
Packit Service |
a04d08 |
"""Ignore interfaces with a master device which would have dup mac."""
|
|
Packit Service |
a04d08 |
mac1 = mac2 = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), mac1)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'master'), "blah")
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'address'), mac2)
|
|
Packit Service |
a04d08 |
expected = [('eth2', mac2, None, None)]
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.get_interfaces())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfailover')
|
|
Packit Service |
a04d08 |
def test_get_interfaces_by_mac_skips_netfailvoer(self, m_netfail):
|
|
Packit Service |
a04d08 |
"""Ignore interfaces if netfailover primary or standby."""
|
|
Packit Service |
a04d08 |
mac = 'aa:bb:cc:aa:bb:cc' # netfailover devs share the same mac
|
|
Packit Service |
a04d08 |
for iface in ['ens3', 'ens3sby', 'enP0s1f3']:
|
|
Packit Service |
a04d08 |
write_file(
|
|
Packit Service |
a04d08 |
os.path.join(self.sysdir, iface, 'addr_assign_type'), '0')
|
|
Packit Service |
a04d08 |
write_file(
|
|
Packit Service |
a04d08 |
os.path.join(self.sysdir, iface, 'address'), mac)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def is_netfail(iface, _driver=None):
|
|
Packit Service |
a04d08 |
# ens3 is the master
|
|
Packit Service |
a04d08 |
if iface == 'ens3':
|
|
Packit Service |
a04d08 |
return False
|
|
Packit Service |
a04d08 |
else:
|
|
Packit Service |
a04d08 |
return True
|
|
Packit Service |
a04d08 |
m_netfail.side_effect = is_netfail
|
|
Packit Service |
a04d08 |
expected = [('ens3', mac, None, None)]
|
|
Packit Service |
a04d08 |
self.assertEqual(expected, net.get_interfaces())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds(
|
|
Packit Service |
a04d08 |
self
|
|
Packit Service |
a04d08 |
):
|
|
Packit Service |
a04d08 |
bridge_mac = 'aa:bb:cc:aa:bb:cc'
|
|
Packit Service |
a04d08 |
bond_mac = 'cc:bb:aa:cc:bb:aa'
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'br0', 'address'), bridge_mac)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'br0', 'bridge'), '')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'bond0', 'address'), bond_mac)
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'bond0', 'bonding'), '')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'address'), bridge_mac)
|
|
Packit Service |
a04d08 |
os.symlink('../br0', os.path.join(self.sysdir, 'eth1', 'master'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth2', 'address'), bond_mac)
|
|
Packit Service |
a04d08 |
os.symlink('../bond0', os.path.join(self.sysdir, 'eth2', 'master'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
interface_names = [interface[0] for interface in net.get_interfaces()]
|
|
Packit Service |
a04d08 |
self.assertEqual(['eth1', 'eth2'], sorted(interface_names))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestInterfaceHasOwnMAC(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestInterfaceHasOwnMAC, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_interface_has_own_mac_false_when_stolen(self):
|
|
Packit Service |
a04d08 |
"""Return False from interface_has_own_mac when address is stolen."""
|
|
Packit Service |
a04d08 |
write_file(os.path.join(self.sysdir, 'eth1', 'addr_assign_type'), '2')
|
|
Packit Service |
a04d08 |
self.assertFalse(net.interface_has_own_mac('eth1'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_interface_has_own_mac_true_when_not_stolen(self):
|
|
Packit Service |
a04d08 |
"""Return False from interface_has_own_mac when mac isn't stolen."""
|
|
Packit Service |
a04d08 |
valid_assign_types = ['0', '1', '3']
|
|
Packit Service |
a04d08 |
assign_path = os.path.join(self.sysdir, 'eth1', 'addr_assign_type')
|
|
Packit Service |
a04d08 |
for _type in valid_assign_types:
|
|
Packit Service |
a04d08 |
write_file(assign_path, _type)
|
|
Packit Service |
a04d08 |
self.assertTrue(net.interface_has_own_mac('eth1'))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_interface_has_own_mac_strict_errors_on_absent_assign_type(self):
|
|
Packit Service |
a04d08 |
"""When addr_assign_type is absent, interface_has_own_mac errors."""
|
|
Packit Service |
a04d08 |
with self.assertRaises(ValueError):
|
|
Packit Service |
a04d08 |
net.interface_has_own_mac('eth1', strict=True)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
9bfd13 |
@mock.patch('cloudinit.net.subp.subp')
|
|
Packit Service |
a04d08 |
class TestEphemeralIPV4Network(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
with_logs = True
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestEphemeralIPV4Network, self).setUp()
|
|
Packit Service |
a04d08 |
sys_mock = mock.patch('cloudinit.net.get_sys_class_path')
|
|
Packit Service |
a04d08 |
self.m_sys_path = sys_mock.start()
|
|
Packit Service |
a04d08 |
self.sysdir = self.tmp_dir() + '/'
|
|
Packit Service |
a04d08 |
self.m_sys_path.return_value = self.sysdir
|
|
Packit Service |
a04d08 |
self.addCleanup(sys_mock.stop)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp):
|
|
Packit Service |
a04d08 |
"""No required params for EphemeralIPv4Network can be None."""
|
|
Packit Service |
a04d08 |
required_params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
|
|
Packit Service |
a04d08 |
for key in required_params.keys():
|
|
Packit Service |
a04d08 |
params = copy.deepcopy(required_params)
|
|
Packit Service |
a04d08 |
params[key] = None
|
|
Packit Service |
a04d08 |
with self.assertRaises(ValueError) as context_manager:
|
|
Packit Service |
a04d08 |
net.EphemeralIPv4Network(**params)
|
|
Packit Service |
a04d08 |
error = context_manager.exception
|
|
Packit Service |
a04d08 |
self.assertIn('Cannot init network on', str(error))
|
|
Packit Service |
a04d08 |
self.assertEqual(0, m_subp.call_count)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp):
|
|
Packit Service |
a04d08 |
"""Raise an error when prefix_or_mask is not a netmask or prefix."""
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'broadcast': '192.168.2.255'}
|
|
Packit Service |
a04d08 |
invalid_masks = ('invalid', 'invalid.', '123.123.123')
|
|
Packit Service |
a04d08 |
for error_val in invalid_masks:
|
|
Packit Service |
a04d08 |
params['prefix_or_mask'] = error_val
|
|
Packit Service |
a04d08 |
with self.assertRaises(ValueError) as context_manager:
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
pass
|
|
Packit Service |
a04d08 |
error = context_manager.exception
|
|
Packit Service |
a04d08 |
self.assertIn('Cannot setup network: netmask', str(error))
|
|
Packit Service |
a04d08 |
self.assertEqual(0, m_subp.call_count)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_performs_teardown(self, m_subp):
|
|
Packit Service |
a04d08 |
"""EphemeralIPv4Network performs teardown on the device if setup."""
|
|
Packit Service |
a04d08 |
expected_setup_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'}),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
|
|
Packit Service |
a04d08 |
capture=True)]
|
|
Packit Service |
a04d08 |
expected_teardown_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0',
|
|
Packit Service |
a04d08 |
'down'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'del', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'dev', 'eth0'], capture=True)]
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
self.assertEqual(expected_setup_calls, m_subp.call_args_list)
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls(expected_teardown_calls)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.readurl')
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_no_network_if_url_connectivity(
|
|
Packit Service |
a04d08 |
self, m_readurl, m_subp):
|
|
Packit Service |
a04d08 |
"""No network setup is performed if we can successfully connect to
|
|
Packit Service |
a04d08 |
connectivity_url."""
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
|
|
Packit Service |
a04d08 |
'connectivity_url': 'http://example.org/index.html'}
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
self.assertEqual([mock.call('http://example.org/index.html',
|
|
Packit Service |
a04d08 |
timeout=5)], m_readurl.call_args_list)
|
|
Packit Service |
a04d08 |
# Ensure that no teardown happens:
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls([])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp):
|
|
Packit Service |
a04d08 |
"""EphemeralIPv4Network handles exception when address is setup.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
It performs no cleanup as the interface was already setup.
|
|
Packit Service |
a04d08 |
"""
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255'}
|
|
Packit Service |
a04d08 |
m_subp.side_effect = ProcessExecutionError(
|
|
Packit Service |
a04d08 |
'', 'RTNETLINK answers: File exists', 2)
|
|
Packit Service |
a04d08 |
expected_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'})]
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
pass
|
|
Packit Service |
a04d08 |
self.assertEqual(expected_calls, m_subp.call_args_list)
|
|
Packit Service |
a04d08 |
self.assertIn(
|
|
Packit Service |
a04d08 |
'Skip ephemeral network setup, eth0 already has address',
|
|
Packit Service |
a04d08 |
self.logs.getvalue())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_with_prefix(self, m_subp):
|
|
Packit Service |
a04d08 |
"""EphemeralIPv4Network takes a valid prefix to setup the network."""
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '24', 'broadcast': '192.168.2.255'}
|
|
Packit Service |
a04d08 |
for prefix_val in ['24', 16]: # prefix can be int or string
|
|
Packit Service |
a04d08 |
params['prefix_or_mask'] = prefix_val
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
pass
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls([mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'})])
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls([mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/16',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'})])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_with_new_default_route(self, m_subp):
|
|
Packit Service |
a04d08 |
"""Add the route when router is set and no default route exists."""
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
|
|
Packit Service |
a04d08 |
'router': '192.168.2.1'}
|
|
Packit Service |
a04d08 |
m_subp.return_value = '', '' # Empty response from ip route gw check
|
|
Packit Service |
a04d08 |
expected_setup_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'}),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
|
|
Packit Service |
a04d08 |
capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', 'route', 'show', '0.0.0.0/0'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(['ip', '-4', 'route', 'add', '192.168.2.1',
|
|
Packit Service |
a04d08 |
'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-4', 'route', 'add', 'default', 'via',
|
|
Packit Service |
a04d08 |
'192.168.2.1', 'dev', 'eth0'], capture=True)]
|
|
Packit Service |
a04d08 |
expected_teardown_calls = [
|
|
Packit Service |
a04d08 |
mock.call(['ip', '-4', 'route', 'del', 'default', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True),
|
|
Packit Service |
a04d08 |
mock.call(['ip', '-4', 'route', 'del', '192.168.2.1',
|
|
Packit Service |
a04d08 |
'dev', 'eth0', 'src', '192.168.2.2'], capture=True),
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
self.assertEqual(expected_setup_calls, m_subp.call_args_list)
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls(expected_teardown_calls)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp):
|
|
Packit Service |
a04d08 |
params = {
|
|
Packit Service |
a04d08 |
'interface': 'eth0', 'ip': '192.168.2.2',
|
|
Packit Service |
a04d08 |
'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255',
|
|
Packit Service |
a04d08 |
'static_routes': [('169.254.169.254/32', '192.168.2.1'),
|
|
Packit Service |
a04d08 |
('0.0.0.0/0', '192.168.2.1')],
|
|
Packit Service |
a04d08 |
'router': '192.168.2.1'}
|
|
Packit Service |
a04d08 |
expected_setup_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'add', '192.168.2.2/24',
|
|
Packit Service |
a04d08 |
'broadcast', '192.168.2.255', 'dev', 'eth0'],
|
|
Packit Service |
a04d08 |
capture=True, update_env={'LANG': 'C'}),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'link', 'set', 'dev', 'eth0', 'up'],
|
|
Packit Service |
a04d08 |
capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-4', 'route', 'add', '169.254.169.254/32',
|
|
Packit Service |
a04d08 |
'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-4', 'route', 'add', '0.0.0.0/0',
|
|
Packit Service |
a04d08 |
'via', '192.168.2.1', 'dev', 'eth0'], capture=True)]
|
|
Packit Service |
a04d08 |
expected_teardown_calls = [
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-4', 'route', 'del', '0.0.0.0/0',
|
|
Packit Service |
a04d08 |
'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-4', 'route', 'del', '169.254.169.254/32',
|
|
Packit Service |
a04d08 |
'via', '192.168.2.1', 'dev', 'eth0'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'link', 'set', 'dev',
|
|
Packit Service |
a04d08 |
'eth0', 'down'], capture=True),
|
|
Packit Service |
a04d08 |
mock.call(
|
|
Packit Service |
a04d08 |
['ip', '-family', 'inet', 'addr', 'del',
|
|
Packit Service |
a04d08 |
'192.168.2.2/24', 'dev', 'eth0'], capture=True)
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
with net.EphemeralIPv4Network(**params):
|
|
Packit Service |
a04d08 |
self.assertEqual(expected_setup_calls, m_subp.call_args_list)
|
|
Packit Service |
a04d08 |
m_subp.assert_has_calls(expected_setup_calls + expected_teardown_calls)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestApplyNetworkCfgNames(CiTestCase):
|
|
Packit Service |
a04d08 |
V1_CONFIG = textwrap.dedent("""\
|
|
Packit Service |
a04d08 |
version: 1
|
|
Packit Service |
a04d08 |
config:
|
|
Packit Service |
a04d08 |
- type: physical
|
|
Packit Service |
a04d08 |
name: interface0
|
|
Packit Service |
a04d08 |
mac_address: "52:54:00:12:34:00"
|
|
Packit Service |
a04d08 |
subnets:
|
|
Packit Service |
a04d08 |
- type: static
|
|
Packit Service |
a04d08 |
address: 10.0.2.15
|
|
Packit Service |
a04d08 |
netmask: 255.255.255.0
|
|
Packit Service |
a04d08 |
gateway: 10.0.2.2
|
|
Packit Service |
a04d08 |
""")
|
|
Packit Service |
a04d08 |
V2_CONFIG = textwrap.dedent("""\
|
|
Packit Service |
a04d08 |
version: 2
|
|
Packit Service |
a04d08 |
ethernets:
|
|
Packit Service |
a04d08 |
interface0:
|
|
Packit Service |
a04d08 |
match:
|
|
Packit Service |
a04d08 |
macaddress: "52:54:00:12:34:00"
|
|
Packit Service |
a04d08 |
addresses:
|
|
Packit Service |
a04d08 |
- 10.0.2.15/24
|
|
Packit Service |
a04d08 |
gateway4: 10.0.2.2
|
|
Packit Service |
a04d08 |
set-name: interface0
|
|
Packit Service |
a04d08 |
""")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
V2_CONFIG_NO_SETNAME = textwrap.dedent("""\
|
|
Packit Service |
a04d08 |
version: 2
|
|
Packit Service |
a04d08 |
ethernets:
|
|
Packit Service |
a04d08 |
interface0:
|
|
Packit Service |
a04d08 |
match:
|
|
Packit Service |
a04d08 |
macaddress: "52:54:00:12:34:00"
|
|
Packit Service |
a04d08 |
addresses:
|
|
Packit Service |
a04d08 |
- 10.0.2.15/24
|
|
Packit Service |
a04d08 |
gateway4: 10.0.2.2
|
|
Packit Service |
a04d08 |
""")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
V2_CONFIG_NO_MAC = textwrap.dedent("""\
|
|
Packit Service |
a04d08 |
version: 2
|
|
Packit Service |
a04d08 |
ethernets:
|
|
Packit Service |
a04d08 |
interface0:
|
|
Packit Service |
a04d08 |
match:
|
|
Packit Service |
a04d08 |
driver: virtio-net
|
|
Packit Service |
a04d08 |
addresses:
|
|
Packit Service |
a04d08 |
- 10.0.2.15/24
|
|
Packit Service |
a04d08 |
gateway4: 10.0.2.2
|
|
Packit Service |
a04d08 |
set-name: interface0
|
|
Packit Service |
a04d08 |
""")
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.device_devid')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.device_driver')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net._rename_interfaces')
|
|
Packit Service |
a04d08 |
def test_apply_v1_renames(self, m_rename_interfaces, m_device_driver,
|
|
Packit Service |
a04d08 |
m_device_devid):
|
|
Packit Service |
a04d08 |
m_device_driver.return_value = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_device_devid.return_value = '0x15d8'
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
net.apply_network_config_names(yaml.load(self.V1_CONFIG))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
|
|
Packit Service |
a04d08 |
m_rename_interfaces.assert_called_with([call])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.device_devid')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.device_driver')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net._rename_interfaces')
|
|
Packit Service |
a04d08 |
def test_apply_v2_renames(self, m_rename_interfaces, m_device_driver,
|
|
Packit Service |
a04d08 |
m_device_devid):
|
|
Packit Service |
a04d08 |
m_device_driver.return_value = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_device_devid.return_value = '0x15d8'
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
net.apply_network_config_names(yaml.load(self.V2_CONFIG))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
call = ['52:54:00:12:34:00', 'interface0', 'virtio_net', '0x15d8']
|
|
Packit Service |
a04d08 |
m_rename_interfaces.assert_called_with([call])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net._rename_interfaces')
|
|
Packit Service |
a04d08 |
def test_apply_v2_renames_skips_without_setname(self, m_rename_interfaces):
|
|
Packit Service |
a04d08 |
net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_SETNAME))
|
|
Packit Service |
a04d08 |
m_rename_interfaces.assert_called_with([])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net._rename_interfaces')
|
|
Packit Service |
a04d08 |
def test_apply_v2_renames_skips_without_mac(self, m_rename_interfaces):
|
|
Packit Service |
a04d08 |
net.apply_network_config_names(yaml.load(self.V2_CONFIG_NO_MAC))
|
|
Packit Service |
a04d08 |
m_rename_interfaces.assert_called_with([])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_apply_v2_renames_raises_runtime_error_on_unknown_version(self):
|
|
Packit Service |
a04d08 |
with self.assertRaises(RuntimeError):
|
|
Packit Service |
a04d08 |
net.apply_network_config_names(yaml.load("version: 3"))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestHasURLConnectivity(HttprettyTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestHasURLConnectivity, self).setUp()
|
|
Packit Service |
a04d08 |
self.url = 'http://fake/'
|
|
Packit Service |
a04d08 |
self.kwargs = {'allow_redirects': True, 'timeout': 5.0}
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.readurl')
|
|
Packit Service |
a04d08 |
def test_url_timeout_on_connectivity_check(self, m_readurl):
|
|
Packit Service |
a04d08 |
"""A timeout of 5 seconds is provided when reading a url."""
|
|
Packit Service |
a04d08 |
self.assertTrue(
|
|
Packit Service |
a04d08 |
net.has_url_connectivity(self.url), 'Expected True on url connect')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_true_on_url_connectivity_success(self):
|
|
Packit Service |
a04d08 |
httpretty.register_uri(httpretty.GET, self.url)
|
|
Packit Service |
a04d08 |
self.assertTrue(
|
|
Packit Service |
a04d08 |
net.has_url_connectivity(self.url), 'Expected True on url connect')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('requests.Session.request')
|
|
Packit Service |
a04d08 |
def test_true_on_url_connectivity_timeout(self, m_request):
|
|
Packit Service |
a04d08 |
"""A timeout raised accessing the url will return False."""
|
|
Packit Service |
a04d08 |
m_request.side_effect = requests.Timeout('Fake Connection Timeout')
|
|
Packit Service |
a04d08 |
self.assertFalse(
|
|
Packit Service |
a04d08 |
net.has_url_connectivity(self.url),
|
|
Packit Service |
a04d08 |
'Expected False on url timeout')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_true_on_url_connectivity_failure(self):
|
|
Packit Service |
a04d08 |
httpretty.register_uri(httpretty.GET, self.url, body={}, status=404)
|
|
Packit Service |
a04d08 |
self.assertFalse(
|
|
Packit Service |
a04d08 |
net.has_url_connectivity(self.url), 'Expected False on url fail')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _mk_v1_phys(mac, name, driver, device_id):
|
|
Packit Service |
a04d08 |
v1_cfg = {'type': 'physical', 'name': name, 'mac_address': mac}
|
|
Packit Service |
a04d08 |
params = {}
|
|
Packit Service |
a04d08 |
if driver:
|
|
Packit Service |
a04d08 |
params.update({'driver': driver})
|
|
Packit Service |
a04d08 |
if device_id:
|
|
Packit Service |
a04d08 |
params.update({'device_id': device_id})
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
if params:
|
|
Packit Service |
a04d08 |
v1_cfg.update({'params': params})
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
return v1_cfg
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _mk_v2_phys(mac, name, driver=None, device_id=None):
|
|
Packit Service |
a04d08 |
v2_cfg = {'set-name': name, 'match': {'macaddress': mac}}
|
|
Packit Service |
a04d08 |
if driver:
|
|
Packit Service |
a04d08 |
v2_cfg['match'].update({'driver': driver})
|
|
Packit Service |
a04d08 |
if device_id:
|
|
Packit Service |
a04d08 |
v2_cfg['match'].update({'device_id': device_id})
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
return v2_cfg
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestExtractPhysdevs(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestExtractPhysdevs, self).setUp()
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.device_driver', 'm_driver')
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.device_devid', 'm_devid')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_extract_physdevs_looks_up_driver_v1(self):
|
|
Packit Service |
a04d08 |
driver = 'virtio'
|
|
Packit Service |
a04d08 |
self.m_driver.return_value = driver
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 1,
|
|
Packit Service |
a04d08 |
'config': [_mk_v1_phys(*args) for args in physdevs],
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
# insert the driver value for verification
|
|
Packit Service |
a04d08 |
physdevs[0][2] = driver
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
self.m_driver.assert_called_with('eth0')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_extract_physdevs_looks_up_driver_v2(self):
|
|
Packit Service |
a04d08 |
driver = 'virtio'
|
|
Packit Service |
a04d08 |
self.m_driver.return_value = driver
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', None, '0x1000'],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 2,
|
|
Packit Service |
a04d08 |
'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
# insert the driver value for verification
|
|
Packit Service |
a04d08 |
physdevs[0][2] = driver
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
self.m_driver.assert_called_with('eth0')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_extract_physdevs_looks_up_devid_v1(self):
|
|
Packit Service |
a04d08 |
devid = '0x1000'
|
|
Packit Service |
a04d08 |
self.m_devid.return_value = devid
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 1,
|
|
Packit Service |
a04d08 |
'config': [_mk_v1_phys(*args) for args in physdevs],
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
# insert the driver value for verification
|
|
Packit Service |
a04d08 |
physdevs[0][3] = devid
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
self.m_devid.assert_called_with('eth0')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_extract_physdevs_looks_up_devid_v2(self):
|
|
Packit Service |
a04d08 |
devid = '0x1000'
|
|
Packit Service |
a04d08 |
self.m_devid.return_value = devid
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', None],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 2,
|
|
Packit Service |
a04d08 |
'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
# insert the driver value for verification
|
|
Packit Service |
a04d08 |
physdevs[0][3] = devid
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
self.m_devid.assert_called_with('eth0')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_v1_type_physical(self):
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
|
|
Packit Service |
a04d08 |
['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
|
|
Packit Service |
a04d08 |
['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 1,
|
|
Packit Service |
a04d08 |
'config': [_mk_v1_phys(*args) for args in physdevs],
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_v2_type_physical(self):
|
|
Packit Service |
a04d08 |
physdevs = [
|
|
Packit Service |
a04d08 |
['aa:bb:cc:dd:ee:ff', 'eth0', 'virtio', '0x1000'],
|
|
Packit Service |
a04d08 |
['00:11:22:33:44:55', 'ens3', 'e1000', '0x1643'],
|
|
Packit Service |
a04d08 |
['09:87:65:43:21:10', 'ens0p1', 'mlx4_core', '0:0:1000'],
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 2,
|
|
Packit Service |
a04d08 |
'ethernets': {args[1]: _mk_v2_phys(*args) for args in physdevs},
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
self.assertEqual(sorted(physdevs),
|
|
Packit Service |
a04d08 |
sorted(net.extract_physdevs(netcfg)))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_v2_type_physical_skips_if_no_set_name(self):
|
|
Packit Service |
a04d08 |
netcfg = {
|
|
Packit Service |
a04d08 |
'version': 2,
|
|
Packit Service |
a04d08 |
'ethernets': {
|
|
Packit Service |
a04d08 |
'ens3': {
|
|
Packit Service |
a04d08 |
'match': {'macaddress': '00:11:22:33:44:55'},
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
self.assertEqual([], net.extract_physdevs(netcfg))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_runtime_error_on_unknown_netcfg_version(self):
|
|
Packit Service |
a04d08 |
with self.assertRaises(RuntimeError):
|
|
Packit Service |
a04d08 |
net.extract_physdevs({'version': 3, 'awesome_config': []})
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestNetFailOver(CiTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestNetFailOver, self).setUp()
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.util', 'm_util')
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.read_sys_net', 'm_read_sys_net')
|
|
Packit Service |
a04d08 |
self.add_patch('cloudinit.net.device_driver', 'm_device_driver')
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_dev_features(self):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
features = self.random_string()
|
|
Packit Service |
a04d08 |
self.m_read_sys_net.return_value = features
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
self.assertEqual(features, net.get_dev_features(devname))
|
|
Packit Service |
a04d08 |
self.assertEqual(1, self.m_read_sys_net.call_count)
|
|
Packit Service |
a04d08 |
self.assertEqual(mock.call(devname, 'device/features'),
|
|
Packit Service |
a04d08 |
self.m_read_sys_net.call_args_list[0])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_dev_features_none_returns_empty_string(self):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
self.m_read_sys_net.side_effect = Exception('error')
|
|
Packit Service |
a04d08 |
self.assertEqual('', net.get_dev_features(devname))
|
|
Packit Service |
a04d08 |
self.assertEqual(1, self.m_read_sys_net.call_count)
|
|
Packit Service |
a04d08 |
self.assertEqual(mock.call(devname, 'device/features'),
|
|
Packit Service |
a04d08 |
self.m_read_sys_net.call_args_list[0])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.get_dev_features')
|
|
Packit Service |
a04d08 |
def test_has_netfail_standby_feature(self, m_dev_features):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
standby_features = ('0' * 62) + '1' + '0'
|
|
Packit Service |
a04d08 |
m_dev_features.return_value = standby_features
|
|
Packit Service |
a04d08 |
self.assertTrue(net.has_netfail_standby_feature(devname))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.get_dev_features')
|
|
Packit Service |
a04d08 |
def test_has_netfail_standby_feature_short_is_false(self, m_dev_features):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
standby_features = self.random_string()
|
|
Packit Service |
a04d08 |
m_dev_features.return_value = standby_features
|
|
Packit Service |
a04d08 |
self.assertFalse(net.has_netfail_standby_feature(devname))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.get_dev_features')
|
|
Packit Service |
a04d08 |
def test_has_netfail_standby_feature_not_present_is_false(self,
|
|
Packit Service |
a04d08 |
m_dev_features):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
standby_features = '0' * 64
|
|
Packit Service |
a04d08 |
m_dev_features.return_value = standby_features
|
|
Packit Service |
a04d08 |
self.assertFalse(net.has_netfail_standby_feature(devname))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.get_dev_features')
|
|
Packit Service |
a04d08 |
def test_has_netfail_standby_feature_no_features_is_false(self,
|
|
Packit Service |
a04d08 |
m_dev_features):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
standby_features = None
|
|
Packit Service |
a04d08 |
m_dev_features.return_value = standby_features
|
|
Packit Service |
a04d08 |
self.assertFalse(net.has_netfail_standby_feature(devname))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_master(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = False # no master sysfs attr
|
|
Packit Service |
a04d08 |
m_standby.return_value = True # has standby feature flag
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_netfail_master(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_master_checks_master_attr(self, m_sysdev):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_sysdev.return_value = self.random_string()
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_master(devname, driver))
|
|
Packit Service |
a04d08 |
self.assertEqual(1, m_sysdev.call_count)
|
|
Packit Service |
a04d08 |
self.assertEqual(mock.call(devname, path='master'),
|
|
Packit Service |
a04d08 |
m_sysdev.call_args_list[0])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_master_wrong_driver(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string()
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_master(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_master_has_master_attr(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_master(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_master_no_standby_feat(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = False # no master sysfs attr
|
|
Packit Service |
a04d08 |
m_standby.return_value = False # no standby feature flag
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_master(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string() # device not virtio_net
|
|
Packit Service |
a04d08 |
master_devname = self.random_string()
|
|
Packit Service |
a04d08 |
m_sysdev.return_value = "%s/%s" % (self.random_string(),
|
|
Packit Service |
a04d08 |
master_devname)
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
self.m_device_driver.return_value = 'virtio_net' # master virtio_net
|
|
Packit Service |
a04d08 |
m_standby.return_value = True # has standby feature flag
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_netfail_primary(devname, driver))
|
|
Packit Service |
a04d08 |
self.assertEqual(1, self.m_device_driver.call_count)
|
|
Packit Service |
a04d08 |
self.assertEqual(mock.call(master_devname),
|
|
Packit Service |
a04d08 |
self.m_device_driver.call_args_list[0])
|
|
Packit Service |
a04d08 |
self.assertEqual(1, m_standby.call_count)
|
|
Packit Service |
a04d08 |
self.assertEqual(mock.call(master_devname),
|
|
Packit Service |
a04d08 |
m_standby.call_args_list[0])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_primary_wrong_driver(self, m_sysdev, m_exists,
|
|
Packit Service |
a04d08 |
m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_primary(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string() # device not virtio_net
|
|
Packit Service |
a04d08 |
m_exists.return_value = False # no master sysfs attr
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_primary(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_primary_bad_master(self, m_sysdev, m_exists,
|
|
Packit Service |
a04d08 |
m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string() # device not virtio_net
|
|
Packit Service |
a04d08 |
master_devname = self.random_string()
|
|
Packit Service |
a04d08 |
m_sysdev.return_value = "%s/%s" % (self.random_string(),
|
|
Packit Service |
a04d08 |
master_devname)
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
self.m_device_driver.return_value = 'XXXX' # master not virtio_net
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_primary(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.sys_dev_path')
|
|
Packit Service |
a04d08 |
def test_is_netfail_primary_no_standby(self, m_sysdev, m_exists,
|
|
Packit Service |
a04d08 |
m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string() # device not virtio_net
|
|
Packit Service |
a04d08 |
master_devname = self.random_string()
|
|
Packit Service |
a04d08 |
m_sysdev.return_value = "%s/%s" % (self.random_string(),
|
|
Packit Service |
a04d08 |
master_devname)
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
self.m_device_driver.return_value = 'virtio_net' # master virtio_net
|
|
Packit Service |
a04d08 |
m_standby.return_value = False # master has no standby feature flag
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_primary(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_standby(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
m_standby.return_value = True # has standby feature flag
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_netfail_standby(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_standby_wrong_driver(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string()
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_standby(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_standby_no_master(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = False # has master sysfs attr
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_standby(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.has_netfail_standby_feature')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.os.path.exists')
|
|
Packit Service |
a04d08 |
def test_is_netfail_standby_no_standby_feature(self, m_exists, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = 'virtio_net'
|
|
Packit Service |
a04d08 |
m_exists.return_value = True # has master sysfs attr
|
|
Packit Service |
a04d08 |
m_standby.return_value = False # has standby feature flag
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfail_standby(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_standby')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_primary')
|
|
Packit Service |
a04d08 |
def test_is_netfailover_primary(self, m_primary, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string()
|
|
Packit Service |
a04d08 |
m_primary.return_value = True
|
|
Packit Service |
a04d08 |
m_standby.return_value = False
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_netfailover(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_standby')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_primary')
|
|
Packit Service |
a04d08 |
def test_is_netfailover_standby(self, m_primary, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string()
|
|
Packit Service |
a04d08 |
m_primary.return_value = False
|
|
Packit Service |
a04d08 |
m_standby.return_value = True
|
|
Packit Service |
a04d08 |
self.assertTrue(net.is_netfailover(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_standby')
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.net.is_netfail_primary')
|
|
Packit Service |
a04d08 |
def test_is_netfailover_returns_false(self, m_primary, m_standby):
|
|
Packit Service |
a04d08 |
devname = self.random_string()
|
|
Packit Service |
a04d08 |
driver = self.random_string()
|
|
Packit Service |
a04d08 |
m_primary.return_value = False
|
|
Packit Service |
a04d08 |
m_standby.return_value = False
|
|
Packit Service |
a04d08 |
self.assertFalse(net.is_netfailover(devname, driver))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
class TestIsIpAddress:
|
|
Packit Service |
9bfd13 |
"""Tests for net.is_ip_address.
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
Instead of testing with values we rely on the ipaddress stdlib module to
|
|
Packit Service |
9bfd13 |
handle all values correctly, so simply test that is_ip_address defers to
|
|
Packit Service |
9bfd13 |
the ipaddress module correctly.
|
|
Packit Service |
9bfd13 |
"""
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
@pytest.mark.parametrize('ip_address_side_effect,expected_return', (
|
|
Packit Service |
9bfd13 |
(ValueError, False),
|
|
Packit Service |
9bfd13 |
(lambda _: ipaddress.IPv4Address('192.168.0.1'), True),
|
|
Packit Service |
9bfd13 |
(lambda _: ipaddress.IPv6Address('2001:db8::'), True),
|
|
Packit Service |
9bfd13 |
))
|
|
Packit Service |
9bfd13 |
def test_is_ip_address(self, ip_address_side_effect, expected_return):
|
|
Packit Service |
9bfd13 |
with mock.patch('cloudinit.net.ipaddress.ip_address',
|
|
Packit Service |
9bfd13 |
side_effect=ip_address_side_effect) as m_ip_address:
|
|
Packit Service |
9bfd13 |
ret = net.is_ip_address(mock.sentinel.ip_address_in)
|
|
Packit Service |
9bfd13 |
assert expected_return == ret
|
|
Packit Service |
9bfd13 |
expected_call = mock.call(mock.sentinel.ip_address_in)
|
|
Packit Service |
9bfd13 |
assert [expected_call] == m_ip_address.call_args_list
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
class TestIsIpv4Address:
|
|
Packit Service |
9bfd13 |
"""Tests for net.is_ipv4_address.
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
Instead of testing with values we rely on the ipaddress stdlib module to
|
|
Packit Service |
9bfd13 |
handle all values correctly, so simply test that is_ipv4_address defers to
|
|
Packit Service |
9bfd13 |
the ipaddress module correctly.
|
|
Packit Service |
9bfd13 |
"""
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
@pytest.mark.parametrize('ipv4address_mock,expected_return', (
|
|
Packit Service |
9bfd13 |
(mock.Mock(side_effect=ValueError), False),
|
|
Packit Service |
9bfd13 |
(mock.Mock(return_value=ipaddress.IPv4Address('192.168.0.1')), True),
|
|
Packit Service |
9bfd13 |
))
|
|
Packit Service |
9bfd13 |
def test_is_ip_address(self, ipv4address_mock, expected_return):
|
|
Packit Service |
9bfd13 |
with mock.patch('cloudinit.net.ipaddress.IPv4Address',
|
|
Packit Service |
9bfd13 |
ipv4address_mock) as m_ipv4address:
|
|
Packit Service |
9bfd13 |
ret = net.is_ipv4_address(mock.sentinel.ip_address_in)
|
|
Packit Service |
9bfd13 |
assert expected_return == ret
|
|
Packit Service |
9bfd13 |
expected_call = mock.call(mock.sentinel.ip_address_in)
|
|
Packit Service |
9bfd13 |
assert [expected_call] == m_ipv4address.call_args_list
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
9bfd13 |
|
|
Packit Service |
a04d08 |
# vi: ts=4 expandtab
|