tests/dhclient: import scapy inside vnet

This fixes a [CI failure][0].

[0] https://ci.freebsd.org/job/FreeBSD-main-amd64-test/29813/testReport/sbin.dhclient.functional/py/__test_cases_list__/

PR:		299641
Reviewed by:	markj, ngie
MFC after:	3 days
Sponsored by:	The FreeBSD Foundation
Pull Request:	https://ron-dev.freebsd.org/FreeBSD/src/pulls/83

sbin/dhclient/tests/functional.py

@@ -10,7 +10,6 @@
 import subprocess
 
 import pytest
-import scapy.all as sp
 from atf_python.sys.net.tools import ToolsHelper
 from atf_python.sys.net.vnet import VnetTestTemplate
 
@@ -21,44 +20,6 @@ def mac_to_chaddr(mac_str):
     return mac_bytes + b"\x00" * (16 - len(mac_bytes))
 
 
-def get_dhcp_type(pkt):
-    """Extract the DHCP message type from a packet."""
-    if not pkt.haslayer(sp.DHCP):
-        return None
-    for opt in pkt[sp.DHCP].options:
-        if isinstance(opt, tuple) and opt[0] == "message-type":
-            return sp.DHCPTypes[opt[1]]
-    return None
-
-
-def build_reply(iface, server_ip, client_ip, client_mac, xid, msg_type,
-                extra_opts=None, sname=b"", file=b""):
-    """Build a DHCP OFFER or ACK packet."""
-    options = [
-        ("message-type", msg_type),
-        ("server_id", server_ip),
-        ("lease_time", 3600),
-        ("subnet_mask", "255.255.255.0"),
-        ("router", server_ip),
-    ]
-    if extra_opts:
-        options.extend(extra_opts)
-    options.append("end")
-
-    return (
-        sp.Ether(dst=client_mac, src=sp.get_if_hwaddr(iface))
-        / sp.IP(src=server_ip, dst="255.255.255.255")
-        / sp.UDP(sport=67, dport=68)
-        / sp.BOOTP(op=2,
-            xid=xid,
-            yiaddr=client_ip,
-            siaddr=server_ip,
-            chaddr=mac_to_chaddr(client_mac),
-            sname=sname,
-            file=file,
-        )
-        / sp.DHCP(options=options)
-    )
 
 
 class TestDhclient(VnetTestTemplate):
@@ -106,6 +67,9 @@ class TestDhclient(VnetTestTemplate):
         ToolsHelper.print_output(
             "/sbin/ifconfig {} inet {}/24 up".format(ifname, self._SERVER_IP)
         )
+        # Import in the correct vnet, so as to not confuse Scapy
+        import scapy.all as sp
+        self.sp = sp
         return ifname
 
     def _do_dora(self, ifname, first_extra_opts=None, extra_opts=None,
@@ -123,33 +87,33 @@ class TestDhclient(VnetTestTemplate):
         def handle_dhcp(pkt):
             nonlocal discover_count, done
 
-            msg_type = get_dhcp_type(pkt)
+            msg_type = self.get_dhcp_type(pkt)
             if msg_type is None:
                 return
 
-            client_mac = pkt[sp.Ether].src
-            xid = pkt[sp.BOOTP].xid
+            client_mac = pkt[self.sp.Ether].src
+            xid = pkt[self.sp.BOOTP].xid
 
             if msg_type == "discover":
                 discover_count += 1
                 opts = extra_opts
                 if discover_count == 1 and first_extra_opts is not None:
                     opts = first_extra_opts
-                reply = build_reply(
+                reply = self.build_reply(
                     ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
                     xid, "offer", extra_opts=opts, sname=sname, file=file,
                 )
-                sp.sendp(reply, iface=ifname, verbose=0)
+                self.sp.sendp(reply, iface=ifname, verbose=0)
 
             elif msg_type == "request":
-                reply = build_reply(
+                reply = self.build_reply(
                     ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
                     xid, "ack", sname=sname, file=file,
                 )
-                sp.sendp(reply, iface=ifname, verbose=0)
+                self.sp.sendp(reply, iface=ifname, verbose=0)
                 done = True
 
-        sniffer = sp.AsyncSniffer(
+        sniffer = self.sp.AsyncSniffer(
             iface=ifname,
             filter="udp and dst port 67",
             prn=handle_dhcp,
@@ -165,6 +129,45 @@ class TestDhclient(VnetTestTemplate):
         result = self.wait_object(self.vnet_map["vnet1"].pipe, timeout=15)
         return discover_count, result
 
+
+    def get_dhcp_type(self, pkt):
+        """Extract the DHCP message type from a packet."""
+        if not pkt.haslayer(self.sp.DHCP):
+            return None
+        for opt in pkt[self.sp.DHCP].options:
+            if isinstance(opt, tuple) and opt[0] == "message-type":
+                return self.sp.DHCPTypes[opt[1]]
+        return None
+
+    def build_reply(self, iface, server_ip, client_ip, client_mac, xid, msg_type,
+                    extra_opts=None, sname=b"", file=b""):
+        """Build a DHCP OFFER or ACK packet."""
+        options = [
+            ("message-type", msg_type),
+            ("server_id", server_ip),
+            ("lease_time", 3600),
+            ("subnet_mask", "255.255.255.0"),
+            ("router", server_ip),
+        ]
+        if extra_opts:
+            options.extend(extra_opts)
+        options.append("end")
+
+        return (
+            self.sp.Ether(dst=client_mac, src=self.sp.get_if_hwaddr(iface))
+            / self.sp.IP(src=server_ip, dst="255.255.255.255")
+            / self.sp.UDP(sport=67, dport=68)
+            / self.sp.BOOTP(op=2,
+                xid=xid,
+                yiaddr=client_ip,
+                siaddr=server_ip,
+                chaddr=mac_to_chaddr(client_mac),
+                sname=sname,
+                file=file,
+            )
+            / self.sp.DHCP(options=options)
+        )
+
     @pytest.mark.require_user("root")
     def test_classless_routes_truncated(self):
         """