Fixing flake8

tests/qemu/machines.py

@@ -1,3 +1,5 @@
+# flake8: noqa E131
+
 import pathlib
 from runners import TestLeaveAllDefault
 
@@ -9,7 +11,6 @@ parameters = {
 			'-name', '"archinstall-test"',
 			'-display', 'none',
 			'-monitor', 'none',
-			#'-serial', 'none',
 			'-nographic',
 			'-m', '4096',
 			'-smp', '4,sockets=1,dies=1,cores=4,threads=1',
@@ -36,8 +37,8 @@ parameters = {
 			'-device', 'pcie-root-port,port=0x11,chassis=13,id=pci.13,bus=pcie.0,addr=0x6.0x3',
 			'-drive', 'if=pflash,format=raw,readonly=on,file=/usr/share/ovmf/x64/OVMF_CODE.secboot.fd',
 			'-drive', f'if=pflash,format=raw,file={pathlib.Path(__file__).parent}/OVMF_VARS.fd',
-			#'-tpmdev', 'passthrough,id=tpm0,path=/dev/tpm0,cancel-path=/tmp/foo-cancel2',
-			#'-device', 'tpm-tis,tpmdev=tpm0',
+			# '-tpmdev', 'passthrough,id=tpm0,path=/dev/tpm0,cancel-path=/tmp/foo-cancel2',
+			# '-device', 'tpm-tis,tpmdev=tpm0',
 			'-object', 'iothread,id=iothread1',
 				'-device', 'virtio-scsi-pci,bus=pcie.0,id=scsi2,addr=0x8',
 				'-device', 'virtio-scsi-pci,iothread=iothread1,id=scsi0,num_queues=8,bus=pci.13,addr=0x0',

tests/qemu/run_test.py

@@ -4,13 +4,10 @@ import threading
 import pathlib
 import socket
 import select
-import os
-import json
 import logging
 import sys
-import enum
 from subprocess import Popen, PIPE, STDOUT
-from qemu.qmp import QMPClient, Message
+from qemu.qmp import QMPClient
 from machines import parameters
 
 logger = logging.getLogger("archtest")
@@ -64,19 +61,25 @@ class SerialMonitor(threading.Thread):
 				'send-key',
 				arguments={
 					'keys': [
-						{ "type": "qcode", "data": "e" }
+						{
+							"type": "qcode",
+							"data": "e"
+						}
 					]
 				}
 			)
 		)
 
 		await asyncio.sleep(1)
