mirror of
https://github.com/kevinveenbirkenbach/create-linux-swapfile.git
synced 2026-10-25 15:35:51 +00:00
On a btrfs root the swapfile was written with dd from /dev/zero, so a swapfile the size of the RAM cost tens of GiB of writes and stalled every other writer on the filesystem. A caller that bounds the run, such as an async Ansible task, killed dd halfway, and the next run found the wrong size and started over. btrfs accepts a preallocated NOCOW swapfile and the swap directory already carries +C, so every filesystem now allocates with fallocate. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
173 lines
5.8 KiB
Python
173 lines
5.8 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from swap_forge.core import (
|
|
BTRFS_BINARIES,
|
|
BTRFS_SWAP_PATH,
|
|
DEFAULT_SWAP_PATH,
|
|
REQUIRED_BINARIES,
|
|
MissingBinaryError,
|
|
create_swap,
|
|
get_swap_size,
|
|
parse_size_to_mib,
|
|
remove_swap,
|
|
run,
|
|
swap_fstab_line,
|
|
swap_path_for,
|
|
)
|
|
|
|
|
|
class TestRequiredBinaries(unittest.TestCase):
|
|
def test_declares_every_external_command_the_package_calls(self):
|
|
declared = set(REQUIRED_BINARIES) | set(BTRFS_BINARIES)
|
|
|
|
self.assertEqual(
|
|
declared,
|
|
{
|
|
"findmnt",
|
|
"swapoff",
|
|
"fallocate",
|
|
"chmod",
|
|
"mkswap",
|
|
"swapon",
|
|
"chattr",
|
|
"btrfs",
|
|
},
|
|
)
|
|
|
|
def test_filesystem_specific_binaries_are_not_declared_as_always_required(self):
|
|
self.assertFalse(set(REQUIRED_BINARIES) & set(BTRFS_BINARIES))
|
|
|
|
def test_missing_binary_is_reported_with_its_name(self):
|
|
with mock.patch("swap_forge.core.subprocess.run") as subprocess_run:
|
|
subprocess_run.side_effect = FileNotFoundError
|
|
|
|
with self.assertRaises(MissingBinaryError) as caught:
|
|
run(["mkswap", "/swapfile"])
|
|
|
|
self.assertEqual(caught.exception.binary, "mkswap")
|
|
|
|
|
|
class TestParseSizeToMib(unittest.TestCase):
|
|
def test_gibibytes(self):
|
|
for value in ("64G", "64g", "64GB", " 64G "):
|
|
with self.subTest(value=value):
|
|
self.assertEqual(parse_size_to_mib(value), 65536)
|
|
|
|
def test_mebibytes(self):
|
|
for value in ("2048M", "2048m", "2048MB"):
|
|
with self.subTest(value=value):
|
|
self.assertEqual(parse_size_to_mib(value), 2048)
|
|
|
|
def test_bare_number_is_mebibytes(self):
|
|
self.assertEqual(parse_size_to_mib("512"), 512)
|
|
|
|
def test_invalid_format(self):
|
|
for value in ("", "abc", "64T", "-5G", "6.5G"):
|
|
with self.subTest(value=value), self.assertRaises(ValueError):
|
|
parse_size_to_mib(value)
|
|
|
|
def test_unsupported_unit(self):
|
|
with self.assertRaises(ValueError):
|
|
parse_size_to_mib("64b")
|
|
|
|
|
|
class TestSwapPathFor(unittest.TestCase):
|
|
def test_btrfs_uses_the_dedicated_subvolume_path(self):
|
|
self.assertEqual(swap_path_for("btrfs"), BTRFS_SWAP_PATH)
|
|
|
|
def test_other_filesystems_use_the_root_swapfile(self):
|
|
self.assertEqual(swap_path_for("ext4"), DEFAULT_SWAP_PATH)
|
|
|
|
|
|
class TestGetSwapSize(unittest.TestCase):
|
|
def test_missing_file_is_zero(self):
|
|
self.assertEqual(get_swap_size(Path("/nonexistent/swapfile")), 0)
|
|
|
|
def test_reports_size_in_mebibytes(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
path = Path(base) / "swapfile"
|
|
path.write_bytes(b"\0" * (3 * 1024 * 1024))
|
|
|
|
self.assertEqual(get_swap_size(path), 3)
|
|
|
|
|
|
class TestSwapFstabLine(unittest.TestCase):
|
|
def test_line_shape(self):
|
|
self.assertEqual(
|
|
swap_fstab_line(Path("/swapfile")), "/swapfile none swap sw 0 0"
|
|
)
|
|
|
|
|
|
class TestRemoveSwap(unittest.TestCase):
|
|
def test_drops_file_and_fstab_entry(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
swapfile = Path(base) / "swapfile"
|
|
swapfile.write_bytes(b"\0")
|
|
fstab = Path(base) / "fstab"
|
|
fstab.write_text(
|
|
f"UUID=abc / ext4 defaults 0 1\n{swapfile} none swap sw 0 0\n"
|
|
)
|
|
|
|
with mock.patch("swap_forge.core.run"):
|
|
remove_swap(swapfile, fstab=fstab)
|
|
|
|
self.assertFalse(swapfile.exists())
|
|
self.assertEqual(fstab.read_text(), "UUID=abc / ext4 defaults 0 1\n")
|
|
|
|
def test_keeps_unrelated_fstab_entries(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
swapfile = Path(base) / "swapfile"
|
|
fstab = Path(base) / "fstab"
|
|
fstab.write_text("/other none swap sw 0 0\n")
|
|
|
|
with mock.patch("swap_forge.core.run"):
|
|
remove_swap(swapfile, fstab=fstab)
|
|
|
|
self.assertEqual(fstab.read_text(), "/other none swap sw 0 0\n")
|
|
|
|
def test_missing_fstab_is_tolerated(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
with mock.patch("swap_forge.core.run"):
|
|
remove_swap(Path(base) / "swapfile", fstab=Path(base) / "fstab")
|
|
|
|
|
|
class TestCreateSwap(unittest.TestCase):
|
|
def test_non_btrfs_uses_fallocate_and_appends_to_fstab(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
swapfile = Path(base) / "swapfile"
|
|
fstab = Path(base) / "fstab"
|
|
fstab.write_text("")
|
|
|
|
with mock.patch("swap_forge.core.run") as run:
|
|
create_swap(swapfile, 2048, "ext4", fstab=fstab)
|
|
|
|
commands = [call.args[0][0] for call in run.call_args_list]
|
|
self.assertEqual(commands, ["fallocate", "chmod", "mkswap", "swapon"])
|
|
self.assertEqual(fstab.read_text(), f"{swapfile} none swap sw 0 0\n")
|
|
|
|
def test_btrfs_disables_copy_on_write_before_fallocate(self):
|
|
with tempfile.TemporaryDirectory() as base:
|
|
swapfile = Path(base) / "swap" / "swapfile"
|
|
fstab = Path(base) / "fstab"
|
|
fstab.write_text("")
|
|
|
|
with mock.patch("swap_forge.core.run") as run:
|
|
create_swap(swapfile, 1024, "btrfs", fstab=fstab)
|
|
|
|
commands = [call.args[0][0] for call in run.call_args_list]
|
|
self.assertEqual(
|
|
commands, ["chattr", "btrfs", "fallocate", "chmod", "mkswap", "swapon"]
|
|
)
|
|
self.assertEqual(
|
|
run.call_args_list[2].args[0],
|
|
["fallocate", "-l", "1024M", str(swapfile)],
|
|
)
|
|
self.assertTrue(swapfile.parent.is_dir())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|