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,56 @@
+from __future__ import annotations
+
+import pytest
+
+import logging
+from contextlib import contextmanager
+
+from lib.common import Defer, wait_for
+from lib.vm import VM
+
+from typing import Generator
+
+@contextmanager
+def tcpdump(
+    defer: Defer,
+    pcapfile: str,
+    vm: VM,
+    interface: str,
+    filter: str | None = None,
+    *,
+    count: int | None = None,
+) -> Generator[None, None, None]:
+    """
+    Run tcpdump on the vm with specific interface.
+    tcpdump is running only in the returned context.
+    """
+    def cleanup():
+        # if vm is gone, do not fail
+        if vm.exists() and vm.is_running():
+            vm.ssh("xargs kill < {pcapfile}.pid; rm -f -- {pcapfile}.pid")
+
+    # check no concurrent run
+    if vm.file_exists(f"{pcapfile}.pid"):
+        pytest.fail(f"concurrent run detected: {pcapfile}.pid already exists")
+
+    # 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,
+    )
+    defer(cleanup)
+
+    # check tcpdump has properly started
+    wait_for(lambda: vm.file_exists(pcapfile), timeout_secs=2, retry_delay_secs=1)
+
+    yield
+    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,12 @@ 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")
 
-        # 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"
-        )
+        with tcpdump(defer, "tcpdump.pcap", vm_reader, ifaceR, "vlan 42 and arp", count=1):
+            # send some packets on VLAN 42 (ARP packets will be send)
+            vm_writer.ssh("ping -c1 -w1 192.168.42.2", check=False)
+
+        # 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 +91,9 @@ 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")
 
-        # 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"
-        )
+        with tcpdump(defer, "tcpdump.pcap", vm_reader, ifaceR, "vlan 42 and arp", count=1):
+            # send some packets on VLAN 42 (ARP packets will be send)
+            vm_writer.ssh("ping -c1 -w1 192.168.42.2", check=False)
+
+        # check packets are *NOT* seen on vm_reader (trunks=52)
+        vm_reader.ssh("test $(tcpdump -n -r tcpdump.pcap | wc -l) -eq 0")