mirror of
https://github.com/kevinveenbirkenbach/create-linux-swapfile.git
synced 2026-08-16 20:02:47 +00:00
feat: package as pip-installable distribution
Turns the loose main.py into the 'swap-forge' distribution with a src layout, shipping both the swap-forge and the older swafo command. A failing swap command no longer produces a traceback: CalledProcessError is caught and reported as "'mkswap' failed with exit code 1" at exit 1, a missing binary as exit 127. Required binaries are declared per filesystem path so an ext4 host is never asked for btrfs tooling. The fstab entry is still written last, after swapon, so a failed activation leaves /etc/fstab untouched. The e2e suite proves that byte for byte across a create, a same-size rerun and a resize cycle, with real fallocate, chmod and file sizes. Adds unit, integration and container-based e2e tests, ruff and markdown/mermaid linting, CodeQL and Dependabot, and pins the core metadata to 2.4 so twine accepts the artifacts. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
0
src/swap_forge/__init__.py
Normal file
0
src/swap_forge/__init__.py
Normal file
6
src/swap_forge/__main__.py
Normal file
6
src/swap_forge/__main__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
import sys
|
||||
|
||||
from swap_forge.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
74
src/swap_forge/cli.py
Normal file
74
src/swap_forge/cli.py
Normal file
@@ -0,0 +1,74 @@
|
||||
import argparse
|
||||
import sys
|
||||
from subprocess import CalledProcessError
|
||||
|
||||
from swap_forge.core import (
|
||||
EXIT_COMMAND_FAILED,
|
||||
EXIT_MISSING_BINARY,
|
||||
MissingBinaryError,
|
||||
create_swap,
|
||||
detect_root_fs,
|
||||
get_swap_size,
|
||||
parse_size_to_mib,
|
||||
remove_swap,
|
||||
swap_path_for,
|
||||
)
|
||||
|
||||
|
||||
def build_parser():
|
||||
"""Return the argument parser for the swap-forge command."""
|
||||
parser = argparse.ArgumentParser(description="SwapForge Python edition")
|
||||
parser.add_argument("size", help="Swapfile size (e.g. 2048M or 64G)")
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
"""Create or resize the system swapfile.
|
||||
|
||||
Args:
|
||||
argv: Argument list, defaults to sys.argv[1:].
|
||||
|
||||
Returns:
|
||||
Process exit code.
|
||||
"""
|
||||
args = build_parser().parse_args(argv)
|
||||
|
||||
try:
|
||||
new_size = parse_size_to_mib(args.size)
|
||||
except ValueError as error:
|
||||
print(f"Error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
fs = detect_root_fs()
|
||||
path = swap_path_for(fs)
|
||||
old_size = get_swap_size(path)
|
||||
|
||||
if old_size == new_size and old_size > 0:
|
||||
print(
|
||||
f"Swapfile {path} already exists with correct size "
|
||||
f"({old_size} MiB). Skipping."
|
||||
)
|
||||
return 0
|
||||
|
||||
if old_size > 0:
|
||||
print(
|
||||
f"Existing swapfile {path} has size {old_size} MiB, "
|
||||
f"expected {new_size} MiB. Recreating..."
|
||||
)
|
||||
remove_swap(path)
|
||||
|
||||
create_swap(path, new_size, fs)
|
||||
except MissingBinaryError as error:
|
||||
print(f"Error: {error}", file=sys.stderr)
|
||||
return EXIT_MISSING_BINARY
|
||||
except CalledProcessError as error:
|
||||
command = error.cmd[0] if isinstance(error.cmd, list | tuple) else error.cmd
|
||||
print(
|
||||
f"Error: '{command}' failed with exit code {error.returncode}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return EXIT_COMMAND_FAILED
|
||||
|
||||
print("Done.")
|
||||
return 0
|
||||
160
src/swap_forge/core.py
Normal file
160
src/swap_forge/core.py
Normal file
@@ -0,0 +1,160 @@
|
||||
import re
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
FSTAB_FILE = Path("/etc/fstab")
|
||||
BTRFS_SWAP_DIR = Path("/var/swap")
|
||||
BTRFS_SWAP_PATH = BTRFS_SWAP_DIR / "swapfile"
|
||||
DEFAULT_SWAP_PATH = Path("/swapfile")
|
||||
|
||||
SIZE_PATTERN = re.compile(r"^(\d+)([gGmM]?[bB]?)?$")
|
||||
MIB_PER_GIB = 1024
|
||||
|
||||
REQUIRED_BINARIES = ("findmnt", "swapoff", "chmod", "mkswap", "swapon")
|
||||
BTRFS_BINARIES = ("chattr", "btrfs", "dd")
|
||||
NON_BTRFS_BINARIES = ("fallocate",)
|
||||
EXIT_MISSING_BINARY = 127
|
||||
EXIT_COMMAND_FAILED = 1
|
||||
|
||||
|
||||
class MissingBinaryError(RuntimeError):
|
||||
"""A required external command is not installed.
|
||||
|
||||
Args:
|
||||
binary: Name of the missing executable.
|
||||
"""
|
||||
|
||||
def __init__(self, binary):
|
||||
super().__init__(f"required command '{binary}' is not installed")
|
||||
self.binary = binary
|
||||
|
||||
|
||||
def run(cmd, check=True, capture=False):
|
||||
"""Run a command and return the completed process.
|
||||
|
||||
Args:
|
||||
cmd: Command argument list.
|
||||
check: Raise CalledProcessError when the command fails.
|
||||
capture: Capture stdout and stderr as text.
|
||||
"""
|
||||
try:
|
||||
if capture:
|
||||
return subprocess.run(cmd, check=check, capture_output=True, text=True)
|
||||
return subprocess.run(cmd, check=check)
|
||||
except FileNotFoundError as error:
|
||||
raise MissingBinaryError(cmd[0]) from error
|
||||
|
||||
|
||||
def detect_root_fs():
|
||||
"""Return the filesystem type mounted at /."""
|
||||
return run(["findmnt", "-no", "FSTYPE", "/"], capture=True).stdout.strip()
|
||||
|
||||
|
||||
def parse_size_to_mib(size):
|
||||
"""Convert a size such as '64G' or '2048M' into MiB.
|
||||
|
||||
Args:
|
||||
size: Size string with an optional G/M unit, defaulting to MiB.
|
||||
"""
|
||||
match = SIZE_PATTERN.match(size.strip())
|
||||
if not match:
|
||||
raise ValueError(f"Invalid size format: {size}")
|
||||
|
||||
number, unit = match.groups()
|
||||
unit = (unit or "M").lower()
|
||||
|
||||
if unit in ("g", "gb"):
|
||||
return int(number) * MIB_PER_GIB
|
||||
if unit in ("m", "mb"):
|
||||
return int(number)
|
||||
raise ValueError(f"Unsupported unit: {unit}")
|
||||
|
||||
|
||||
def get_swap_size(path):
|
||||
"""Return the size of the swapfile in MiB, or 0 when it does not exist.
|
||||
|
||||
Args:
|
||||
path: Path of the swapfile.
|
||||
"""
|
||||
if not path.exists():
|
||||
return 0
|
||||
return path.stat().st_size // (1024 * 1024)
|
||||
|
||||
|
||||
def swap_fstab_line(path):
|
||||
"""Return the fstab line registering the given swapfile."""
|
||||
return f"{path} none swap sw 0 0"
|
||||
|
||||
|
||||
def remove_swap(path, fstab=FSTAB_FILE):
|
||||
"""Disable the swapfile and drop it plus its fstab entry.
|
||||
|
||||
Args:
|
||||
path: Path of the swapfile.
|
||||
fstab: Path of the fstab file to clean up.
|
||||
"""
|
||||
run(["swapoff", str(path)], check=False)
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
|
||||
if not fstab.exists():
|
||||
return
|
||||
|
||||
kept = [
|
||||
line
|
||||
for line in fstab.read_text().splitlines()
|
||||
if not re.search(rf"^{re.escape(str(path))}\s+none\s+swap", line)
|
||||
]
|
||||
fstab.write_text("\n".join(kept) + "\n")
|
||||
|
||||
|
||||
def create_swap(path, size_mib, fs, fstab=FSTAB_FILE):
|
||||
"""Create, activate and register a swapfile.
|
||||
|
||||
Args:
|
||||
path: Path of the swapfile.
|
||||
size_mib: Desired size in MiB.
|
||||
fs: Filesystem type of the root mount.
|
||||
fstab: Path of the fstab file to append to.
|
||||
"""
|
||||
if fs == "btrfs":
|
||||
print(f"Creating {size_mib} MiB swapfile on btrfs at {path}")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
run(["chattr", "+C", str(path.parent)], check=False)
|
||||
run(
|
||||
[
|
||||
"btrfs",
|
||||
"property",
|
||||
"set",
|
||||
"-ts",
|
||||
str(path.parent),
|
||||
"compression",
|
||||
"none",
|
||||
],
|
||||
check=False,
|
||||
)
|
||||
run(
|
||||
[
|
||||
"dd",
|
||||
"if=/dev/zero",
|
||||
f"of={path}",
|
||||
"bs=1M",
|
||||
f"count={size_mib}",
|
||||
"status=progress",
|
||||
]
|
||||
)
|
||||
else:
|
||||
print(f"Creating {size_mib} MiB swapfile on {fs} at {path}")
|
||||
run(["fallocate", "-l", f"{size_mib}M", str(path)])
|
||||
|
||||
run(["chmod", "600", str(path)])
|
||||
run(["mkswap", str(path)])
|
||||
run(["swapon", str(path)])
|
||||
|
||||
with fstab.open("a") as handle:
|
||||
handle.write(swap_fstab_line(path) + "\n")
|
||||
|
||||
|
||||
def swap_path_for(fs):
|
||||
"""Return the swapfile path matching the given root filesystem type."""
|
||||
return BTRFS_SWAP_PATH if fs == "btrfs" else DEFAULT_SWAP_PATH
|
||||
Reference in New Issue
Block a user