JakeHillion/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

nixos/test: update test-driver.py for mypy 0.770

Browse Source
This commit is contained in:
Rouven Czerwinski 2020-03-27 08:25:07 +01:00 committed by Jon
parent 279438e7f8
commit 1685cd7fa7
1 changed files with 4 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
nixos/lib/test-driver/test-driver.py
View File

@ -6,6 +6,7 @@ from xml.sax.saxutils import XMLGenerator
import _thread import _thread
import atexit import atexit
import base64 import base64
import codecs
import os import os
import pathlib import pathlib
import ptpython.repl import ptpython.repl
@ -115,6 +116,7 @@ def create_vlan(vlan_nr: str) -> Tuple[str, str, "subprocess.Popen[bytes]", Any]
fd.write("version\n") fd.write("version\n")
# TODO: perl version checks if this can be read from # TODO: perl version checks if this can be read from
# an if not, dies. we could hang here forever. Fix it. # an if not, dies. we could hang here forever. Fix it.
assert vde_process.stdout is not None
vde_process.stdout.readline() vde_process.stdout.readline()
if not os.path.exists(os.path.join(vde_socket, "ctl")): if not os.path.exists(os.path.join(vde_socket, "ctl")):
raise Exception("cannot start vde_switch") raise Exception("cannot start vde_switch")
@ -139,7 +141,7 @@ def retry(fn: Callable) -> None:
class Logger: class Logger:
def __init__(self) -> None: def __init__(self) -> None:
self.logfile = os.environ.get("LOGFILE", "/dev/null") self.logfile = os.environ.get("LOGFILE", "/dev/null")
self.logfile_handle = open(self.logfile, "wb") self.logfile_handle = codecs.open(self.logfile, "wb")
self.xml = XMLGenerator(self.logfile_handle, encoding="utf-8") self.xml = XMLGenerator(self.logfile_handle, encoding="utf-8")
self.queue: "Queue[Dict[str, str]]" = Queue(1000) self.queue: "Queue[Dict[str, str]]" = Queue(1000)
@ -739,6 +741,7 @@ class Machine:
self.shell, _ = self.shell_socket.accept() self.shell, _ = self.shell_socket.accept()
def process_serial_output() -> None: def process_serial_output() -> None:
assert self.process.stdout is not None
for _line in self.process.stdout: for _line in self.process.stdout:
# Ignore undecodable bytes that may occur in boot menus # Ignore undecodable bytes that may occur in boot menus
line = _line.decode(errors="ignore").replace("\r", "").rstrip() line = _line.decode(errors="ignore").replace("\r", "").rstrip()