diff --git a/tests/qemu/machines.py b/tests/qemu/machines.py index 8bcc00b1..e478453d 100644 --- a/tests/qemu/machines.py +++ b/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', diff --git a/tests/qemu/run_test.py b/tests/qemu/run_test.py index f8e9679f..bb653951 100644 --- a/tests/qemu/run_test.py +++ b/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: diff --git a/tests/qemu/testbase.py b/tests/qemu/testbase.py index 342b300e..e7a4760f 100644 --- a/tests/qemu/testbase.py +++ b/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())