Add L1 tests

Signed-off-by: Stepan Vovk <stepan.vovk@plvision.eu>

DentOS_Framework/DentOsTestbed/src/dent_os_testbed/constants.py

@@ -71,6 +71,7 @@
     "suite_functional_ipv4": "Functional IPv4 tests",
     "suite_functional_bridging": "Functional bridging tests",
     "suite_functional_vlan": "VLAN functional tests",
+    "suite_functional_l1": "L1 functional tests",
 }
 
 PYTEST_SUITE_GROUPS = {
@@ -107,5 +108,7 @@
     "suite_group_functional": [
         "suite_functional_vlan",
         "suite_functional_bridging",
-        "suite_functional_ipv4"]
+        "suite_functional_ipv4",
+        "suite_functional_l1"
+    ],
 }

DentOS_Framework/DentOsTestbed/src/dent_os_testbed/test/test_suite/functional/L1/test_l1_autodetect.py

@@ -0,0 +1,153 @@
+import pytest
+import asyncio
+
+from dent_os_testbed.lib.ip.ip_link import IpLink
+from dent_os_testbed.lib.ethtool.ethtool import Ethtool
+
+from dent_os_testbed.utils.test_utils.tgen_utils import (
+    tgen_utils_get_dent_devices_with_tgen,
+    tgen_utils_get_traffic_stats,
+    tgen_utils_setup_streams,
+    tgen_utils_start_traffic,
+    tgen_utils_stop_traffic,
+    tgen_utils_dev_groups_from_config,
+    tgen_utils_traffic_generator_connect,
+    tgen_utils_get_loss,
+    tgen_utils_update_l1_config,
+    tgen_utils_clear_traffic_items
+)
+
+from dent_os_testbed.utils.test_utils.tb_utils import tb_get_qualified_ports
+
+pytestmark = [pytest.mark.suite_functional_l1,
+              pytest.mark.asyncio,
+              pytest.mark.usefixtures("cleanup_bridges", "cleanup_tgen")]
+
+
+@pytest.mark.parametrize("speed , duplex",
+                         [
+                            pytest.param(10, "full"),
+                            pytest.param(10, "half"),
+                            pytest.param(100, "full"),
+                            pytest.param(100, "half"),
+                            pytest.param(1000, "full"),
+                         ])
+async def test_l1_autodetect(testbed, speed, duplex):
+    """
+    Test Name: L1: autodetect
+    Test Suite: suite_functional_l1
+    Test Overview: Verify speed, duplex on ports with configured autodetect
+    Test Procedure:
+    1. Create bridge entity
+    2. Enslave ports to the created bridge entity
+    3. Configure dut ports:
+        speed -> None
+        autoneg -> off
+        advertise -> on
+       Configure ixia ports:
+        speed -> speed
+        autoneg -> on
+    4. Verify port duplex and speed was configured
+    5. Set up traffic according to duplex mode
+    6. Send traffic
+    7. Verify no traffic loss
+    """
+
+    tgen_dev, dent_devices = await tgen_utils_get_dent_devices_with_tgen(testbed, [], 2)
+    if not tgen_dev or not dent_devices:
+        print("The testbed does not have enough dent with tgen connections")
+        return
+    dent_dev = dent_devices[0]
+    device_host_name = dent_dev.host_name
+    tg_ports = tgen_dev.links_dict[device_host_name][0]
+    ports = tgen_dev.links_dict[device_host_name][1][:2]
+    speed_ports = await tb_get_qualified_ports(dent_dev, ports, speed, duplex)
+
+    # 1. Create bridge entity
+    out = await IpLink.add(input_data=[{device_host_name: [{"device": "br0", "type": "bridge"}]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed creating bridge."
+
+    out = await IpLink.set(input_data=[{device_host_name: [{"device": "br0", "operstate": "up"}]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Verify that bridge set to 'UP' state."
+
+    # 2. Enslave ports to the created bridge entity
+    out = await IpLink.set(input_data=[{device_host_name: [{
+        "device": port,
+        "operstate": "up",
+        "master": "br0"
+    } for port in ports]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed setting link to state up."
+
+    # 3. Configure dut ports:
+    out = await Ethtool.set(input_data=[{device_host_name: [{
+        "devname": port,
+        "speed": speed,
+        "autoneg": "off",
+        "duplex": duplex} for port in ports]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed setting port duplex, speed."
+
+    dev_groups = tgen_utils_dev_groups_from_config(
+        [{"ixp": tg_ports[0], "ip": "100.1.1.2", "gw": "100.1.1.6", "plen": 24, },
+         {"ixp": tg_ports[1], "ip": "100.1.1.3", "gw": "100.1.1.6", "plen": 24}])
+    await tgen_utils_traffic_generator_connect(tgen_dev, tg_ports, ports, dev_groups)
+    #  Configure ixia ports
+    await tgen_utils_update_l1_config(tgen_dev, tg_ports, speed=None, autoneg=True, duplex=duplex)
+    await asyncio.sleep(20)  # wait needed in case port was down before
+
+    # 4. Verify port duplex and speed per was configured
+    for port in ports:
+        out = await Ethtool.show(input_data=[{device_host_name: [{"devname": port}]}],  parse_output=True)
+        assert speed == int(out[0][device_host_name]['parsed_output']['speed'][:-4]), "Failed speed test"
+        assert duplex.capitalize() == out[0][device_host_name]['parsed_output']['duplex'], "Failed duplex test"
+
+    supported_speed_ports = [name for name in speed_ports.keys()]
+    first_port_index = ports.index(supported_speed_ports[0])
+    second_port_index = ports.index(supported_speed_ports[1])
+
+    traffic_loss = {}
+    streams = {
+        f"{tg_ports[first_port_index]} --> {tg_ports[second_port_index]}": {
+            "type": "ethernet",
+            "ep_source": ports[first_port_index],
+            "ep_destination": ports[second_port_index],
+            "frame_rate_type": "line_rate",
+            "rate": 100
+        },
+        f"{tg_ports[second_port_index]} --> {tg_ports[first_port_index]}": {
+            "type": "ethernet",
+            "ep_source": ports[second_port_index],
+            "ep_destination": ports[first_port_index],
+            "frame_rate_type": "line_rate",
+            "rate": 100
+        }
+    }
+
+    # 5. Set up traffic according to duplex mode
+    if duplex.capitalize() == "Half":
+        for stream_name, stream in streams.items():
+
+            await tgen_utils_setup_streams(tgen_dev, config_file_name=None, streams={stream_name: stream})
+            # 6. Send traffic to rx_ports
+            await tgen_utils_start_traffic(tgen_dev)
+            await asyncio.sleep(1)
+            await tgen_utils_stop_traffic(tgen_dev)
+
+            stats = await tgen_utils_get_traffic_stats(tgen_dev, "Flow Statistics")
+            for row in stats.Rows:
+                traffic_loss[stream_name] = (tgen_utils_get_loss(row))
+            await tgen_utils_clear_traffic_items(tgen_dev)
+    else:
+        await tgen_utils_setup_streams(tgen_dev, config_file_name=None, streams=streams)
+
+        # 6. Send traffic to rx_ports
+        await tgen_utils_start_traffic(tgen_dev)
+        await asyncio.sleep(1)
+        await tgen_utils_stop_traffic(tgen_dev)
+
+        stats = await tgen_utils_get_traffic_stats(tgen_dev, "Flow Statistics")
+        for row in stats.Rows:
+            traffic_loss[row["Traffic Item"]] = tgen_utils_get_loss(row)
+
+    # 7. Verify no traffic loss on rx ports
+    for stream, loss in traffic_loss.items():
+        assert loss == 0.000, f'No traffic for traffic item {stream}: expected 0.000 got : {loss}'

DentOS_Framework/DentOsTestbed/src/dent_os_testbed/test/test_suite/functional/L1/test_l1_autoneg.py

@@ -0,0 +1,164 @@
+import pytest
+import asyncio
+
+from dent_os_testbed.lib.ip.ip_link import IpLink
+from dent_os_testbed.lib.ethtool.ethtool import Ethtool
+
+from dent_os_testbed.utils.test_utils.tgen_utils import (
+    tgen_utils_get_dent_devices_with_tgen,
+    tgen_utils_get_traffic_stats,
+    tgen_utils_setup_streams,
+    tgen_utils_start_traffic,
+    tgen_utils_stop_traffic,
+    tgen_utils_dev_groups_from_config,
+    tgen_utils_traffic_generator_connect,
+    tgen_utils_get_loss,
+    tgen_utils_update_l1_config,
+    tgen_utils_clear_traffic_items
+)
+
+from dent_os_testbed.utils.test_utils.tb_utils import tb_get_qualified_ports
+
+pytestmark = [pytest.mark.suite_functional_l1,
+              pytest.mark.asyncio,
+              pytest.mark.usefixtures("cleanup_bridges", "cleanup_tgen")]
+
+
+@pytest.mark.parametrize("speed , duplex",
+                         [
+                            pytest.param(10, "full"),
+                            pytest.param(10, "half"),
+                            pytest.param(100, "full"),
+                            pytest.param(100, "half"),
+                            pytest.param(1000, "full"),
+                         ])
+async def test_l1_autoneg(testbed, speed, duplex):
+    """
+    Test Name: L1: autoneg on both ports
+    Test Suite: suite_functional_l1
+    Test Overview: Verify speed, duplex on ports with configured autoneg
+    Test Procedure:
+    1. Create bridge entity
+    2. Enslave ports to the created bridge entity
+    3. Configure dut ports:
+        speed -> speed
+        autoneg -> on
+        advertise -> on
+       Configure ixia ports:
+        speed -> none
+        autoneg -> on
+    4. Verify port duplex and speed configured
+    5. Set up traffic according to duplex mode
+    6. Send traffic
+    7. Verify no traffic loss
+    """
+
+    tgen_dev, dent_devices = await tgen_utils_get_dent_devices_with_tgen(testbed, [], 2)
+    if not tgen_dev or not dent_devices:
+        print("The testbed does not have enough dent with tgen connections")
+        return
+    dent_dev = dent_devices[0]
+    device_host_name = dent_dev.host_name
+    tg_ports = tgen_dev.links_dict[device_host_name][0]
+    ports = tgen_dev.links_dict[device_host_name][1][:2]
+    advertise_map = {
+        '10_half': '0x001',
+        '10_full': '0x002',
+        '100_half': '0x004',
+        '100_full': '0x008',
+        '1000_full': '0x020',
+        '10000_full': '0x1000'
+    }
+    speed_ports = await tb_get_qualified_ports(dent_dev, ports, speed, duplex)
+
+    # 1. Create bridge entity
+    out = await IpLink.add(input_data=[{device_host_name: [{"device": "br0", "type": "bridge"}]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed creating bridge."
+
+    out = await IpLink.set(input_data=[{device_host_name: [{"device": "br0", "operstate": "up"}]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Verify that bridge set to 'UP' state."
+
+    # 2. Enslave ports to the created bridge entity
+    out = await IpLink.set(input_data=[{device_host_name: [{
+        "device": port,
+        "operstate": "up",
+        "master": "br0"
+    } for port in ports]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed setting link to state up."
+
+    # 3.  Configure dut ports
+    out = await Ethtool.set(input_data=[{device_host_name: [{
+        "devname": port,
+        "speed": speed,
+        "autoneg": "on",
+        "advertise": advertise_map[f"{speed}_{duplex}"],
+        "duplex": duplex} for port in ports]}])
+    assert out[0][device_host_name]["rc"] == 0, f"Failed setting port duplex, speed."
+
+    dev_groups = tgen_utils_dev_groups_from_config(
+        [{"ixp": tg_ports[0], "ip": "100.1.1.2", "gw": "100.1.1.6", "plen": 24, },
+         {"ixp": tg_ports[1], "ip": "100.1.1.3", "gw": "100.1.1.6", "plen": 24}])
+    await tgen_utils_traffic_generator_connect(tgen_dev, tg_ports, ports, dev_groups)
+
+    #   Configure ixia ports
+    await tgen_utils_update_l1_config(tgen_dev, tg_ports, speed=None, autoneg=True, duplex=duplex)
+    await asyncio.sleep(20)  # wait needed in case port was down before
+
+    # 4. Verify port duplex and speed was configured correctly
+    for port in ports:
+        out = await Ethtool.show(input_data=[{device_host_name: [{"devname": port}]}],  parse_output=True)
+        assert speed == int(out[0][device_host_name]['parsed_output']['speed'][:-4]), "Failed speed test"
+        assert duplex.capitalize() == out[0][device_host_name]['parsed_output']['duplex'], "Failed duplex test"
+
+    # 5. Set up traffic according to duplex mode
+    supported_speed_ports = [name for name in speed_ports.keys()]
+    first_port_index = ports.index(supported_speed_ports[0])
+    second_port_index = ports.index(supported_speed_ports[1])
+
+    traffic_loss = {}
+    streams = {
+        f"{tg_ports[first_port_index]} --> {tg_ports[second_port_index]}": {
+            "type": "ethernet",
+            "ep_source": ports[first_port_index],
+            "ep_destination": ports[second_port_index],
+            "frame_rate_type": "line_rate",
+            "rate": 100
+        },
+        f"{tg_ports[second_port_index]} --> {tg_ports[first_port_index]}": {
+            "type": "ethernet",
+            "ep_source": ports[second_port_index],
+            "ep_destination": ports[first_port_index],
+            "frame_rate_type": "line_rate",
+            "rate": 100
+        }
+    }
+
+    # 5. Set up traffic according to duplex mode
+    if duplex.capitalize() == "Half":
+        for stream_name, stream in streams.items():
+
+            await tgen_utils_setup_streams(tgen_dev, config_file_name=None, streams={stream_name: stream})
+            # 6. Send traffic to rx_ports
+            await tgen_utils_start_traffic(tgen_dev)
+            await asyncio.sleep(1)
+            await tgen_utils_stop_traffic(tgen_dev)
+
+            stats = await tgen_utils_get_traffic_stats(tgen_dev, "Flow Statistics")
+            for row in stats.Rows:
+                traffic_loss[stream_name] = (tgen_utils_get_loss(row))
+            await tgen_utils_clear_traffic_items(tgen_dev)
+    else:
+        await tgen_utils_setup_streams(tgen_dev, config_file_name=None, streams=streams)
+
+        # 6. Send traffic to rx_ports
+        await tgen_utils_start_traffic(tgen_dev)
+        await asyncio.sleep(1)
+        await tgen_utils_stop_traffic(tgen_dev)
+
+        stats = await tgen_utils_get_traffic_stats(tgen_dev, "Flow Statistics")
+        for row in stats.Rows:
+            traffic_loss[row["Traffic Item"]] = tgen_utils_get_loss(row)
+
+    # 7. Verify no traffic loss on rx ports
+    for stream, loss in traffic_loss.items():
+        assert loss == 0.000, f'No traffic for traffic item {stream}: expected 0.000 got : {loss}'