diff --git a/examples/network-test/env.yaml b/examples/network-test/env.yaml new file mode 100644 index 000000000..1f8b9b10f --- /dev/null +++ b/examples/network-test/env.yaml @@ -0,0 +1,14 @@ +targets: + main: + resources: + NetworkService: + address: 192.168.1.5 + username: root + NetworkInterface: + ifname: enp2s0f3 + drivers: + SSHDriver: {} + RawNetworkInterfaceDriver: {} + options: + local_iface_to_dut_iface: + enp2s0f3: uplink diff --git a/examples/network-test/pkg-replay-record.py b/examples/network-test/pkg-replay-record.py new file mode 100755 index 000000000..1333fbaa4 --- /dev/null +++ b/examples/network-test/pkg-replay-record.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +# Generates an Ethernet frame via scapy using pcap, copies pcap to DUT, replays pcap on interface, +# records frame locally (or on exporter, adjust env.yaml accordingly), and compares both. + +import logging +from tempfile import NamedTemporaryFile, TemporaryDirectory + +from labgrid import Environment +from labgrid.logging import basicConfig, StepLogger +from scapy.all import Ether, Raw, rdpcap, wrpcap, conf + +def generate_frame(): + frame = [Ether(dst="11:22:33:44:55:66", src="66:55:44:33:22:11", type=0x9000)] + padding = "\x00" * (conf.min_pkt_size - len(frame)) + frame = frame[0] / Raw(load=padding) + return frame + + +basicConfig(level=logging.DEBUG) +StepLogger.start() +env = Environment("env.yaml") +target = env.get_target() + +netdrv = target.get_driver("RawNetworkInterfaceDriver") +ssh = target.get_driver("SSHDriver") + +# interface names +exporter_iface = netdrv.iface.ifname +dut_iface = env.config.get_target_option(target.name, "local_iface_to_dut_iface")[exporter_iface] + +# generate test frame +generated_frame = generate_frame() + +# write pcap, copy to DUT +remote_pcap = "/tmp/pcap" +with NamedTemporaryFile() as pcap: + wrpcap(pcap.name, generated_frame) + ssh.put(pcap.name, remote_pcap) + +# start record on exporter +ethers = filter(lambda frame: isinstance(frame, Ether), generated_frame) +cap_proc = netdrv.record(count=len(list(ethers))) + +# replay pcap on DUT +ssh.run_check(f"ip link set {dut_iface} up") +ssh.run_check(f"tcpreplay -i {dut_iface} {remote_pcap}") + +# copy recorded pcap from DUT, compare with generated frame +with TemporaryDirectory() as tempdir: + tempf = f"{tempdir}/record.pcap" + netdrv.get_record(cap_proc, tempf) + remote_frame = rdpcap(tempf) + assert remote_frame[0] == generated_frame[0]