nixos/lib/test-driver/test_driver/vlan.py

   1 import io
   2 import os
   3 import pty
   4 import subprocess
   5 from pathlib import Path
   6
   7 from test_driver.logger import AbstractLogger
   8
   9
  10 class VLan:
  11     """This class handles a VLAN that the run-vm scripts identify via its
  12     number handles. The network's lifetime equals the object's lifetime.
  13     """
  14
  15     nr: int
  16     socket_dir: Path
  17
  18     process: subprocess.Popen
  19     pid: int
  20     fd: io.TextIOBase
  21
  22     logger: AbstractLogger
  23
  24     def __repr__(self) -> str:
  25         return f"<Vlan Nr. {self.nr}>"
  26
  27     def __init__(self, nr: int, tmp_dir: Path, logger: AbstractLogger):
  28         self.nr = nr
  29         self.socket_dir = tmp_dir / f"vde{self.nr}.ctl"
  30         self.logger = logger
  31
  32         # TODO: don't side-effect environment here
  33         os.environ[f"QEMU_VDE_SOCKET_{self.nr}"] = str(self.socket_dir)
  34
  35         self.logger.info("start vlan")
  36         pty_master, pty_slave = pty.openpty()
  37
  38         # The --hub is required for the scenario determined by
  39         # nixos/tests/networking.nix vlan-ping.
  40         # VLAN Tagged traffic (802.1Q) seams to be blocked if a vde_switch is
  41         # used without the hub mode (flood packets to all ports).
  42         self.process = subprocess.Popen(
  43             ["vde_switch", "-s", self.socket_dir, "--dirmode", "0700", "--hub"],
  44             stdin=pty_slave,
  45             stdout=subprocess.PIPE,
  46             stderr=subprocess.PIPE,
  47             shell=False,
  48         )
  49         self.pid = self.process.pid
  50         self.fd = os.fdopen(pty_master, "w")
  51         self.fd.write("version\n")
  52
  53         # TODO: perl version checks if this can be read from
  54         # an if not, dies. we could hang here forever. Fix it.
  55         assert self.process.stdout is not None
  56         self.process.stdout.readline()
  57         if not (self.socket_dir / "ctl").exists():
  58             self.logger.error("cannot start vde_switch")
  59
  60         self.logger.info(f"running vlan (pid {self.pid}; ctl {self.socket_dir})")
  61
  62     def __del__(self) -> None:
  63         self.logger.info(f"kill vlan (pid {self.pid})")
  64         self.fd.close()
  65         self.process.terminate()