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