mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
Migrate tests/integration/modules/test_nilrt_ip.py to pytest
Signed-off-by: Pedro Algarvio <palgarvio@vmware.com>
This commit is contained in:
Pedro Algarvio
Pedro Algarvio
parent
db054f6232
commit
49b494014e
2 changed files with 386 additions and 424 deletions
|
|
@ -1,424 +0,0 @@
|
|||
"""
|
||||
integration tests for nilirt_ip
|
||||
"""
|
||||
|
||||
import configparser
|
||||
import re
|
||||
import shutil
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
import salt.modules.nilrt_ip as ip
|
||||
import salt.utils.files
|
||||
import salt.utils.platform
|
||||
from tests.support.case import ModuleCase
|
||||
from tests.support.helpers import requires_system_grains
|
||||
|
||||
try:
|
||||
import pyiface
|
||||
from pyiface.ifreqioctls import IFF_LOOPBACK, IFF_RUNNING
|
||||
except ImportError:
|
||||
pyiface = None
|
||||
|
||||
try:
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
except ImportError:
|
||||
CaseInsensitiveDict = None
|
||||
|
||||
INTERFACE_FOR_TEST = "eth1"
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.skip_initial_gh_actions_failure(
|
||||
reason="This was skipped on older golden images and is failing on newer."
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.skip_if_not_root
|
||||
@pytest.mark.skipif(
|
||||
pyiface is None, reason="The python pyiface package is not installed"
|
||||
)
|
||||
@pytest.mark.skipif(
|
||||
CaseInsensitiveDict is None, reason="The python package requests is not installed"
|
||||
)
|
||||
@pytest.mark.destructive_test
|
||||
class NilrtIpModuleTest(ModuleCase):
|
||||
"""
|
||||
Validate the nilrt_ip module
|
||||
"""
|
||||
|
||||
@requires_system_grains
|
||||
@classmethod
|
||||
def setUpClass(cls, grains): # pylint: disable=arguments-differ
|
||||
if grains["os_family"] != "NILinuxRT":
|
||||
raise pytest.skip.Exception(
|
||||
"Tests applicable only to NILinuxRT", _use_item_location=True
|
||||
)
|
||||
cls.initialState = {}
|
||||
cls.grains = grains
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.initialState = cls.grains = None
|
||||
|
||||
@staticmethod
|
||||
def setup_loader_modules():
|
||||
"""
|
||||
Setup loader modules
|
||||
"""
|
||||
return {ip: {}}
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
Get current settings
|
||||
"""
|
||||
# save files from var/lib/connman*
|
||||
super().setUp()
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
shutil.move("/etc/natinst/share/ni-rt.ini", "/tmp/ni-rt.ini")
|
||||
else:
|
||||
shutil.move("/var/lib/connman", "/tmp/connman")
|
||||
|
||||
def tearDown(self):
|
||||
"""
|
||||
Reset to original settings
|
||||
"""
|
||||
# restore files
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
shutil.move("/tmp/ni-rt.ini", "/etc/natinst/share/ni-rt.ini")
|
||||
self.run_function("cmd.run", ["/etc/init.d/networking restart"])
|
||||
else:
|
||||
shutil.move("/tmp/connman", "/var/lib/connman")
|
||||
self.run_function("service.restart", ["connman"])
|
||||
time.sleep(10) # wait 10 seconds for connman to be fully loaded
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
self.run_function("ip.up", [interface.name])
|
||||
|
||||
@staticmethod
|
||||
def __connected(interface):
|
||||
"""
|
||||
Check if an interface is up or down
|
||||
:param interface: pyiface.Interface object
|
||||
:return: True, if interface is up, otherwise False.
|
||||
"""
|
||||
return interface.flags & IFF_RUNNING != 0
|
||||
|
||||
@staticmethod
|
||||
def __interfaces():
|
||||
"""
|
||||
Return the list of all interfaces without loopback
|
||||
"""
|
||||
return [
|
||||
interface
|
||||
for interface in pyiface.getIfaces()
|
||||
if interface.flags & IFF_LOOPBACK == 0
|
||||
]
|
||||
|
||||
def __check_ethercat(self):
|
||||
"""
|
||||
Check if ethercat is installed.
|
||||
|
||||
:return: True if ethercat is installed, otherwise False.
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] != "nilrt":
|
||||
return False
|
||||
with salt.utils.files.fopen("/etc/natinst/share/ni-rt.ini", "r") as config_file:
|
||||
config_parser = configparser.RawConfigParser(dict_type=CaseInsensitiveDict)
|
||||
config_parser.readfp(config_file)
|
||||
return (
|
||||
"ethercat"
|
||||
in config_parser.get(
|
||||
"lvrt", "AdditionalNetworkProtocols", fallback=""
|
||||
).lower()
|
||||
)
|
||||
|
||||
def test_down(self):
|
||||
"""
|
||||
Test ip.down function
|
||||
"""
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
result = self.run_function("ip.down", [interface.name])
|
||||
self.assertTrue(result)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
self.assertEqual(interface["adapter_mode"], "disabled")
|
||||
self.assertFalse(
|
||||
self.__connected(pyiface.Interface(name=interface["connectionid"]))
|
||||
)
|
||||
|
||||
def test_up(self):
|
||||
"""
|
||||
Test ip.up function
|
||||
"""
|
||||
interfaces = self.__interfaces()
|
||||
# first down all interfaces
|
||||
for interface in interfaces:
|
||||
self.run_function("ip.down", [interface.name])
|
||||
self.assertFalse(self.__connected(interface))
|
||||
# up interfaces
|
||||
for interface in interfaces:
|
||||
result = self.run_function("ip.up", [interface.name])
|
||||
self.assertTrue(result)
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
|
||||
def test_set_dhcp_linklocal_all(self):
|
||||
"""
|
||||
Test ip.set_dhcp_linklocal_all function
|
||||
"""
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
result = self.run_function("ip.set_dhcp_linklocal_all", [interface.name])
|
||||
self.assertTrue(result)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
|
||||
def test_set_dhcp_only_all(self):
|
||||
"""
|
||||
Test ip.set_dhcp_only_all function
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] != "nilrt":
|
||||
self.skipTest("Test not applicable to newer nilrt")
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
result = self.run_function("ip.set_dhcp_only_all", [interface.name])
|
||||
self.assertTrue(result)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_only")
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
|
||||
def test_set_linklocal_only_all(self):
|
||||
"""
|
||||
Test ip.set_linklocal_only_all function
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] != "nilrt":
|
||||
self.skipTest("Test not applicable to newer nilrt")
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
result = self.run_function("ip.set_linklocal_only_all", [interface.name])
|
||||
self.assertTrue(result)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "linklocal_only")
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
|
||||
def test_static_all(self):
|
||||
"""
|
||||
Test ip.set_static_all function
|
||||
"""
|
||||
interfaces = self.__interfaces()
|
||||
for interface in interfaces:
|
||||
result = self.run_function(
|
||||
"ip.set_static_all",
|
||||
[
|
||||
interface.name,
|
||||
"192.168.10.4",
|
||||
"255.255.255.0",
|
||||
"192.168.10.1",
|
||||
"8.8.4.4 8.8.8.8",
|
||||
],
|
||||
)
|
||||
self.assertTrue(result)
|
||||
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if self.grains["lsb_distrib_id"] != "nilrt":
|
||||
self.assertIn("8.8.4.4", interface["ipv4"]["dns"])
|
||||
self.assertIn("8.8.8.8", interface["ipv4"]["dns"])
|
||||
else:
|
||||
self.assertEqual(interface["ipv4"]["dns"], ["8.8.4.4"])
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "static")
|
||||
self.assertEqual(interface["ipv4"]["address"], "192.168.10.4")
|
||||
self.assertEqual(interface["ipv4"]["netmask"], "255.255.255.0")
|
||||
self.assertEqual(interface["ipv4"]["gateway"], "192.168.10.1")
|
||||
|
||||
def test_supported_adapter_modes(self):
|
||||
"""
|
||||
Test supported adapter modes for each interface
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] != "nilrt":
|
||||
self.skipTest("Test is just for older nilrt distros")
|
||||
interface_pattern = re.compile("^eth[0-9]+$")
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == "eth0":
|
||||
self.assertEqual(interface["supported_adapter_modes"], ["tcpip"])
|
||||
else:
|
||||
self.assertIn("tcpip", interface["supported_adapter_modes"])
|
||||
if not interface_pattern.match(interface["connectionid"]):
|
||||
self.assertNotIn("ethercat", interface["supported_adapter_modes"])
|
||||
elif self.__check_ethercat():
|
||||
self.assertIn("ethercat", interface["supported_adapter_modes"])
|
||||
|
||||
def test_ethercat(self):
|
||||
"""
|
||||
Test ip.set_ethercat function
|
||||
"""
|
||||
if not self.__check_ethercat():
|
||||
self.skipTest("Test is just for systems with Ethercat")
|
||||
self.assertTrue(self.run_function("ip.set_ethercat", [INTERFACE_FOR_TEST, 19]))
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["adapter_mode"], "ethercat")
|
||||
self.assertEqual(int(interface["ethercat"]["masterid"]), 19)
|
||||
break
|
||||
self.assertTrue(
|
||||
self.run_function("ip.set_dhcp_linklocal_all", [INTERFACE_FOR_TEST])
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["adapter_mode"], "tcpip")
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
break
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_dhcp_disable(self):
|
||||
"""
|
||||
Test cases:
|
||||
- dhcp -> disable
|
||||
- disable -> dhcp
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
self.skipTest("Test is just for newer nilrt distros")
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function("ip.set_dhcp_linklocal_all", [INTERFACE_FOR_TEST])
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
break
|
||||
|
||||
self.assertTrue(self.run_function("ip.disable", [INTERFACE_FOR_TEST]))
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "disabled")
|
||||
break
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function("ip.set_dhcp_linklocal_all", [INTERFACE_FOR_TEST])
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
break
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_dhcp_static(self):
|
||||
"""
|
||||
Test cases:
|
||||
- dhcp -> static
|
||||
- static -> dhcp
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
self.skipTest("Test is just for newer nilrt distros")
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function("ip.set_dhcp_linklocal_all", [INTERFACE_FOR_TEST])
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
break
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function(
|
||||
"ip.set_static_all",
|
||||
[
|
||||
INTERFACE_FOR_TEST,
|
||||
"192.168.1.125",
|
||||
"255.255.255.0",
|
||||
"192.168.1.1",
|
||||
"8.8.8.8 8.8.8.4",
|
||||
],
|
||||
)
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "static")
|
||||
self.assertEqual(interface["ipv4"]["address"], "192.168.1.125")
|
||||
self.assertEqual(interface["ipv4"]["netmask"], "255.255.255.0")
|
||||
self.assertIn("8.8.8.4", interface["ipv4"]["dns"])
|
||||
self.assertIn("8.8.8.8", interface["ipv4"]["dns"])
|
||||
break
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function("ip.set_dhcp_linklocal_all", [INTERFACE_FOR_TEST])
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "dhcp_linklocal")
|
||||
break
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_static_disable(self):
|
||||
"""
|
||||
Test cases:
|
||||
- static -> disable
|
||||
- disable -> static
|
||||
"""
|
||||
if self.grains["lsb_distrib_id"] == "nilrt":
|
||||
self.skipTest("Test is just for newer nilrt distros")
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function(
|
||||
"ip.set_static_all",
|
||||
[
|
||||
INTERFACE_FOR_TEST,
|
||||
"192.168.1.125",
|
||||
"255.255.255.0",
|
||||
"192.168.1.1",
|
||||
"8.8.8.8",
|
||||
],
|
||||
)
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "static")
|
||||
self.assertEqual(interface["ipv4"]["address"], "192.168.1.125")
|
||||
self.assertEqual(interface["ipv4"]["netmask"], "255.255.255.0")
|
||||
self.assertEqual(interface["ipv4"]["dns"], ["8.8.8.8"])
|
||||
break
|
||||
|
||||
self.assertTrue(self.run_function("ip.disable", [INTERFACE_FOR_TEST]))
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "disabled")
|
||||
break
|
||||
|
||||
self.assertTrue(
|
||||
self.run_function(
|
||||
"ip.set_static_all",
|
||||
[INTERFACE_FOR_TEST, "192.168.1.125", "255.255.255.0", "192.168.1.1"],
|
||||
)
|
||||
)
|
||||
info = self.run_function("ip.get_interfaces_details", timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
self.assertEqual(interface["ipv4"]["requestmode"], "static")
|
||||
self.assertEqual(interface["ipv4"]["address"], "192.168.1.125")
|
||||
self.assertEqual(interface["ipv4"]["netmask"], "255.255.255.0")
|
||||
self.assertEqual(interface["ipv4"]["dns"], [])
|
||||
break
|
||||
386
tests/pytests/functional/modules/test_nilrt_ip.py
Normal file
386
tests/pytests/functional/modules/test_nilrt_ip.py
Normal file
|
|
@ -0,0 +1,386 @@
|
|||
"""
|
||||
integration tests for nilirt_ip
|
||||
"""
|
||||
|
||||
import configparser
|
||||
import re
|
||||
import shutil
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
import salt.utils.files
|
||||
import salt.utils.platform
|
||||
|
||||
try:
|
||||
import pyiface
|
||||
from pyiface.ifreqioctls import IFF_LOOPBACK, IFF_RUNNING
|
||||
except ImportError:
|
||||
pyiface = None
|
||||
|
||||
try:
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
except ImportError:
|
||||
CaseInsensitiveDict = None
|
||||
|
||||
INTERFACE_FOR_TEST = "eth1"
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.skip_if_not_root,
|
||||
pytest.mark.destructive_test,
|
||||
pytest.mark.skipif(
|
||||
'grains["os_family"] != "NILinuxRT"',
|
||||
reason="Tests applicable only to NILinuxRT",
|
||||
),
|
||||
pytest.mark.skip_initial_gh_actions_failure(
|
||||
reason="This was skipped on older golden images and is failing on newer."
|
||||
),
|
||||
pytest.mark.skipif(
|
||||
pyiface is None, reason="The python pyiface package is not installed"
|
||||
),
|
||||
pytest.mark.skipif(
|
||||
CaseInsensitiveDict is None,
|
||||
reason="The python package requests is not installed",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _connected(interface):
|
||||
"""
|
||||
Check if an interface is up or down
|
||||
:param interface: pyiface.Interface object
|
||||
:return: True, if interface is up, otherwise False.
|
||||
"""
|
||||
return interface.flags & IFF_RUNNING != 0
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def interfaces():
|
||||
return [
|
||||
interface
|
||||
for interface in pyiface.getIfaces()
|
||||
if interface.flags & IFF_LOOPBACK == 0
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ethercat_installed(grains):
|
||||
"""
|
||||
Check if ethercat is installed.
|
||||
|
||||
:return: True if ethercat is installed, otherwise False.
|
||||
"""
|
||||
if grains["lsb_distrib_id"] != "nilrt":
|
||||
return False
|
||||
|
||||
with salt.utils.files.fopen("/etc/natinst/share/ni-rt.ini", "r") as config_file:
|
||||
config_parser = configparser.RawConfigParser(dict_type=CaseInsensitiveDict)
|
||||
config_parser.readfp(config_file)
|
||||
protocols = config_parser.get(
|
||||
"lvrt", "AdditionalNetworkProtocols", fallback=""
|
||||
).lower()
|
||||
return "ethercat" in protocols
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ip(modules, grains, interfaces):
|
||||
try:
|
||||
# save files from var/lib/connman*
|
||||
if grains.get("lsb_distrib_id") == "nilrt":
|
||||
shutil.move("/etc/natinst/share/ni-rt.ini", "/tmp/ni-rt.ini")
|
||||
else:
|
||||
shutil.move("/var/lib/connman", "/tmp/connman")
|
||||
|
||||
# Run the test
|
||||
yield modules.ip
|
||||
finally:
|
||||
# Restore files
|
||||
if grains.get("lsb_distrib_id") == "nilrt":
|
||||
shutil.move("/tmp/ni-rt.ini", "/etc/natinst/share/ni-rt.ini")
|
||||
modules.cmd.run("/etc/init.d/networking restart")
|
||||
else:
|
||||
shutil.move("/tmp/connman", "/var/lib/connman")
|
||||
modules.service.restart("connman")
|
||||
time.sleep(10) # wait 10 seconds for connman to be fully loaded
|
||||
for interface in interfaces:
|
||||
modules.ip.up(interface.name)
|
||||
|
||||
|
||||
def test_down(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.down function
|
||||
"""
|
||||
for interface in interfaces:
|
||||
result = ip.down(interface.name)
|
||||
assert result
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
assert interface["adapter_mode"] == "disabled"
|
||||
assert not _connected(pyiface.Interface(name=interface["connectionid"]))
|
||||
|
||||
|
||||
def test_up(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.up function
|
||||
"""
|
||||
# first down all interfaces
|
||||
for interface in interfaces:
|
||||
ip.down(interface.name)
|
||||
assert not _connected(interface)
|
||||
# up interfaces
|
||||
for interface in interfaces:
|
||||
result = ip.up(interface.name)
|
||||
assert result
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
|
||||
|
||||
def test_set_dhcp_linklocal_all(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.set_dhcp_linklocal_all function
|
||||
"""
|
||||
for interface in interfaces:
|
||||
result = ip.set_dhcp_linklocal_all(interface.name)
|
||||
assert result
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
|
||||
|
||||
def test_set_dhcp_only_all(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.set_dhcp_only_all function
|
||||
"""
|
||||
if grains["lsb_distrib_id"] != "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test not applicable to newer nilrt", _use_item_location=True
|
||||
)
|
||||
for interface in interfaces:
|
||||
result = ip.set_dhcp_only_all(interface.name)
|
||||
assert result
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_only"
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
|
||||
|
||||
def test_set_linklocal_only_all(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.set_linklocal_only_all function
|
||||
"""
|
||||
if grains["lsb_distrib_id"] != "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test not applicable to newer nilrt", _use_item_location=True
|
||||
)
|
||||
for interface in interfaces:
|
||||
result = ip.set_linklocal_only_all(interface.name)
|
||||
assert result
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
assert interface["ipv4"]["requestmode"] == "linklocal_only"
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
|
||||
|
||||
def test_static_all(ip, grains, interfaces):
|
||||
"""
|
||||
Test ip.set_static_all function
|
||||
"""
|
||||
for interface in interfaces:
|
||||
result = ip.set_static_all(
|
||||
interface.name,
|
||||
"192.168.10.4",
|
||||
"255.255.255.0",
|
||||
"192.168.10.1",
|
||||
"8.8.4.4 8.8.8.8",
|
||||
)
|
||||
assert result
|
||||
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if grains["lsb_distrib_id"] != "nilrt":
|
||||
assert "8.8.4.4" in interface["ipv4"]["dns"]
|
||||
assert "8.8.8.8" in interface["ipv4"]["dns"]
|
||||
else:
|
||||
assert interface["ipv4"]["dns"] == ["8.8.4.4"]
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
assert interface["ipv4"]["requestmode"] == "static"
|
||||
assert interface["ipv4"]["address"] == "192.168.10.4"
|
||||
assert interface["ipv4"]["netmask"] == "255.255.255.0"
|
||||
assert interface["ipv4"]["gateway"] == "192.168.10.1"
|
||||
|
||||
|
||||
def test_supported_adapter_modes(ip, grains, ethercat_installed):
|
||||
"""
|
||||
Test supported adapter modes for each interface
|
||||
"""
|
||||
if grains["lsb_distrib_id"] != "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test is just for older nilrt distros", _use_item_location=True
|
||||
)
|
||||
interface_pattern = re.compile("^eth[0-9]+$")
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == "eth0":
|
||||
assert interface["supported_adapter_modes"] == ["tcpip"]
|
||||
else:
|
||||
assert "tcpip" in interface["supported_adapter_modes"]
|
||||
if not interface_pattern.match(interface["connectionid"]):
|
||||
assert "ethercat" not in interface["supported_adapter_modes"]
|
||||
elif ethercat_installed:
|
||||
assert "ethercat" in interface["supported_adapter_modes"]
|
||||
|
||||
|
||||
def test_ethercat(ip, ethercat_installed):
|
||||
"""
|
||||
Test ip.set_ethercat function
|
||||
"""
|
||||
if not ethercat_installed:
|
||||
raise pytest.skip.Exception(
|
||||
"Test is just for systems with Ethercat", _use_item_location=True
|
||||
)
|
||||
assert ip.set_ethercat(INTERFACE_FOR_TEST, 19)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["adapter_mode"] == "ethercat"
|
||||
assert int(interface["ethercat"]["masterid"]) == 19
|
||||
break
|
||||
assert ip.set_dhcp_linklocal_all(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["adapter_mode"] == "tcpip"
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
break
|
||||
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_dhcp_disable(ip, grains):
|
||||
"""
|
||||
Test cases:
|
||||
- dhcp -> disable
|
||||
- disable -> dhcp
|
||||
"""
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test is just for newer nilrt distros", _use_item_location=True
|
||||
)
|
||||
|
||||
assert ip.set_dhcp_linklocal_all(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
break
|
||||
|
||||
assert ip.disable(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "disabled"
|
||||
break
|
||||
|
||||
assert ip.set_dhcp_linklocal_all(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
break
|
||||
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_dhcp_static(ip, grains):
|
||||
"""
|
||||
Test cases:
|
||||
- dhcp -> static
|
||||
- static -> dhcp
|
||||
"""
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test is just for newer nilrt distros", _use_item_location=True
|
||||
)
|
||||
|
||||
assert ip.set_dhcp_linklocal_all(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
break
|
||||
|
||||
assert ip.set_static_all(
|
||||
INTERFACE_FOR_TEST,
|
||||
"192.168.1.125",
|
||||
"255.255.255.0",
|
||||
"192.168.1.1",
|
||||
"8.8.8.8 8.8.8.4",
|
||||
)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "static"
|
||||
assert interface["ipv4"]["address"] == "192.168.1.125"
|
||||
assert interface["ipv4"]["netmask"] == "255.255.255.0"
|
||||
assert "8.8.8.4" in interface["ipv4"]["dns"]
|
||||
assert "8.8.8.8" in interface["ipv4"]["dns"]
|
||||
break
|
||||
|
||||
assert ip.set_dhcp_linklocal_all(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "dhcp_linklocal"
|
||||
break
|
||||
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
def test_static_disable(ip, grains):
|
||||
"""
|
||||
Test cases:
|
||||
- static -> disable
|
||||
- disable -> static
|
||||
"""
|
||||
if grains["lsb_distrib_id"] == "nilrt":
|
||||
raise pytest.skip.Exception(
|
||||
"Test is just for newer nilrt distros", _use_item_location=True
|
||||
)
|
||||
|
||||
assert ip.set_static_all(
|
||||
INTERFACE_FOR_TEST,
|
||||
"192.168.1.125",
|
||||
"255.255.255.0",
|
||||
"192.168.1.1",
|
||||
"8.8.8.8",
|
||||
)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "static"
|
||||
assert interface["ipv4"]["address"] == "192.168.1.125"
|
||||
assert interface["ipv4"]["netmask"] == "255.255.255.0"
|
||||
assert interface["ipv4"]["dns"] == ["8.8.8.8"]
|
||||
break
|
||||
|
||||
assert ip.disable(INTERFACE_FOR_TEST)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "disabled"
|
||||
break
|
||||
|
||||
assert ip.set_static_all(
|
||||
INTERFACE_FOR_TEST, "192.168.1.125", "255.255.255.0", "192.168.1.1"
|
||||
)
|
||||
info = ip.get_interfaces_details(timeout=300)
|
||||
for interface in info["interfaces"]:
|
||||
if interface["connectionid"] == INTERFACE_FOR_TEST:
|
||||
assert interface["ipv4"]["requestmode"] == "static"
|
||||
assert interface["ipv4"]["address"] == "192.168.1.125"
|
||||
assert interface["ipv4"]["netmask"] == "255.255.255.0"
|
||||
assert interface["ipv4"]["dns"] == []
|
||||
break
|
||||
Loading…
Add table
Reference in a new issue