-		ret = await self.QMP.qmp.execute_msg(
+		await self.QMP.qmp.execute_msg(
 			self.QMP.qmp.make_execute_msg(
 				'send-key',
 				arguments={
 					'keys': [
-						{ "type": "qcode", "data": "end" }
+						{
+							"type": "qcode",
+							"data": "end"
+						}
 					]
 				}
 			)
@@ -85,15 +88,15 @@ class SerialMonitor(threading.Thread):
 
 		keys = []
 		# https://gist.github.com/mvidner/8939289
-		keys.append({ "type": "qcode", "data": "spc" })
+		keys.append({"type": "qcode", "data": "spc"})
 		for character in list('console=tty0 console=ttyS0,115200'):
 			if character.isupper():
-				keys.append({ "type": "qcode", "data": 'caps_lock' })
-			keys.append({ "type": "qcode", "data": character.lower().replace('=', 'equal').replace(',', 'comma').replace(' ', 'spc') })
+				keys.append({"type": "qcode", "data": 'caps_lock'})
+			keys.append({"type": "qcode", "data": character.lower().replace('=', 'equal').replace(',', 'comma').replace(' ', 'spc')})
 			if character.isupper():
-				keys.append({ "type": "qcode", "data": 'caps_lock' })
+				keys.append({"type": "qcode", "data": 'caps_lock'})
 
-		ret = await self.QMP.qmp.execute_msg(
+		await self.QMP.qmp.execute_msg(
 			self.QMP.qmp.make_execute_msg(
 				'send-key',
 				arguments={
@@ -103,12 +106,15 @@ class SerialMonitor(threading.Thread):
 		)
 
 		await asyncio.sleep(1)
-		ret = await self.QMP.qmp.execute_msg(
+		await self.QMP.qmp.execute_msg(
 			self.QMP.qmp.make_execute_msg(
 				'send-key',
 				arguments={
 					'keys': [
-						{ "type": "qcode", "data": "kp_enter" }
+						{
+							"type": "qcode",
+							"data": "kp_enter"
+						}
 					]
 				}
 			)
@@ -181,7 +187,6 @@ class QemuSession(threading.Thread):
 		self.start()
 
 	def run(self):
-		#print(self.cmd)
 		self.handle = Popen(
 			' '.join(self.cmd),
 			stdout=PIPE,
@@ -200,9 +205,9 @@ class QemuSession(threading.Thread):
 				if fd == self.handle.stdout.fileno():
 					if (output := self.handle.stdout.read()):
 						print(output)
-				#elif fd == self.handle.stderr.fileno():
-				#	if (output := self.handle.stderr.read()):
-				#		print(output)
+				# elif fd == self.handle.stderr.fileno():
+				# 	if (output := self.handle.stderr.read()):
+				# 		print(output)
 			# No exit signal yet
 			time.sleep(0.25)
 
@@ -210,16 +215,17 @@ class QemuSession(threading.Thread):
 		for fd in r:
 			if fd == self.handle.stdout.fileno():
 				if (output := self.handle.stdout.read()):
-						print(output)
-				#elif fd == self.handle.stderr.fileno():
-				#	if (output := self.handle.stderr.read()):
-				#		print(output)
+					print(output)
+				# elif fd == self.handle.stderr.fileno():
+				# 	if (output := self.handle.stderr.read()):
+				# 		print(output)
 
 		self.handle.stdin.close()
 		self.handle.stdout.close()
 		# self.handle.stderr.close()
 		logger.warning("Qemu closed..")
 
+
 # .. todo::
 #    Needs a bit of more work to allow for multiple runners and test benches.
 for profile in parameters:

tests/qemu/testbase.py

@@ -10,7 +10,7 @@ logger = logging.getLogger("archtest")
 class Keyboard(enum.Enum):
 	# https://vt100.net/docs/vt100-ug/chapter3.html
 	# https://espterm.github.io/docs/VT100%20escape%20codes.html
-	arrow_up='\033[A'
+	arrow_up = '\033[A'
 	arrow_down = '\033[B'
 	arrow_right = '\033[C'
 	arrow_left = '\033[D'
@@ -29,11 +29,11 @@ class TestBase(threading.Thread):
 	def send_string(self, chars, delay=None):
 		logger.debug(f"Sending string: {chars.encode('UTF-8')}")
 
-		while len((r := select.select([self.serial_monitor.client_socket], [], [], 0.2)[0])):
+		while len(select.select([self.serial_monitor.client_socket], [], [], 0.2)[0]):
 			# Waiting for read buffer to finish to not collide
 			time.sleep(0.02)
 
-		if len(w := select.select([], [self.serial_monitor.client_socket], [], 0.2)[1]):
+		if len(select.select([], [self.serial_monitor.client_socket], [], 0.2)[1]):
 			# asyncio.run_coroutine_threadsafe(self._send_key(vt100_key), loop=self.serial_monitor.QMP.loop)
 			self.serial_monitor.client_socket.send(chars.encode('UTF-8'), socket.MSG_WAITALL) # flags=socket.MSG_WAITALL
 			# os.fsync(self.serial_monitor.client_socket.fileno())
@@ -52,11 +52,11 @@ class TestBase(threading.Thread):
 
 		logger.debug(f"Sending key: {key} \033[0;32m({vt100_key})\033[0m")
 
-		while len((r := select.select([self.serial_monitor.client_socket], [], [], 0.2)[0])):
+		while len(select.select([self.serial_monitor.client_socket], [], [], 0.2)[0]):
 			# Waiting for read buffer to finish to not collide
 			time.sleep(0.02)
 
-		if len(w := select.select([], [self.serial_monitor.client_socket], [], 0.2)[1]):
+		if len(select.select([], [self.serial_monitor.client_socket], [], 0.2)[1]):
 			# asyncio.run_coroutine_threadsafe(self._send_key(vt100_key), loop=self.serial_monitor.QMP.loop)
 			self.serial_monitor.client_socket.send(vt100_key, socket.MSG_WAITALL) # flags=socket.MSG_WAITALL
 			# os.fsync(self.serial_monitor.client_socket.fileno())