diff --git a/components/esp_eth/test_apps/pytest_esp_eth.py b/components/esp_eth/test_apps/pytest_esp_eth.py index aba0ead6cf..b7d6971e99 100644 --- a/components/esp_eth/test_apps/pytest_esp_eth.py +++ b/components/esp_eth/test_apps/pytest_esp_eth.py @@ -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