network: add tcpdump() function and use it

test_vlan_filtering.py is running tcpdump for various tests.
as it something we want to use more in network tests, create
a function.

Signed-off-by: Sebastien Rodot <sebastien.rodot@vates.tech>

Author: semarie

tests/network/__init__.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import logging
+
+from lib.common import wait_for
+from lib.vm import VM
+
+from typing import Callable
+
+def tcpdump(
+    pcapfile: str,
+    vm: VM,
+    interface: str,
+    filter: str | None = None,
+    *,
+    count: int | None = None,
+) -> Callable[[], None]:
+    """
+    Run tcpdump on the vm with specific interface.
+    Returns a lambda to permit to cleanup (kill tcpdump + file cleanup).
+    """
+    def cleanup():
+        # if vm is gone, do not fail
+        if vm.exists() and vm.is_running():
+            vm.ssh(
+                f"xargs kill < {pcapfile}.pid || true; "
+                f"rm -f -- {pcapfile}.pid; ",
+                check=False,
+            )
+
+    # check no concurrent run
+    assert not vm.file_exists(f"{pcapfile}.pid")
+
+    # tcpdump arguments
+    args = f"-n -w {pcapfile} -i {interface}"
+    if count is not None:
+        args += f" -c {count}"
+    if filter is not None:
+        args += f" '{filter}'"
+
+    # run tcpdump in background
+    logging.info(f"Running tcpdump on '{vm.name()}'")
+    vm.ssh(
+        f"tcpdump --immediate-mode {args} & "
+        f"echo $! > {pcapfile}.pid; ",
+        background=True,
+    )
+
+    # wait tcpdump to start properly
+    wait_for(lambda: vm.file_exists(pcapfile), timeout_secs=2)
+
+    return cleanup

tests/network/test_vlan_filtering.py

@@ -6,6 +6,8 @@
 from lib.network import Network
 from lib.vm import VM
 
+from . import tcpdump
+
 # Requirements:
 # From --hosts parameter:
 # - host(A1): an XCP-ng host
@@ -59,18 +61,15 @@ def test_passing(self, defer: Defer, empty_network: Network, imported_vm: VM, vm
         vm_writer.ssh(f"ip addr add 192.168.42.1/24 dev {ifaceW}.42")
         vm_writer.ssh(f"ip link set {ifaceW}.42 up")
 
+        tcpdump_stop = tcpdump("tcpdump.pcap", vm_reader, ifaceR, "vlan 42 and arp", count=1)
+        defer(tcpdump_stop)
+
         # send some packets on VLAN 42 (ARP packets will be send)
-        # the ping process could be still running after the test,
-        # but vm_writer will be destroyed, so it isn't a problem.
-        vm_writer.ssh("ping -c1 -w1 192.168.42.2", background=True, check=False)
-
-        # check packets are seen on vm_reader
-        # fail if /tmp/out is empty
-        vm_reader.ssh(
-            f"tcpdump -i {ifaceR} -w /tmp/out -c1 -n 'vlan 42 and arp' &"
-            "pid=$! ; sleep 5 ; kill $pid ;"
-            "test -s /tmp/out"
-        )
+        vm_writer.ssh("ping -c1 -w1 192.168.42.2", check=False)
+        tcpdump_stop()
+
+        # check packets are seen on vm_reader (trunks=42)
+        vm_reader.ssh("test $(tcpdump -n -r tcpdump.pcap | wc -l) -gt 0")
 
     def test_filtered(self, defer: Defer, empty_network: Network, imported_vm: VM, vm_with_tcpdump_scope_function: VM):
         vm_writer, ifaceW = start_vm_on_trunk(
@@ -95,15 +94,12 @@ def test_filtered(self, defer: Defer, empty_network: Network, imported_vm: VM, v
         vm_writer.ssh(f"ip addr add 192.168.42.1/24 dev {ifaceW}.42")
         vm_writer.ssh(f"ip link set {ifaceW}.42 up")
 
+        tcpdump_stop = tcpdump("tcpdump.pcap", vm_reader, ifaceR, "vlan 42 and arp", count=1)
+        defer(tcpdump_stop)
+
         # send some packets on VLAN 42 (ARP packets will be send)
-        # the ping process could be still running after the test,
-        # but vm_writer will be destroyed, so it isn't a problem.
-        vm_writer.ssh("ping -c1 -w1 192.168.42.2", background=True, check=False)
-
-        # check packets are seen on vm_reader
-        # fail if /tmp/out is not empty
-        vm_reader.ssh(
-            f"tcpdump -i {ifaceR} -w /tmp/out -c1 -n 'vlan 42 and arp' &"
-            "pid=$! ; sleep 5 ; kill $pid ;"
-            "test ! -s /tmp/out"
-        )
+        vm_writer.ssh("ping -c1 -w1 192.168.42.2", check=False)
+        tcpdump_stop()
+
+        # check packets are *NOT* seen on vm_reader (trunks=52)
+        vm_reader.ssh("test $(tcpdump -n -r tcpdump.pcap | wc -l) -eq 0")