diff --git a/archinstall/lib/disk/device_handler.py b/archinstall/lib/disk/device_handler.py index 1e5d52224a..71ef1e12e1 100644 --- a/archinstall/lib/disk/device_handler.py +++ b/archinstall/lib/disk/device_handler.py @@ -13,6 +13,7 @@ get_lsblk_info, linux_root_guid, mount, + swapoff, udev_sync, umount, ) @@ -513,6 +514,8 @@ def umount_all_existing(self, device_path: Path) -> None: # un-mount for existing encrypted partitions if partition.fs_type == FilesystemType.CRYPTO_LUKS: Luks2(partition.path).lock() + elif partition.fs_type == FilesystemType.LINUX_SWAP: + swapoff(partition.path) else: umount(partition.path, recursive=True) diff --git a/archinstall/lib/disk/utils.py b/archinstall/lib/disk/utils.py index 32b69130ef..4fbf0caafb 100644 --- a/archinstall/lib/disk/utils.py +++ b/archinstall/lib/disk/utils.py @@ -1,3 +1,5 @@ +from codecs import escape_decode +from os import fsdecode from pathlib import Path from subprocess import CalledProcessError @@ -198,6 +200,17 @@ def swapon(path: Path) -> None: raise DiskError(f'Could not enable swap {path}:\n{err.message}') +def swapoff(path: Path) -> None: + try: + output = SysCommand(['swapon', '--show=NAME', '--noheadings', '--raw']).output() + # --raw escapes filename bytes as \xNN. + active = {Path(fsdecode(escape_decode(line)[0])).resolve() for line in output.splitlines()} + if path.resolve() in active: + SysCommand(['swapoff', str(path)]) + except SysCallError as err: + raise DiskError(f'Could not disable swap {path}:\n{err.message}') + + def linux_root_guid(arch: str | None) -> PartitionGUID: if arch == 'aarch64': return PartitionGUID.LINUX_ROOT_AARCH64 diff --git a/archinstall/lib/models/device.py b/archinstall/lib/models/device.py index b266eaeb00..515c63010a 100644 --- a/archinstall/lib/models/device.py +++ b/archinstall/lib/models/device.py @@ -1631,6 +1631,15 @@ def convert_size(cls, value: Any, info: ValidationInfo) -> Any: return Size(value, Unit.B, sector_size) return value + @field_validator('mountpoint', 'mountpoints', mode='before') + @classmethod + def remove_swap_mountpoints(cls, value: Any) -> Any: + if value == '[SWAP]': + return None + if isinstance(value, list): + return [item for item in value if item != '[SWAP]'] + return value + @field_validator('mountpoints', 'fsroots', mode='before') @classmethod def remove_none(cls, value: Any) -> Any: diff --git a/tests/test_lsblk_swap.py b/tests/test_lsblk_swap.py new file mode 100644 index 0000000000..57077e559b --- /dev/null +++ b/tests/test_lsblk_swap.py @@ -0,0 +1,134 @@ +from os import fsdecode, fsencode +from pathlib import Path +from typing import Any +from unittest.mock import Mock, call + +import pytest + +from archinstall.lib.disk import utils +from archinstall.lib.exceptions import DiskError, SysCallError +from archinstall.lib.log import logger +from archinstall.lib.models.device import FilesystemType, LsblkInfo + +SAMPLE_PARTITION: dict[str, Any] = { + 'name': 'sda2', + 'path': '/dev/sda2', + 'pkname': 'sda', + 'log-sec': 512, + 'size': 4294967296, + 'pttype': 'gpt', + 'ptuuid': '5f1e1b8a', + 'rota': True, + 'tran': 'sata', + 'partn': 2, + 'partuuid': '0d2a1f7c', + 'parttype': '0657fd6d-a4ab-43c4-84e5-0933c84b4f4f', + 'uuid': 'e3c9b4a1', + 'fstype': 'swap', + 'fsver': '1', + 'fsavail': None, + 'fsuse%': None, + 'type': 'part', + 'mountpoint': None, + 'mountpoints': [None], + 'fsroots': [], +} + +SWAPON_QUERY = ['swapon', '--show=NAME', '--noheadings', '--raw'] + + +@pytest.mark.parametrize( + ('mountpoint', 'mountpoints', 'expected'), + [ + ('[SWAP]', ['[SWAP]'], (None, [])), + ('[SWAP]', [None], (None, [])), + (None, ['[SWAP]'], (None, [])), + (None, [None], (None, [])), + ('/home', ['/home', None], (Path('/home'), [Path('/home')])), + ('/mnt/[SWAP]', ['/mnt/[SWAP]'], (Path('/mnt/[SWAP]'), [Path('/mnt/[SWAP]')])), + ], +) +def test_swap_mountpoints(mountpoint: str | None, mountpoints: list[str | None], expected: tuple[Path | None, list[Path]]) -> None: + info = LsblkInfo.model_validate(SAMPLE_PARTITION | {'mountpoint': mountpoint, 'mountpoints': mountpoints}) + assert (info.mountpoint, info.mountpoints) == expected + + +@pytest.mark.parametrize('active', [False, True]) +def test_swapoff_only_disables_active_swap(monkeypatch: pytest.MonkeyPatch, active: bool) -> None: + command = Mock() + command.return_value.output.return_value = b'/dev/sda2\n' if active else b'/dev/sdb1\n' + monkeypatch.setattr(utils, 'SysCommand', command) + + utils.swapoff(Path('/dev/sda2')) + + expected = [call(SWAPON_QUERY)] + if active: + expected.append(call(['swapoff', '/dev/sda2'])) + assert command.call_args_list == expected + + +@pytest.mark.parametrize( + ('name', 'encoded'), + [ + ('sda2', b'sda2'), + ('swap file', b'swap\\x20file'), + ('swap\nfile', b'swap\\x0afile'), + ('swap\\x20file', b'swap\\x5cx20file'), + ('swäp', 'swäp'.encode()), + (fsdecode(b'swap-\xff'), b'swap-\\xff'), + ], +) +def test_swapoff_matches_escaped_paths_and_aliases(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, name: str, encoded: bytes) -> None: + device = tmp_path / name + device.touch() + alias = tmp_path / 'by-uuid' + alias.symlink_to(device) + command = Mock() + command.return_value.output.return_value = fsencode(tmp_path) + b'/' + encoded + b'\n' + monkeypatch.setattr(utils, 'SysCommand', command) + + utils.swapoff(alias) + + assert command.call_args_list == [call(SWAPON_QUERY), call(['swapoff', str(alias)])] + + +@pytest.mark.parametrize('query_fails', [False, True]) +def test_swapoff_errors(monkeypatch: pytest.MonkeyPatch, query_fails: bool) -> None: + result = Mock() + result.output.return_value = b'/dev/sda2\n' + error = SysCallError('command failed', exit_code=1) + command = Mock(side_effect=[error] if query_fails else [result, error]) + monkeypatch.setattr(utils, 'SysCommand', command) + + with pytest.raises(DiskError, match='Could not disable swap /dev/sda2:'): + utils.swapoff(Path('/dev/sda2')) + + expected = [call(SWAPON_QUERY)] + if not query_fails: + expected.append(call(['swapoff', '/dev/sda2'])) + assert command.call_args_list == expected + + +def test_existing_partitions_disable_swap_and_unmount_filesystems(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr(logger, '_path', tmp_path) + from archinstall.lib.disk.device_handler import DeviceHandler + + handler = DeviceHandler.__new__(DeviceHandler) + handler._devices = { + Path('/dev/sda'): Mock( + partition_infos=[ + Mock(path=Path('/dev/sda1'), fs_type=FilesystemType.EXT4), + Mock(path=Path('/dev/sda2'), fs_type=FilesystemType.LINUX_SWAP), + ] + ), + } + command = Mock() + command.return_value.output.return_value = b'/dev/sda2\n' + monkeypatch.setattr(utils, 'SysCommand', command) + unmount = Mock() + monkeypatch.setattr('archinstall.lib.disk.device_handler.umount', unmount) + + handler.umount_all_existing(Path('/dev/sda')) + + unmount.assert_called_once_with(Path('/dev/sda1'), recursive=True) + assert command.call_args_list == [call(SWAPON_QUERY), call(['swapoff', '/dev/sda2'])]