pyrox.dev
1import io
2import os
3import pty
4import subprocess
5from pathlib import Path
6
7from test_driver.logger import AbstractLogger
8
9
10class 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 stop(self) -> None:
63 self.logger.info(f"kill vlan (pid {self.pid})")
64 self.fd.close()
65 self.process.terminate()