-
Notifications
You must be signed in to change notification settings - Fork 7k
/
pytest_esp_eth.py
191 lines (165 loc) · 6.98 KB
/
pytest_esp_eth.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
# SPDX-FileCopyrightText: 2022-2023 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0
import contextlib
import logging
import os
import socket
from multiprocessing import Pipe, Process, connection
from typing import Iterator
import pytest
from pytest_embedded import Dut
from scapy.all import Ether, raw
ETH_TYPE = 0x3300
class EthTestIntf(object):
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))
for netif in netifs:
# if no interface defined, try to find it automatically
if my_if == '':
if netif.find('eth') == 0 or netif.find('enp') == 0 or netif.find('eno') == 0:
self.target_if = netif
break
else:
if netif.find(my_if) == 0:
self.target_if = my_if
break
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))
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 actual_test(dut: Dut) -> None:
target_if = EthTestIntf(ETH_TYPE)
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('"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')
# 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)
target_if.recv_resp_poke(mac=dut_mac)
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(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('"start_stop_stress_test"')
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()
dut.expect_exact('Ethernet stopped')
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(extra_before=res.group(1))
@pytest.mark.esp32
@pytest.mark.eth_ip101
@pytest.mark.parametrize('config', [
'ip101',
], indirect=True)
def test_esp_eth_ip101(dut: Dut) -> None:
actual_test(dut)
@pytest.mark.esp32
@pytest.mark.eth_lan8720
@pytest.mark.parametrize('config', [
'lan8720',
], indirect=True)
def test_esp_eth_lan8720(dut: Dut) -> None:
actual_test(dut)