diff --git a/.gitignore b/.gitignore index a6075cc4..f2f14f43 100644 --- a/.gitignore +++ b/.gitignore @@ -23,6 +23,8 @@ SAFETY_LOCK **/**.qcow2 **/**.log **/**.fd +**/**.cast +**/**.pid /test*.py **/archiso /guided.py diff --git a/tests/qemu/machines.py b/tests/qemu/machines.py index add91c08..8bcc00b1 100644 --- a/tests/qemu/machines.py +++ b/tests/qemu/machines.py @@ -1,51 +1,55 @@ import pathlib +from runners import TestLeaveAllDefault parameters = { - "pci_emulation" : [ - '/usr/bin/qemu-system-x86_64', - '-name', '"archinstall-test"', - '-display', 'none', - '-monitor', 'none', - #'-serial', 'none', - '-nographic', - '-usb', '-device', 'usb-host,vendorid=0x1050,productid=0x0407', - '-pidfile', f'{pathlib.Path(__file__).parent}/archinstall-test.pid', - '-cpu', 'host,topoext,kvm=off,hv_relaxed,hv_spinlocks=0x1fff,hv_vapic,hv_time', - '-enable-kvm', - '-object', 'rng-random,filename=/dev/urandom,id=rng0', - '-device', 'virtio-rng-pci,rng=rng0', - '-usbdevice', 'mouse', - '-global', 'driver=cfi.pflash01,property=secure,value=on', - '-machine', 'type=q35,accel=kvm,kernel_irqchip=on', - '-smbios', '"type=0,vendor=American Megatrends Inc.,version=P4.60,date=08/03/2021,release=08.03.2021"', - '-smbios', '"type=1,manufacturer=Inet_AB,product=To Be Filled By O.E.M.,version=To Be Filled By O.E.M.,serial=245797,uuid=4154a2b8-7b7f-0000-0000-000000000000,sku=To Be Filled By O.E.M.,family=To Be Filled By O.E.M."', - '-smbios', '"type=2,manufacturer=ASRock,product=X570 Taichi,version=,serial=M80-D00000012340asset=,location="', - '-smbios', '"type=3,manufacturer=To Be Filled By O.E.M.,version=To Be Filled By O.E.M.,serial=214345,asset=To Be Filled By O.E.M.,sku=To Be Filled By O.E.M."', - '-smbios', '"type=4,sock_pfx=AM4,manufacturer=Advanced Micro Devices,, Inc.,version=AMD Ryzen 9 5900X 12-Core Processor,serial=Unknown,asset=Unknown,part=Unknown"', - '-smbios', '"type=17,loc_pfx=DIMM 1,bank=P0 CHANNEL A,manufacturer=Unknown,serial=00000000,asset=Not Specified,part=CMK64GX4M2E3200C16,speed=2133"', - '-smbios', '"type=17,loc_pfx=DIMM 1,bank=P0 CHANNEL B,manufacturer=Unknown,serial=00000000,asset=Not Specified,part=CMK64GX4M2E3200C16,speed=2133"', - '-m', '4096', - '-smp', '4,sockets=1,dies=1,cores=4,threads=1', - '-device', 'intel-iommu,device-iotlb=on,caching-mode=on', - '-device', 'pcie-root-port,port=0xe,chassis=8,id=pci.8,bus=pcie.0,multifunction=on,addr=0x6', - '-device', 'virtio-keyboard-pci,id=input1,bus=pci.8,addr=0x0', - '-device', 'pcie-root-port,port=0xf,chassis=9,id=pci.9,bus=pcie.0,addr=0x6.0x1', - '-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', 'if=pflash,format=raw,file=/home/anton/archtest.fd', - #'-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', - '-device', '"scsi-hd,serial=S5GBAD12345ABCE,drive=libvirt-1-format,bus=scsi2.0,id=scsi0-0-0-0,channel=0,scsi-id=0,lun=0,device_id=drive-scsi0-0-0-0,bootindex=2,write-cache=on"', - '-blockdev', '\'{"driver":"file","filename":"/home/anton/archtest.img","aio":"threads","node-name":"libvirt-1-storage","cache":{"direct":true,"no-flush":false},"auto-read-only":true,"discard":"unmap"}\'', - '-blockdev', '\'{"node-name":"libvirt-1-format","read-only":false,"discard":"unmap","cache":{"direct":true,"no-flush":false},"driver":"qcow2","file":"libvirt-1-storage","backing":null}\'', - '-device', 'pcie-root-port,multifunction=on,bus=pcie.0,id=port9-0,addr=0x9,chassis=0', - '-device', 'virtio-net-pci,mac=FE:00:00:00:00:01,id=network0,netdev=network0.0,status=on,bus=port9-0', - '-netdev', 'tap,ifname=tap0,id=network0.0,script=/home/anton/Archtest_up.sh,downscript=no', - '-audiodev', 'pipewire,id=win11', - '-device', 'ich9-intel-hda,id=sound0,bus=pcie.0,addr=0x1b', - '-device', 'hda-micro,audiodev=win11', - ] + "pci_emulation" : { + 'test_class' : TestLeaveAllDefault, + 'arguments' : [ + '/usr/bin/qemu-system-x86_64', + '-name', '"archinstall-test"', + '-display', 'none', + '-monitor', 'none', + #'-serial', 'none', + '-nographic', + '-m', '4096', + '-smp', '4,sockets=1,dies=1,cores=4,threads=1', + '-usb', '-device', 'usb-host,vendorid=0x1050,productid=0x0407', + '-pidfile', f'{pathlib.Path(__file__).parent}/archinstall-test.pid', + '-cpu', 'host,topoext,kvm=off,hv_relaxed,hv_spinlocks=0x1fff,hv_vapic,hv_time', + '-enable-kvm', + '-object', 'rng-random,filename=/dev/urandom,id=rng0', + '-device', 'virtio-rng-pci,rng=rng0', + '-usbdevice', 'mouse', + '-global', 'driver=cfi.pflash01,property=secure,value=on', + '-machine', 'type=q35,accel=kvm,kernel_irqchip=on', + '-smbios', '"type=0,vendor=American Megatrends Inc.,version=P4.60,date=08/03/2021,release=08.03.2021"', + '-smbios', '"type=1,manufacturer=Inet_AB,product=To Be Filled By O.E.M.,version=To Be Filled By O.E.M.,serial=245797,uuid=4154a2b8-7b7f-0000-0000-000000000000,sku=To Be Filled By O.E.M.,family=To Be Filled By O.E.M."', + '-smbios', '"type=2,manufacturer=ASRock,product=X570 Taichi,version=,serial=M80-D00000012340asset=,location="', + '-smbios', '"type=3,manufacturer=To Be Filled By O.E.M.,version=To Be Filled By O.E.M.,serial=214345,asset=To Be Filled By O.E.M.,sku=To Be Filled By O.E.M."', + '-smbios', '"type=4,sock_pfx=AM4,manufacturer=Advanced Micro Devices,, Inc.,version=AMD Ryzen 9 5900X 12-Core Processor,serial=Unknown,asset=Unknown,part=Unknown"', + '-smbios', '"type=17,loc_pfx=DIMM 1,bank=P0 CHANNEL A,manufacturer=Unknown,serial=00000000,asset=Not Specified,part=CMK64GX4M2E3200C16,speed=2133"', + '-smbios', '"type=17,loc_pfx=DIMM 1,bank=P0 CHANNEL B,manufacturer=Unknown,serial=00000000,asset=Not Specified,part=CMK64GX4M2E3200C16,speed=2133"', + '-device', 'intel-iommu,device-iotlb=on,caching-mode=on', + '-device', 'pcie-root-port,port=0xe,chassis=8,id=pci.8,bus=pcie.0,multifunction=on,addr=0x6', + '-device', 'virtio-keyboard-pci,id=input1,bus=pci.8,addr=0x0', + '-device', 'pcie-root-port,port=0xf,chassis=9,id=pci.9,bus=pcie.0,addr=0x6.0x1', + '-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', + '-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', + '-device', '"scsi-hd,serial=S5GBAD12345ABCE,drive=libvirt-1-format,bus=scsi2.0,id=scsi0-0-0-0,channel=0,scsi-id=0,lun=0,device_id=drive-scsi0-0-0-0,bootindex=2,write-cache=on"', + '-blockdev', '\'{"driver":"file","filename":"/home/anton/archtest.img","aio":"threads","node-name":"libvirt-1-storage","cache":{"direct":true,"no-flush":false},"auto-read-only":true,"discard":"unmap"}\'', + '-blockdev', '\'{"node-name":"libvirt-1-format","read-only":false,"discard":"unmap","cache":{"direct":true,"no-flush":false},"driver":"qcow2","file":"libvirt-1-storage","backing":null}\'', + '-device', 'pcie-root-port,multifunction=on,bus=pcie.0,id=port9-0,addr=0x9,chassis=0', + '-device', 'virtio-net-pci,mac=FE:00:00:00:00:01,id=network0,netdev=network0.0,status=on,bus=port9-0', + '-netdev', 'tap,ifname=tap0,id=network0.0,script=no,downscript=no', + '-audiodev', 'pipewire,id=win11', + '-device', 'ich9-intel-hda,id=sound0,bus=pcie.0,addr=0x1b', + '-device', 'hda-micro,audiodev=win11', + ] + } } \ No newline at end of file diff --git a/tests/qemu/run_test.py b/tests/qemu/run_test.py index 43051240..f8e9679f 100644 --- a/tests/qemu/run_test.py +++ b/tests/qemu/run_test.py @@ -8,14 +8,15 @@ import os import json import logging import sys +import enum from subprocess import Popen, PIPE, STDOUT from qemu.qmp import QMPClient, Message from machines import parameters -logger = logging.getLogger(__name__) -logger.setLevel(logging.DEBUG) +logger = logging.getLogger("archtest") +logger.setLevel(logging.INFO) handler = logging.StreamHandler(sys.stdout) -handler.setLevel(logging.DEBUG) +handler.setLevel(logging.INFO) logger.addHandler(handler) class QMPClientMonitor: @@ -28,7 +29,7 @@ class QMPClientMonitor: async def watch_events(self): try: async for event in self.qmp.events: - print(f"Event: {event['event']}") + print(f"QMP Event: {event['event']}") except asyncio.CancelledError: return @@ -44,18 +45,18 @@ class QMPClientMonitor: except: pass - class SerialMonitor(threading.Thread): - def __init__(self, profile, QMP, serial_socket_path): + def __init__(self, profile, QMP, serial_socket_path, test_case): self.profile = profile self.QMP = QMP self.serial_socket_path = serial_socket_path + self.test_case = test_case(serial_monitor=self) threading.Thread.__init__(self) self.start() async def edit_boot(self): - logger.info("Sending 'e'") + logger.info("Adding 'console=tty0 console=ttyS0,115200' to default boot option") # https://github.com/coreos/qemu/blob/master/qmp-commands.hx await self.QMP.qmp.execute_msg( @@ -69,8 +70,7 @@ class SerialMonitor(threading.Thread): ) ) - logger.info("Sending 'end'") - asyncio.sleep(1) + await asyncio.sleep(1) ret = await self.QMP.qmp.execute_msg( self.QMP.qmp.make_execute_msg( 'send-key', @@ -81,7 +81,7 @@ class SerialMonitor(threading.Thread): } ) ) - asyncio.sleep(1) + await asyncio.sleep(1) keys = [] # https://gist.github.com/mvidner/8939289 @@ -92,7 +92,6 @@ class SerialMonitor(threading.Thread): 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": "kp_enter" }) ret = await self.QMP.qmp.execute_msg( self.QMP.qmp.make_execute_msg( @@ -116,73 +115,62 @@ class SerialMonitor(threading.Thread): ) async def login_root(self): - keys = [] - for character in list('root'): - if character.isupper(): - 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' }) + self.client_socket.send(b'root\015') - ret = await self.QMP.qmp.execute_msg( - self.QMP.qmp.make_execute_msg( - 'send-key', - arguments={ - 'keys': keys - } - ) - ) + async def run_archinstall(self): + # self.client_socket.send(b'tput cols\015') + # self.client_socket.send(b'tput lines\015') - await asyncio.sleep(1) - ret = await self.QMP.qmp.execute_msg( - self.QMP.qmp.make_execute_msg( - 'send-key', - arguments={ - 'keys': [ - { "type": "qcode", "data": "kp_enter" } - ] - } - ) - ) - - await asyncio.sleep(1) - ret = await self.QMP.qmp.execute_msg( - self.QMP.qmp.make_execute_msg( - 'send-key', - arguments={ - 'keys': [ - { "type": "qcode", "data": "kp_enter" } - ] - } - ) - ) + # For some reason, while running in this test mode, + # building archinstall fails randomly with "No such file or directory"." + # So a safe bet is to just re-run it manually before starting. + self.client_socket.send(b'python -m build --verbose --wheel --no-isolation\015') + time.sleep(1) + self.client_socket.send(b'pip install dist/archinstall*.whl --break-system-packages\015') + time.sleep(1) + self.client_socket.send(b'archinstall\015') def run(self): - client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - client_socket.connect(str(self.serial_socket_path)) + self.client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.client_socket.connect(str(self.serial_socket_path)) alive = True + entered_test_case = False + # The output of serial.log can be displayed with: + # tail -f serial.log + # Or record with: + # asciinema rec demo.cast -c "tail -f serial.log" with open('serial.log', 'wb') as fh: - while alive: - r, w, x = select.select([client_socket], [], [], 0.2) + while alive and self.test_case.exit_code == -1: + r, w, x = select.select([self.client_socket], [], [], 0.2) for fd in r: - if (output := client_socket.recv(8192)): - if b'Boot in' in output: - logger.info("Found booting") - asyncio.run_coroutine_threadsafe(self.edit_boot(), loop=self.QMP.loop) - elif b'archiso login:' in output: - logger.info("Found login prompt") - asyncio.run_coroutine_threadsafe(self.login_root(), loop=self.QMP.loop) - + if (output := self.client_socket.recv(8192)): fh.write(output) fh.flush() + # This block should be moved into the test class + if b'Boot in' in output and entered_test_case is False: + logger.info("Found boot prompt") + asyncio.run_coroutine_threadsafe(self.edit_boot(), loop=self.QMP.loop) + elif b'archiso login:' in output and entered_test_case is False: + logger.info("Found login prompt") + asyncio.run_coroutine_threadsafe(self.login_root(), loop=self.QMP.loop) + elif b'Type archinstall to launch the installer.' in output and entered_test_case is False: + logger.info("Found archinstall start point") + asyncio.run_coroutine_threadsafe(self.run_archinstall(), loop=self.QMP.loop) + entered_test_case = True + # ------- + elif entered_test_case: + self.test_case.feed(output) + else: - client_socket.close() + self.client_socket.close() alive = False break time.sleep(0.025) + print(f"Serial died: {alive}, {self.test_case.exit_code}") + class QemuSession(threading.Thread): def __init__(self, cmd, qmp_socket, serial_socket): self.cmd = cmd @@ -204,6 +192,8 @@ class QemuSession(threading.Thread): pass_fds=[self.qmp_socket.fileno(), self.serial_socket.fileno()] ) + # Run the qemu process until complete. + # And deal with the different buffers accordingly while self.handle.poll() is None: r, w, x = select.select([self.handle.stdout.fileno(), self.handle.stdout.fileno()], [], [], 0.2) for fd in r: @@ -230,6 +220,8 @@ class QemuSession(threading.Thread): # 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: qmp_socket_path = pathlib.Path(__file__).parent / "qmp.socket" serial_socket_path = pathlib.Path(__file__).parent / "serial.socket" @@ -237,6 +229,7 @@ for profile in parameters: qmp_socket_path.unlink(missing_ok=True) serial_socket_path.unlink(missing_ok=True) + logger.info(f"Creating serial ttyS0 and QMP sockets for use in Qemu") with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as qmp_socket: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as serial_socket: qmp_socket.bind(str(qmp_socket_path)) @@ -244,16 +237,17 @@ for profile in parameters: qmp_socket.listen(2) serial_socket.listen(2) - args = parameters[profile] + [ + args = parameters[profile]['arguments'] + [ '-chardev', f'socket,id=qmp1,fd={qmp_socket.fileno()},server=on,wait=off', '-chardev', f'socket,id=serial1,fd={serial_socket.fileno()},server=on,wait=on', '-mon', f'chardev=qmp1,mode=control,pretty=off', '-serial', f"chardev:serial1", - '-drive', f'file=/home/anton/Downloads/archlinux-2024.05.11-x86_64.iso,media=cdrom,cache=none,id=cdrom0,index=0' + '-drive', f'file=/home/anton/Downloads/archlinux-2024.05.13-x86_64.iso,media=cdrom,cache=none,id=cdrom0,index=0' ] + logger.info(f"Spawning Qemu test profile {profile}") session = QemuSession(args, qmp_socket, serial_socket) monitor = QMPClientMonitor(profile, str(qmp_socket_path)) - serial = SerialMonitor(profile, monitor, serial_socket_path) + serial = SerialMonitor(profile, monitor, serial_socket_path, test_case=parameters[profile]['test_class']) asyncio.run(monitor.run()) \ No newline at end of file diff --git a/tests/qemu/runners.py b/tests/qemu/runners.py new file mode 100644 index 00000000..fcdbc870 --- /dev/null +++ b/tests/qemu/runners.py @@ -0,0 +1,172 @@ +import threading +import time +import logging +import asyncio +from testbase import TestBase + +logger = logging.getLogger("archtest") + +class TestLeaveAllDefault(TestBase): + def __init__(self, serial_monitor): + super().__init__(serial_monitor) + self.running_test = False + self.installation_complete = False + self.installation_successful = False + self.fix_post_install_boot = False + self.entered_disk_enc_pw = False + + self.exit_code = -1 + + threading.Thread.__init__(self) + self.start() + + def run(self): + while True: + if b'Set/Modify the below options' in self.buffer and self.running_test is False: + # This is our entrypoint, from here on we'll have to fly blind because + # the menu system uses ANSI/VT100 escape codes to put the arrow and stuff. + # So we can't check for `> Archinstall langauge` for instance. + # So lets fire away our test sequence and monitor for some break-point values. + logger.info(f"Running through a typical installation scenario accepting all defaults.") + self.running_test = True + time.sleep(1) + + logger.info(f"Selecting a good regional mirror") + # Navigate to Mirror selection + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + self.send_key('enter', delay=0.2) + time.sleep(2) # Let the UI buffer complete before we start searching + # Search for a known mirror region with good speed relative to source of execution + self.send_key('/', delay=0.2) + self.send_string('Sweden', delay=0.2) + self.send_key('enter', delay=0.2) + # Exit the mirror selection screen + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Enter 'locale' option + logger.info(f"Setting a keyboard locale to something known") + self.send_key('enter', delay=0.2) + # Enter keyboard layout + self.send_key('enter', delay=0.2) + time.sleep(1) # Let the UI buffer complete again before searching + self.send_key('/', delay=0.2) + time.sleep(0.5) + self.send_string('sv-latin1', delay=0.2) + self.send_key('enter', delay=0.2) + # Exit locale + logger.info(f"Selecting the one virtual harddrive") + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Enter disk config + self.send_key('enter', delay=0.2) + # Enter partitioning + self.send_key('enter', delay=0.2) + # Use best effort + self.send_key('enter', delay=0.2) + # Select second option, which should be Virtio Block Device + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Select btrfs + logger.info(f"Selecting btrfs + subvolumes + compression") + self.send_key('enter', delay=0.2) + # Use subvolumes + self.send_key('enter', delay=0.2) + # Use compression + self.send_key('enter', delay=0.2) + # Exit partitioning + logger.info(f"Setting disk encryption password") + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Enter disk encryption + self.send_key('enter', delay=0.2) + # Select encryption type + self.send_key('enter', delay=0.2) + # Select LUKS + self.send_key('enter', delay=0.2) + # Enter password selection + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Enter a test password + self.send_string('test', delay=0.2) + self.send_key('enter', delay=0.2) + self.send_string('test', delay=0.2) + self.send_key('enter', delay=0.2) + # Select partition + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Select the default partition + self.send_key('enter', delay=0.2) + # Exit encryption menu + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('enter', delay=0.2) + # Skip some entries in the main menu + logger.info(f"Setting a root password") + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) # Move to "root password" + self.send_key('enter', delay=0.2) + # Enter a test root password + self.send_string('test', delay=0.2) + self.send_key('enter', delay=0.2) + self.send_string('test', delay=0.2) + self.send_key('enter', delay=0.2) + # Skip some entries + logger.info(f"Adding nano as additional password") + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) # Additional packages + self.send_key('enter', delay=0.2) + # Enter a test package that is reliable + self.send_string('nano', delay=0.2) + self.send_key('enter', delay=0.2) + logger.info(f"Setting network config") + # Configure networking + time.sleep(1) + self.send_key('enter', delay=0.2) + self.send_key('enter', delay=0.2) + time.sleep(2) + # Proceed to installation + logger.info(f"Proceeding to install") + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) + self.send_key('arrow_down', delay=0.2) # Install + self.send_key('enter', delay=0.2) + # Proceed to install + self.send_key('enter', delay=0.2) + elif b'Would you like to chroot' in self.buffer and self.installation_complete is False: + logger.info(f"Installation appears to have completed") + self.installation_complete = True + self.send_key('arrow_down', delay=0.2) # No + self.send_key('enter', delay=0.2) + elif b'Installation completed without any errors' in self.buffer and self.installation_successful is False: + logger.info(f"Installation was successful, rebooting") + self.installation_successful = True + self.send_string('reboot', delay=0.2) + self.send_key('enter', delay=0.2) + elif b'Arch Linux (linux-fallback)' in self.buffer and self.installation_successful is True and self.fix_post_install_boot is False: + logger.info(f"Found linux-fallback, assuming bootloader needs adjusting for serial output") + self.fix_post_install_boot = True + asyncio.run_coroutine_threadsafe(self.serial_monitor.edit_boot(), loop=self.serial_monitor.QMP.loop) + elif b'A password is required to access the root volume' in self.buffer and self.installation_successful is True and self.fix_post_install_boot is True and self.entered_disk_enc_pw is False: + logger.info(f"Found disk encryption password, supplying it") + self.entered_disk_enc_pw = True + self.send_string('test', delay=0.2) + self.send_key('enter', delay=0.2) + elif b'archlinux login:' in self.buffer[-500:] and self.installation_successful and self.entered_disk_enc_pw: + logger.info("Installation successful, for real!") + self.exit_code = 0 + break + + time.sleep(0.25) \ No newline at end of file diff --git a/tests/qemu/testbase.py b/tests/qemu/testbase.py new file mode 100644 index 00000000..342b300e --- /dev/null +++ b/tests/qemu/testbase.py @@ -0,0 +1,73 @@ +import enum +import time +import threading +import select +import logging +import socket + +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_down = '\033[B' + arrow_right = '\033[C' + arrow_left = '\033[D' + enter = '\015' + forward_slash = '\057' + escape = '\033' + +class TestBase(threading.Thread): + def __init__(self, serial_monitor): + self.serial_monitor = serial_monitor + self.buffer = b'' + + async def _send_key(self, key): + pass # This is for the QMP socket if needed + + 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])): + # 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]): + # 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()) + else: + logger.error(f"Could not send key, serial was not in a write state!") + + if delay: + time.sleep(delay) + + def send_key(self, key, delay=None): + try: + vt100_key = Keyboard[key].value.encode('UTF-8') + except: + # logger.warning(f"Could not convert key: {key}") + vt100_key = key.encode('UTF-8') + + 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])): + # 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]): + # 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()) + else: + logger.error(f"Could not send key, serial was not in a write state!") + + if delay: + time.sleep(delay) + + # Flushing the serial buffer, as it can quickly become frozen + # self.serial_monitor.client_socket.sendall(b"") + + def feed(self, line): + self.buffer += line \ No newline at end of file