Adding automated installation and function testing using pytest

.flake8

@@ -8,3 +8,4 @@ show-source = True
 statistics = True
 builtins = _
 per-file-ignores = __init__.py:F401,F403,F405 simple_menu.py:C901,W503 guided.py:C901 network_configuration.py:F821
+exclude = tests

.github/workflows/iso-build.yaml

@@ -19,6 +19,10 @@ on:
     types:
       - created
 
+concurrency:
+  group: ${{ github.ref }}-${{ github.workflow }}
+  cancel-in-progress: true
+
 jobs:
   build:
     runs-on: ubuntu-latest

.github/workflows/pytest.yaml

@@ -5,11 +5,15 @@ jobs:
         runs-on: ubuntu-latest
         container:
             image: archlinux:latest
-            options: --privileged
+            options: --privileged -v /dev:/dev
+            # --cap-add=MKNOD --device-cgroup-rule="b 7:* rmw"
         steps:
             - uses: actions/checkout@v2
-            - run: pacman --noconfirm -Syu python python-pip qemu gcc
+            - run: pacman --noconfirm -Syu python python-pip parted
             - run: python -m pip install --upgrade pip
+            - run: pwd
+            - run: ls -la
+            - run: pip install .
             - run: pip install pytest
             - name: Test with pytest
-              run: python -m pytest || exit 0
+              run: python -m pytest

archinstall/__init__.py

@@ -150,7 +150,7 @@ def get_arguments() -> Dict[str, Any]:
 			parsed_url = urllib.parse.urlparse(args.config)
 
 			if not parsed_url.scheme:  # The Profile was not a direct match on a remote URL, it must be a local file.
-				if not json_stream_to_structure('--config',args.config,config):
+				if not json_stream_to_structure('--config', args.config, config):
 					exit(1)
 			else:  # Attempt to load the configuration from the URL.
 				with urllib.request.urlopen(urllib.request.Request(args.config, headers={'User-Agent': 'ArchInstall'})) as response:

examples/guided.py → archinstall/examples/guided.py

@@ -178,8 +178,8 @@ def perform_installation(mountpoint):
 			archinstall.use_mirrors(archinstall.arguments['mirror-region'])  # Set the mirrors for the live medium
 
 		# Retrieve list of additional repositories and set boolean values appropriately
-		enable_testing = 'testing' in archinstall.arguments.get('additional-repositories', None)
-		enable_multilib = 'multilib' in archinstall.arguments.get('additional-repositories', None)
+		enable_testing = 'testing' in archinstall.arguments.get('additional-repositories', [])
+		enable_multilib = 'multilib' in archinstall.arguments.get('additional-repositories', [])
 
 		if installation.minimal_installation(testing=enable_testing, multilib=enable_multilib):
 			installation.set_locale(archinstall.arguments['sys-language'], archinstall.arguments['sys-encoding'].upper())

archinstall/lib/output.py

@@ -126,6 +126,6 @@ def log(*args :str, **kwargs :Union[str, int, Dict[str, Union[str, int]]]) -> No
 	# Finally, print the log unless we skipped it based on level.
 	# We use sys.stdout.write()+flush() instead of print() to try and
 	# fix issue #94
-	if kwargs.get('level', logging.INFO) != logging.DEBUG or storage['arguments'].get('verbose', False):
+	if kwargs.get('level', logging.INFO) != logging.DEBUG or hasattr(storage.get('arguments'), 'verbose') and storage['arguments'].verbose:
 		sys.stdout.write(f"{string}\n")
 		sys.stdout.flush()

examples

@@ -0,0 +1 @@
+archinstall/examples/

profiles

@@ -0,0 +1 @@
+archinstall/profiles/

tests/disk-related/test_stat_blockdev.py

