mirror of
https://github.com/espressif/esp-idf.git
synced 2026-10-01 10:40:47 +03:00
feat(esp_eth): update test_apps to use Ethernet test component
Move test logic into the eth_test_app managed component and slim down the test_apps main target and pytest suite accordingly.
This commit is contained in:
Ondrej Kosta
@@ -1,254 +1,16 @@
|
||||
# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
|
||||
# SPDX-License-Identifier: CC0-1.0
|
||||
import contextlib
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
from collections.abc import Iterator
|
||||
from multiprocessing import Pipe
|
||||
from multiprocessing import Process
|
||||
from multiprocessing import connection
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from pytest_embedded_idf import IdfDut
|
||||
from pytest_embedded_idf.utils import idf_parametrize
|
||||
from scapy.all import Ether
|
||||
from scapy.all import raw
|
||||
|
||||
ETH_TYPE = 0x3300
|
||||
|
||||
|
||||
class EthTestIntf:
|
||||
def __init__(self, eth_type: int, my_if: str = ''):
|
||||
self.target_if = ''
|
||||
self.eth_type = eth_type
|
||||
self.find_target_if(my_if)
|
||||
|
||||
def find_target_if(self, my_if: str = '') -> None:
|
||||
# try to determine which interface to use
|
||||
netifs = os.listdir('/sys/class/net/')
|
||||
# order matters - ETH NIC with the highest number is connected to DUT on CI runner
|
||||
netifs.sort(reverse=True)
|
||||
logging.info('detected interfaces: %s', str(netifs))
|
||||
|
||||
if my_if == '':
|
||||
if 'dut_p1' in netifs:
|
||||
self.target_if = 'dut_p1'
|
||||
else:
|
||||
for netif in netifs:
|
||||
# if no interface defined, try to find it automatically
|
||||
if netif.find('eth') == 0 or netif.find('enp') == 0 or netif.find('eno') == 0:
|
||||
self.target_if = netif
|
||||
break
|
||||
elif my_if in netifs:
|
||||
self.target_if = my_if
|
||||
|
||||
if self.target_if == '':
|
||||
raise RuntimeError('network interface not found')
|
||||
logging.info('Use %s for testing', self.target_if)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def configure_eth_if(self, eth_type: int = 0) -> Iterator[socket.socket]:
|
||||
if eth_type == 0:
|
||||
eth_type = self.eth_type
|
||||
so = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(eth_type)) # type: ignore
|
||||
so.bind((self.target_if, 0))
|
||||
try:
|
||||
yield so
|
||||
finally:
|
||||
so.close()
|
||||
|
||||
def send_eth_packet(self, mac: str) -> None:
|
||||
with self.configure_eth_if() as so:
|
||||
so.settimeout(10)
|
||||
payload = bytearray(1010)
|
||||
for i, _ in enumerate(payload):
|
||||
payload[i] = i & 0xFF
|
||||
eth_frame = Ether(dst=mac, src=so.getsockname()[4], type=self.eth_type) / raw(payload)
|
||||
try:
|
||||
so.send(raw(eth_frame))
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
def recv_resp_poke(self, mac: str, i: int = 0) -> None:
|
||||
eth_type_ctrl = self.eth_type + 1
|
||||
with self.configure_eth_if(eth_type_ctrl) as so:
|
||||
so.settimeout(30)
|
||||
for _ in range(10):
|
||||
try:
|
||||
eth_frame = Ether(so.recv(60))
|
||||
except Exception as e:
|
||||
raise e
|
||||
if mac == eth_frame.src and eth_frame.load[0] == 0xFA:
|
||||
if eth_frame.load[1] != i:
|
||||
raise RuntimeError('Missed Poke Packet')
|
||||
logging.info('Poke Packet received...')
|
||||
eth_frame.dst = eth_frame.src
|
||||
eth_frame.src = so.getsockname()[4]
|
||||
eth_frame.load = bytes.fromhex('fb') # POKE_RESP code
|
||||
so.send(raw(eth_frame))
|
||||
break
|
||||
else:
|
||||
logging.warning('Unexpected Control packet')
|
||||
logging.warning('Expected Ctrl command: 0xfa, actual: 0x%x', eth_frame.load[0])
|
||||
logging.warning('Source MAC %s', eth_frame.src)
|
||||
else:
|
||||
raise RuntimeError('No Poke Packet!')
|
||||
|
||||
def traffic_gen(self, mac: str, pipe_rcv: connection.Connection) -> None:
|
||||
with self.configure_eth_if() as so:
|
||||
payload = bytes.fromhex('ff') # DUMMY_TRAFFIC code
|
||||
payload += bytes(1485)
|
||||
eth_frame = Ether(dst=mac, src=so.getsockname()[4], type=self.eth_type) / raw(payload)
|
||||
try:
|
||||
while pipe_rcv.poll() is not True:
|
||||
so.send(raw(eth_frame))
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
def eth_loopback(self, mac: str, pipe_rcv: connection.Connection) -> None:
|
||||
with self.configure_eth_if(self.eth_type) as so:
|
||||
so.settimeout(30)
|
||||
try:
|
||||
while pipe_rcv.poll() is not True:
|
||||
try:
|
||||
eth_frame = Ether(so.recv(1522))
|
||||
except Exception as e:
|
||||
raise e
|
||||
if mac == eth_frame.src:
|
||||
eth_frame.dst = eth_frame.src
|
||||
eth_frame.src = so.getsockname()[4]
|
||||
so.send(raw(eth_frame))
|
||||
else:
|
||||
logging.warning('Received frame from unexpected source')
|
||||
logging.warning('Source MAC %s', eth_frame.src)
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
|
||||
def ethernet_test(dut: IdfDut) -> None:
|
||||
dut.run_all_single_board_cases(group='ethernet', timeout=980)
|
||||
|
||||
|
||||
def ethernet_int_emac_test(dut: IdfDut) -> None:
|
||||
dut.run_all_single_board_cases(group='esp_emac', timeout=240)
|
||||
|
||||
|
||||
def ethernet_l2_test(dut: IdfDut, test_if: str = '') -> None:
|
||||
target_if = EthTestIntf(ETH_TYPE, test_if)
|
||||
|
||||
dut.expect_exact('Press ENTER to see the list of tests')
|
||||
dut.write('\n')
|
||||
dut.expect_exact('Enter test for running.')
|
||||
|
||||
with target_if.configure_eth_if() as so:
|
||||
so.settimeout(30)
|
||||
dut.write('"ethernet broadcast transmit"')
|
||||
res = dut.expect(
|
||||
r'DUT MAC: ([0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2})'
|
||||
)
|
||||
|
||||
# wait for POKE msg to be sure the switch already started forwarding the port's traffic
|
||||
# (there might be slight delay due to the RSTP execution)
|
||||
dut_mac = res.group(1).decode('utf-8')
|
||||
target_if.recv_resp_poke(mac=dut_mac)
|
||||
|
||||
for _ in range(10):
|
||||
eth_frame = Ether(so.recv(1024))
|
||||
if dut_mac == eth_frame.src:
|
||||
break
|
||||
else:
|
||||
raise RuntimeError('No broadcast received from expected DUT MAC addr')
|
||||
|
||||
for i in range(0, 1010):
|
||||
if eth_frame.load[i] != i & 0xFF:
|
||||
raise RuntimeError('Packet content mismatch')
|
||||
dut.expect_unity_test_output()
|
||||
|
||||
dut.expect_exact("Enter next test, or 'enter' to see menu")
|
||||
dut.write('"ethernet recv_pkt"')
|
||||
res = dut.expect(
|
||||
r'DUT MAC: ([0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2})'
|
||||
)
|
||||
dut_mac = res.group(1).decode('utf-8')
|
||||
target_if.recv_resp_poke(mac=dut_mac)
|
||||
for _ in range(5):
|
||||
# wait to be sure the DUT reconfigured the filter
|
||||
dut.expect_exact('Filter configured')
|
||||
target_if.send_eth_packet('ff:ff:ff:ff:ff:ff') # broadcast frame
|
||||
target_if.send_eth_packet('01:00:5e:00:00:00') # IPv4 multicast frame
|
||||
target_if.send_eth_packet('33:33:00:00:00:00') # IPv6 multicast frame
|
||||
target_if.send_eth_packet(mac=dut_mac) # unicast frame
|
||||
|
||||
dut.expect_unity_test_output(extra_before=res.group(1))
|
||||
|
||||
dut.expect_exact("Enter next test, or 'enter' to see menu")
|
||||
dut.write('"ethernet start/stop stress test under heavy traffic"')
|
||||
res = dut.expect(
|
||||
r'DUT MAC: ([0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2})'
|
||||
)
|
||||
dut_mac = res.group(1).decode('utf-8')
|
||||
# Start/stop under heavy Tx traffic
|
||||
for tx_i in range(10):
|
||||
target_if.recv_resp_poke(dut_mac, tx_i)
|
||||
dut.expect_exact('Ethernet Stopped')
|
||||
|
||||
for rx_i in range(10):
|
||||
target_if.recv_resp_poke(dut_mac, rx_i)
|
||||
# Start/stop under heavy Rx traffic
|
||||
pipe_rcv, pipe_send = Pipe(False)
|
||||
tx_proc = Process(
|
||||
target=target_if.traffic_gen,
|
||||
args=(
|
||||
dut_mac,
|
||||
pipe_rcv,
|
||||
),
|
||||
)
|
||||
tx_proc.start()
|
||||
try:
|
||||
dut.expect_exact('Ethernet Stopped')
|
||||
finally:
|
||||
pipe_send.send(0) # just send some dummy data to stop traffic generation
|
||||
tx_proc.join(5)
|
||||
if tx_proc.exitcode is None:
|
||||
tx_proc.terminate()
|
||||
|
||||
dut.expect_unity_test_output(extra_before=res.group(1))
|
||||
|
||||
|
||||
def ethernet_heap_alloc_test(dut: IdfDut, test_if: str = '') -> None:
|
||||
target_if = EthTestIntf(ETH_TYPE, test_if)
|
||||
dut.expect_exact('Press ENTER to see the list of tests')
|
||||
dut.write('\n')
|
||||
dut.expect_exact('Enter test for running.')
|
||||
dut.write('"heap utilization"')
|
||||
res = dut.expect(r'PHY loopback is (enabled|disabled)')
|
||||
phy_loopback = res.group(1).decode('utf-8')
|
||||
# Some chips do not support internal loopback, we need to loopback at test PC side
|
||||
if phy_loopback == 'disabled':
|
||||
logging.info('Starting loopback server...')
|
||||
res = dut.expect(
|
||||
r'DUT MAC: ([0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2})'
|
||||
)
|
||||
dut_mac = res.group(1).decode('utf-8')
|
||||
pipe_rcv, pipe_send = Pipe(False)
|
||||
loopback_proc = Process(
|
||||
target=target_if.eth_loopback,
|
||||
args=(
|
||||
dut_mac,
|
||||
pipe_rcv,
|
||||
),
|
||||
)
|
||||
loopback_proc.start()
|
||||
|
||||
target_if.recv_resp_poke(mac=dut_mac)
|
||||
dut.expect_exact('Ethernet Stopped')
|
||||
pipe_send.send(0) # just send some dummy data
|
||||
loopback_proc.join(5)
|
||||
if loopback_proc.exitcode is None:
|
||||
loopback_proc.terminate()
|
||||
|
||||
dut.expect_unity_test_output()
|
||||
# Import for type-checking only — at runtime eth_test_runner pulls in dependencies that
|
||||
# are not available during pytest collection. The real import is deferred to the fixture body.
|
||||
if TYPE_CHECKING:
|
||||
from conftest import EthTestRunner
|
||||
|
||||
|
||||
# ----------- IP101 -----------
|
||||
@@ -256,8 +18,8 @@ def ethernet_heap_alloc_test(dut: IdfDut, test_if: str = '') -> None:
|
||||
@pytest.mark.parametrize('config', ['default_generic', 'release_generic', 'single_core_generic'], indirect=True)
|
||||
@pytest.mark.flaky(reruns=3, reruns_delay=5)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_ethernet(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_ethernet(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
|
||||
|
||||
@pytest.mark.eth_ip101
|
||||
@@ -269,10 +31,10 @@ def test_esp_ethernet(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_emac(dut: IdfDut) -> None:
|
||||
ethernet_int_emac_test(dut)
|
||||
def test_esp_emac(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
dut.run_all_single_board_cases(group='esp_emac', timeout=240)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_heap_alloc_test(dut)
|
||||
eth_test_runner.run_ethernet_heap_alloc_test(dut)
|
||||
|
||||
|
||||
@pytest.mark.eth_ip101
|
||||
@@ -284,8 +46,8 @@ def test_esp_emac(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_ip101(dut: IdfDut) -> None:
|
||||
ethernet_l2_test(dut)
|
||||
def test_esp_eth_ip101(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
@pytest.mark.eth_ip101
|
||||
@@ -310,10 +72,10 @@ def test_esp32_emac_clko(dut: IdfDut) -> None:
|
||||
],
|
||||
indirect=['target'],
|
||||
)
|
||||
def test_esp32p4_ethernet(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp32p4_ethernet(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -324,10 +86,10 @@ def test_esp32p4_ethernet(dut: IdfDut) -> None:
|
||||
],
|
||||
indirect=['target'],
|
||||
)
|
||||
def test_esp32p4_emac(dut: IdfDut) -> None:
|
||||
ethernet_int_emac_test(dut)
|
||||
def test_esp32p4_emac(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
dut.run_all_single_board_cases(group='esp_emac', timeout=240)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_heap_alloc_test(dut)
|
||||
eth_test_runner.run_ethernet_heap_alloc_test(dut)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -352,10 +114,10 @@ def test_esp32p4_emac_clko(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_lan8720(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_lan8720(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
# ----------- RTL8201 -----------
|
||||
@@ -368,10 +130,10 @@ def test_esp_eth_lan8720(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_rtl8201(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_rtl8201(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
# ----------- KSZ8041 -----------
|
||||
@@ -384,10 +146,10 @@ def test_esp_eth_rtl8201(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_ksz8041(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_ksz8041(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
# ----------- DP83848 -----------
|
||||
@@ -400,10 +162,10 @@ def test_esp_eth_ksz8041(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_dp83848(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_dp83848(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
|
||||
|
||||
# ----------- W5500 -----------
|
||||
@@ -416,12 +178,12 @@ def test_esp_eth_dp83848(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_w5500(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_w5500(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_heap_alloc_test(dut)
|
||||
eth_test_runner.run_ethernet_heap_alloc_test(dut)
|
||||
|
||||
|
||||
# ----------- KSZ8851SNL -----------
|
||||
@@ -434,12 +196,12 @@ def test_esp_eth_w5500(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_ksz8851snl(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_ksz8851snl(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_heap_alloc_test(dut)
|
||||
eth_test_runner.run_ethernet_heap_alloc_test(dut)
|
||||
|
||||
|
||||
# ----------- DM9051 -----------
|
||||
@@ -452,12 +214,12 @@ def test_esp_eth_ksz8851snl(dut: IdfDut) -> None:
|
||||
indirect=True,
|
||||
)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_esp_eth_dm9051(dut: IdfDut) -> None:
|
||||
ethernet_test(dut)
|
||||
def test_esp_eth_dm9051(dut: IdfDut, eth_test_runner: 'EthTestRunner') -> None:
|
||||
eth_test_runner.run_ethernet_test_apps(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_l2_test(dut)
|
||||
eth_test_runner.run_ethernet_l2_test(dut)
|
||||
dut.serial.hard_reset()
|
||||
ethernet_heap_alloc_test(dut)
|
||||
eth_test_runner.run_ethernet_heap_alloc_test(dut)
|
||||
|
||||
|
||||
# ----------- EMAC Sleep Retention -----------
|
||||
|
||||
Reference in New Issue
Block a user