Files
create-linux-swapfile/tests/unit/test_core.py
T
kevinveenbirkenbachandClaude Opus 5.5 91e1eff8a6 fix(btrfs): allocate the swapfile with fallocate instead of dd
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>
2026-10-23 10:36:04 +02:00

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()