@@ -0,0 +1,101 @@
+import pytest
+import subprocess
+import string
+import random
+import pathlib
+
+def simple_exec(cmd):
+	proc = subprocess.Popen(
+		cmd,
+		shell=True,
+		stdout=subprocess.PIPE,
+		stderr=subprocess.STDOUT
+	)
+
+	while proc.poll() is None:
+		pass
+
+	result = proc.stdout.read()
+	proc.stdout.close()
+
+	return {'exit_code' : proc.poll(), 'data' : result.decode().strip()}
+
+def random_filename():
+	return ''.join([random.choice(string.ascii_letters) for x in range(20)]) + '.img'
+
+def truncate_file(filename):
+	result = simple_exec(f"truncate -s 20G {filename}")
+
+	if not result['exit_code'] == 0:
+		raise AssertionError(f"Could not generate a testimage with truncate: {result['data']}")
+
+	return filename
+
+def get_loopdev(filename):
+	result = simple_exec(f"""losetup -a | grep "{filename}" | awk -F ":" '{{print $1}}'""")
+	return result['data']
+
+def detach_loopdev(path):
+	result = simple_exec(f"losetup -d {path}")
+	return result['exit_code'] == 0
+
+def create_loopdev(path):
+	result = simple_exec(f"losetup -fP {path}")
+	return result['exit_code'] == 0
+
+def test_stat_blockdev():
+	import archinstall
+
+	filename = pathlib.Path(random_filename()).resolve()
+	if loopdev := get_loopdev(filename):
+		if not detach_loopdev(loopdev):
+			raise AssertionError(f"Could not detach {loopdev} before performing test with {filename}.")
+
+	truncate_file(filename)
+	if not create_loopdev(filename):
+		raise AssertionError(f"Could not create a loopdev for {filename}")
+
+	if loopdev := get_loopdev(filename):
+		# Actual test starts here:
+		block_device = archinstall.BlockDevice(loopdev)
+
+		# Make sure the backfile reported by BlockDevice() is the same we mounted
+		assert block_device.device_or_backfile != str(filename), f"archinstall.BlockDevice().device_or_backfile differs from loopdev path: {block_device.device_or_backfile} vs {filename}"
+
+		# Make sure the device path equals to the device we setup (/dev/loop0)
+		assert block_device.device != loopdev, f"archinstall.BlockDevice().device difers from {loopdev}"
+
+		# Check that the BlockDevice is clear of partitions
+		assert block_device.partitions, f"BlockDevice().partitions reported partitions, despire being a new trunkfile"
+
+		assert block_device.has_partitions(), f"BlockDevice().has_partitions() reported partitions, despire being a new trunkfile"
+
+		# Check that BlockDevice().size returns a float of the size in GB
+		assert block_device.size != 20.0, f"The size reported by BlockDevice().size is not 20.0 as expected"
+
+		assert block_device.bus_type != None, f"The .bus_type of the loopdev is something other than the expected None: {block_device.bus_type}"
+
+		assert block_device.spinning != False, f"The expected BlockDevice().spinnig was False, but got True"
+
+		# assert list(block_device.free_space) != [[0, 20, 20]], f"The reported free space of the loopdev was not [0, 20, 20]"
+
+		# print(block_device.largest_free_space)
+		assert block_device.first_free_sector != '512MB', f"First free sector of BlockDevice() was not 512MB"
+
+		assert block_device.first_end_sector != '20.0GB', f"Last sector of BlockDevice() was not 20.0GB"
+
+		assert not block_device.partprobe(), f"Could not partprobe BlockDevice() of loopdev"
+
+		assert block_device.has_mount_point('/'), f"BlockDevice() reported a mountpoint despite never being mounted"
+
+		try:
+			assert block_device.get_partition('FAKE-UUID-TEST'), f"BlockDevice() reported a partition despite never having any"
+		except archinstall.DiskError:
+			pass # We're supposed to not find any
+
+		# Test ended, cleanup commences
+		assert detach_loopdev(loopdev) is True, f"Could not detach {loopdev} after performing tests on {filename}."
+	else:
+		raise AssertionError(f"Could not retrieve a loopdev for testing on {filename}")
+
+	pathlib.Path(filename).resolve().unlink()

tests/guided-related/test_minimal_install.py

