alex/esp-idf
alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2024-10-05 20:47:46 -04:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'bugfix/esp_eth_test_rx' into 'master'

Browse Source 
component_ut_pytest_esp32_ip101 fix

Closes IDFCI-1313

See merge request espressif/esp-idf!18526
This commit is contained in:
Ondrej Kosta 2022-08-05 14:02:39 +08:00
commit f6b2414bb5
1 changed files with 69 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
components/esp_eth/test_apps/pytest_esp_eth.py
View File

@ -15,73 +15,78 @@ from scapy.all import Ether, raw
ETH_TYPE = 0x2222
@contextlib.contextmanager
def configure_eth_if() -> Iterator[socket.socket]:
# try to determine which interface to use
netifs = os.listdir('/sys/class/net/')
logging.info('detected interfaces: %s', str(netifs))
class EthTestIntf(object):
def __init__(self, eth_type: int):
self.target_if = ''
self.eth_type = eth_type
target_if = ''
for netif in netifs:
if netif.find('eth') == 0 or netif.find('enp') == 0 or netif.find('eno') == 0:
target_if = netif
break
if target_if == '':
raise Exception('no network interface found')
logging.info('Use %s for testing', target_if)
def find_target_if(self) -> None:
# try to determine which interface to use
netifs = os.listdir('/sys/class/net/')
logging.info('detected interfaces: %s', str(netifs))
so = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_TYPE))
so.bind((target_if, 0))
for netif in netifs:
if netif.find('eth') == 0 or netif.find('enp') == 0 or netif.find('eno') == 0:
self.target_if = netif
break
if self.target_if == '':
raise Exception('no network interface found')
logging.info('Use %s for testing', self.target_if)
try:
yield so
finally:
so.close()
@contextlib.contextmanager
def configure_eth_if(self) -> Iterator[socket.socket]:
so = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(self.eth_type))
so.bind((self.target_if, 0))
def send_eth_packet(mac: str) -> None:
with 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=ETH_TYPE) / raw(payload)
try:
so.send(raw(eth_frame))
except Exception as e:
raise e
yield so
finally:
so.close()
def recv_resp_poke(i: int) -> None:
with configure_eth_if() as so:
so.settimeout(10)
try:
eth_frame = Ether(so.recv(60))
if eth_frame.type == ETH_TYPE and eth_frame.load[0] == 0xfa:
if eth_frame.load[1] != i:
raise Exception('Missed Poke Packet')
eth_frame.dst = eth_frame.src
eth_frame.src = so.getsockname()[4]
eth_frame.load = bytes.fromhex('fb') # POKE_RESP code
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
except Exception as e:
raise e
def recv_resp_poke(self, i: int) -> None:
with self.configure_eth_if() as so:
so.settimeout(10)
try:
eth_frame = Ether(so.recv(60))
def traffic_gen(mac: str, pipe_rcv:connection.Connection) -> None:
with 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=ETH_TYPE) / raw(payload)
try:
while pipe_rcv.poll() is not True:
so.send(raw(eth_frame))
except Exception as e:
raise e
if eth_frame.type == self.eth_type and eth_frame.load[0] == 0xfa:
if eth_frame.load[1] != i:
raise Exception('Missed Poke Packet')
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))
except Exception as e:
raise e
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 actual_test(dut: Dut) -> None:
target_if = EthTestIntf(ETH_TYPE)
target_if.find_target_if()
dut.expect_exact('Press ENTER to see the list of tests')
dut.write('\n')
@ -94,7 +99,7 @@ def actual_test(dut: Dut) -> None:
dut.expect_unity_test_output()
dut.expect_exact("Enter next test, or 'enter' to see menu")
with configure_eth_if() as so:
with target_if.configure_eth_if() as so:
so.settimeout(30)
dut.write('"ethernet_broadcast_transmit"')
eth_frame = Ether(so.recv(1024))
@ -109,9 +114,9 @@ def actual_test(dut: Dut) -> None:
r'([\s\S]*)'
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})'
)
send_eth_packet('ff:ff:ff:ff:ff:ff') # broadcast frame
send_eth_packet('01:00:00:00:00:00') # multicast frame
send_eth_packet(res.group(2)) # unicast frame
target_if.send_eth_packet('ff:ff:ff:ff:ff:ff') # broadcast frame
target_if.send_eth_packet('01:00:00:00:00:00') # multicast frame
target_if.send_eth_packet(res.group(2)) # unicast frame
dut.expect_unity_test_output(extra_before=res.group(1))
dut.expect_exact("Enter next test, or 'enter' to see menu")
@ -122,21 +127,21 @@ def actual_test(dut: Dut) -> None:
)
# Start/stop under heavy Tx traffic
for tx_i in range(10):
recv_resp_poke(tx_i)
target_if.recv_resp_poke(tx_i)
# Start/stop under heavy Rx traffic
pipe_rcv, pipe_send = Pipe(False)
tx_proc = Process(target=traffic_gen, args=(res.group(2), pipe_rcv, ))
tx_proc = Process(target=target_if.traffic_gen, args=(res.group(2), pipe_rcv, ))
tx_proc.start()
try:
for rx_i in range(10):
recv_resp_poke(rx_i)
target_if.recv_resp_poke(rx_i)
finally:
pipe_send.send(0)
pipe_send.send(0) # just send some dummy data
tx_proc.join(5)
if tx_proc.exitcode is None:
tx_proc.terminate()
dut.expect_unity_test_output()
dut.expect_unity_test_output(extra_before=res.group(1))
@pytest.mark.esp32