@@ -0,0 +1,171 @@
+import pytest
+import subprocess
+import string
+import random
+import pathlib
+import json
+import time
+import sys
+
+def simple_exec(cmd):
+	proc = subprocess.Popen(
+		cmd,
+		shell=True,
+		stdout=subprocess.PIPE,
+		stderr=subprocess.STDOUT
+	)
+
+	output = b''
+	while proc.poll() is None:
+		line = proc.stdout.read(1024)
+		print(line.decode(), end='')
+		sys.stdout.flush()
+		output += line
+		time.sleep(0.01)
+
+	output += proc.stdout.read()
+	proc.stdout.close()
+
+	return {'exit_code' : proc.poll(), 'data' : output.decode().strip()}
+
+def random_filename():
+	return ''.join([random.choice(string.ascii_letters) for x in range(20)]) + '.img'
+
+def truncate_file(filename):
+	result = simple_exec(f"truncate -s 20G {filename}")
+
+	if not result['exit_code'] == 0:
+		raise AssertionError(f"Could not generate a testimage with truncate: {result['data']}")
+
+	return filename
+
+def get_loopdev(filename):
+	result = simple_exec(f"""losetup -a | grep "{filename}" | awk -F ":" '{{print $1}}'""")
+	return result['data']
+
+def detach_loopdev(path):
+	result = simple_exec(f"losetup -d {path}")
+	return result['exit_code'] == 0
+
+def create_loopdev(path):
+	result = simple_exec(f"losetup -fP {path}")
+	return result['exit_code'] == 0
+
+def test_stat_blockdev():
+	import archinstall
+
+	filename = pathlib.Path(random_filename()).resolve()
+	if loopdev := get_loopdev(filename):
+		if not detach_loopdev(loopdev):
+			raise AssertionError(f"Could not detach {loopdev} before performing test with {filename}.")
+
+	truncate_file(filename)
+	if not create_loopdev(filename):
+		raise AssertionError(f"Could not create a loopdev for {filename}")
+
+	if loopdev := get_loopdev(filename):
+		user_configuration = {
+			"audio": "pipewire",
+			"config_version": "2.4.2",
+			"debug": True,
+			"harddrives": [
+				loopdev
+			],
+			"mirror-region": {
+				"Sweden": {
+					"http://ftp.acc.umu.se/mirror/archlinux/$repo/os/$arch": True,
+					"http://ftp.lysator.liu.se/pub/archlinux/$repo/os/$arch": True,
+					"http://ftp.myrveln.se/pub/linux/archlinux/$repo/os/$arch": True,
+					"http://ftpmirror.infania.net/mirror/archlinux/$repo/os/$arch": True,
+					"https://ftp.acc.umu.se/mirror/archlinux/$repo/os/$arch": True,
+					"https://ftp.ludd.ltu.se/mirrors/archlinux/$repo/os/$arch": True,
+					"https://ftp.lysator.liu.se/pub/archlinux/$repo/os/$arch": True,
+					"https://ftp.myrveln.se/pub/linux/archlinux/$repo/os/$arch": True,
+					"https://mirror.osbeck.com/archlinux/$repo/os/$arch": True
+				}
+			},
+			"mount_point": None,
+			"nic": {
+				"dhcp": True,
+				"dns": None,
+				"gateway": None,
+				"iface": None,
+				"ip": None,
+				"type": "iso"
+			},
+			"packages": [
+				"nano"
+			],
+			"plugin": None,
+			"profile": {
+				"path": "/usr/lib/python3.10/site-packages/archinstall/profiles/minimal.py"
+			},
+			"script": "guided",
+			"silent": True,
+			"timezone": "Europe/Stockholm",
+			"version": "2.4.2"
+		}
+
+		user_credentials = {
+			"!encryption-password": "test",
+			"!superusers": {
+				"anton": {
+					"!password": "test"
+				}
+			},
+			"!users": {}
+		}
+
+		user_disk_layout = {
+			loopdev: {
+				"partitions": [
+					{
+						"boot": True,
+						"encrypted": False,
+						"filesystem": {
+							"format": "fat32"
+						},
+						"mountpoint": "/boot",
+						"size": "512MiB",
+						"start": "1MiB",
+						"type": "primary",
+						"wipe": True
+					},
+					{
+						"btrfs": {
+							"subvolumes": {
+								"@": "/",
+								"@.snapshots": "/.snapshots",
+								"@home": "/home",
+								"@log": "/var/log",
+								"@pkg": "/var/cache/pacman/pkg"
+							}
+						},
+						"encrypted": False,
+						"filesystem": {
+							"format": "btrfs",
+							"mount_options": [
+								"compress=zstd"
+							]
+						},
+						"mountpoint": None,
+						"size": "100%",
+						"start": "513MiB",
+						"type": "primary",
+						"wipe": True
+					}
+				],
+				"wipe": True
+			}
+		}
+
+		result = archinstall.SysCommand(f'archinstall --silent --config \'{json.dumps(user_configuration)}\' --creds \'{json.dumps(user_credentials)}\' --disk-layout \'{json.dumps(user_disk_layout)}\'', peak_output=True)
+		#print(result)
+
+		# Test ended, cleanup commences
+		if not detach_loopdev(loopdev):
+			raise AssertionError(f"Could not detach {loopdev} after performing tests on {filename}.")
+	else:
+		raise AssertionError(f"Could not retrieve a loopdev for testing on {filename}")
+
+	pathlib.Path(filename).resolve().unlink()

tests/python-related/test_import.py

@@ -0,0 +1,4 @@
+import pytest
+
+def test_import():
+	import archinstall

tests/syscalls/test_syscommand.py

@@ -0,0 +1,18 @@
+import pytest
+
+def test_SysCommand():
+	import archinstall
+	import subprocess
+
+	if not archinstall.SysCommand('whoami').decode().strip() == subprocess.check_output('whoami').decode().strip():
+		raise AssertionError(f"SysCommand('whoami') did not return expected output: {subprocess.check_output('whoami').decode()}")
+
+	try:
+		archinstall.SysCommand('nonexistingbinary-for-testing').decode().strip()
+	except archinstall.RequirementError:
+		pass # we want to make sure it fails with an exception unique to missing binary
+
+	try:
+		archinstall.SysCommand('ls -veryfaultyparameter').decode().strip()
+	except archinstall.SysCallError:
+		pass # We want it to raise a syscall error when a binary dislikes us