Compare commits
13 commits
012b10f7d5
...
2327d5388d
Author | SHA1 | Date | |
---|---|---|---|
2327d5388d | |||
094d66bc33 | |||
6fab5ce9b4 | |||
0d77038392 | |||
2c6bec253c | |||
eb0f862125 | |||
b05de437d0 | |||
313cc30aec | |||
149ed2dc3b | |||
e9d228ba2c | |||
3940d32195 | |||
2e00140e82 | |||
b9754b5dde |
10 changed files with 196 additions and 50 deletions
|
@ -5,7 +5,11 @@ packages: list[str] = []
|
|||
|
||||
|
||||
def need(command: str) -> None:
|
||||
"""Assert package dependency for a command."""
|
||||
"""Assert package dependency for a command.
|
||||
|
||||
:param command: name of the requested command
|
||||
:type command: str
|
||||
"""
|
||||
package: str | None
|
||||
match command:
|
||||
case "debootstrap":
|
||||
|
|
|
@ -1,16 +1,24 @@
|
|||
"""Wrap SquashFS commands."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from rwx import cmd, ps
|
||||
|
||||
cmd.need("mksquashfs")
|
||||
|
||||
|
||||
def mksquashfs(input_root: str, output_file: str) -> None:
|
||||
"""Make a SquashFS bootable image file."""
|
||||
def mksquashfs(input_root: Path, output_file: Path) -> None:
|
||||
"""Make a SquashFS bootable image file.
|
||||
|
||||
:param input_root: ?
|
||||
:type input_root: Path
|
||||
:param output_file: ?
|
||||
:type output_file: Path
|
||||
"""
|
||||
ps.run(
|
||||
"mksquashfs",
|
||||
input_root,
|
||||
output_file,
|
||||
str(input_root),
|
||||
str(output_file),
|
||||
"-comp",
|
||||
"zstd",
|
||||
"-Xcompression-level",
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
"""Wrap Debian commands."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from rwx import cmd, ps
|
||||
|
||||
cmd.need("debootstrap")
|
||||
|
@ -8,14 +10,22 @@ BOOTSTRAP_ARCHITECTURE = "amd64"
|
|||
BOOTSTRAP_VARIANT = "minbase"
|
||||
|
||||
|
||||
def bootstrap(root_path: str, suite: str, mirror_location: str) -> None:
|
||||
"""Boostrap a base operating filesystem."""
|
||||
command = [
|
||||
("debootstrap",),
|
||||
def bootstrap(root_path: Path, suite: str, mirror_location: str) -> None:
|
||||
"""Boostrap a base operating filesystem.
|
||||
|
||||
:param root_path: target output path
|
||||
:type root_path: Path
|
||||
:param suite: target distribution name
|
||||
:type suite: str
|
||||
:param mirror_location: source input repository
|
||||
:type mirror_location: str
|
||||
"""
|
||||
command = (
|
||||
"debootstrap",
|
||||
("--arch", BOOTSTRAP_ARCHITECTURE),
|
||||
("--variant", BOOTSTRAP_VARIANT),
|
||||
(suite,),
|
||||
(root_path,),
|
||||
(mirror_location,),
|
||||
]
|
||||
suite,
|
||||
str(root_path),
|
||||
mirror_location,
|
||||
)
|
||||
ps.run(*command)
|
||||
|
|
|
@ -9,67 +9,121 @@ from rwx import ps
|
|||
CHARSET = "UTF-8"
|
||||
|
||||
|
||||
def create_image(file_path: str, size_bytes: int) -> None:
|
||||
"""Create a virtual device image file."""
|
||||
def create_image(file_path: Path, size_bytes: int) -> None:
|
||||
"""Create a virtual device image file.
|
||||
|
||||
:param file_path: target image file
|
||||
:type file_path: Path
|
||||
:param size_bytes: virtual volume
|
||||
:type size_bytes: int
|
||||
"""
|
||||
ps.run(
|
||||
("qemu-img", "create"),
|
||||
("-f", "qcow2"),
|
||||
(file_path, str(size_bytes)),
|
||||
(str(file_path), str(size_bytes)),
|
||||
)
|
||||
|
||||
|
||||
def empty_file(path: Path) -> None:
|
||||
"""Empty the file at provided path."""
|
||||
"""Empty the file at provided path.
|
||||
|
||||
:param path: target file to empty
|
||||
:type path: Path
|
||||
"""
|
||||
write(path, "")
|
||||
|
||||
|
||||
def get_mount_uuid(path: str) -> str:
|
||||
"""Return the filesystem UUID of a mountpoint path."""
|
||||
def get_mount_uuid(path: Path) -> str:
|
||||
"""Return the filesystem UUID of a mountpoint path.
|
||||
|
||||
:param path: mountpoint path
|
||||
:type path: Path
|
||||
:rtype: str
|
||||
"""
|
||||
return ps.run_line(
|
||||
("findmnt",),
|
||||
("--noheadings",),
|
||||
"findmnt",
|
||||
"--noheadings",
|
||||
("--output", "UUID"),
|
||||
(path,),
|
||||
str(path),
|
||||
)
|
||||
|
||||
|
||||
def get_path_mount(path: str) -> str:
|
||||
"""Return the mountpoint path of an arbitrary path."""
|
||||
return ps.run_line(
|
||||
"stat",
|
||||
("--format", "%m"),
|
||||
path,
|
||||
def get_path_mount(path: Path) -> Path:
|
||||
"""Return the mountpoint path of an arbitrary path.
|
||||
|
||||
:param path: arbitrary path
|
||||
:type path: Path
|
||||
:rtype: Path
|
||||
"""
|
||||
return Path(
|
||||
ps.run_line(
|
||||
"stat",
|
||||
("--format", "%m"),
|
||||
str(path),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def get_path_uuid(path: str) -> str:
|
||||
"""Return the filesystem UUID of an arbitrary path."""
|
||||
def get_path_uuid(path: Path) -> str:
|
||||
"""Return the filesystem UUID of an arbitrary path.
|
||||
|
||||
:param path: arbitrary path
|
||||
:type path: Path
|
||||
:rtype: str
|
||||
"""
|
||||
return get_mount_uuid(get_path_mount(path))
|
||||
|
||||
|
||||
def make_directory(path: Path) -> None:
|
||||
"""Make a directory (and its parents) from a path."""
|
||||
"""Make a directory (and its parents) from a path.
|
||||
|
||||
:param path: directory to create
|
||||
:type path: Path
|
||||
"""
|
||||
path.mkdir(exist_ok=True, parents=True)
|
||||
|
||||
|
||||
def read_file_bytes(file_path: Path) -> bytes:
|
||||
"""Read whole file bytes."""
|
||||
"""Read whole file bytes.
|
||||
|
||||
:param file_path: source input file
|
||||
:type file_path: Path
|
||||
:rtype: bytes
|
||||
"""
|
||||
with file_path.open("br") as file_object:
|
||||
return file_object.read()
|
||||
|
||||
|
||||
def read_file_lines(file_path: Path, charset: str = CHARSET) -> list[str]:
|
||||
"""Read whole file lines."""
|
||||
"""Read whole file lines.
|
||||
|
||||
:param file_path: source input file
|
||||
:type file_path: Path
|
||||
:param charset: charset to use for decoding input
|
||||
:type charset: str
|
||||
:rtype: list[str]
|
||||
"""
|
||||
return read_file_text(file_path, charset).split(os.linesep)
|
||||
|
||||
|
||||
def read_file_text(file_path: Path, charset: str = CHARSET) -> str:
|
||||
"""Read whole file text."""
|
||||
"""Read whole file text.
|
||||
|
||||
:param file_path: source input file
|
||||
:type file_path: Path
|
||||
:param charset: charset to use for decoding input
|
||||
:type charset: str
|
||||
:rtype: str
|
||||
"""
|
||||
return read_file_bytes(file_path).decode(charset)
|
||||
|
||||
|
||||
def wipe(path: Path) -> None:
|
||||
"""Wipe provided path, whether directory or file."""
|
||||
"""Wipe provided path, whether directory or file.
|
||||
|
||||
:param path: target path
|
||||
:type path: Path
|
||||
"""
|
||||
try:
|
||||
shutil.rmtree(path)
|
||||
except NotADirectoryError:
|
||||
|
@ -79,6 +133,14 @@ def wipe(path: Path) -> None:
|
|||
|
||||
|
||||
def write(file_path: Path, text: str, charset: str = CHARSET) -> None:
|
||||
"""Write text into a file."""
|
||||
"""Write text into a file.
|
||||
|
||||
:param file_path: target file path
|
||||
:type file_path: Path
|
||||
:param text: content to write
|
||||
:type text: str
|
||||
:param charset: charset to use for encoding ouput
|
||||
:type charset: str
|
||||
"""
|
||||
with file_path.open(encoding=charset, mode="w") as file_object:
|
||||
file_object.write(text)
|
||||
|
|
|
@ -5,7 +5,12 @@ import sys
|
|||
|
||||
|
||||
def get_file_logger(name: str) -> logging.Logger:
|
||||
"""Return a file logger."""
|
||||
"""Return a file logger.
|
||||
|
||||
:param name: arbitrary name
|
||||
:type name: str
|
||||
:rtype: logging.Logger
|
||||
"""
|
||||
# formatter
|
||||
items = [
|
||||
"%(name)s: %(asctime)s",
|
||||
|
@ -27,7 +32,12 @@ def get_file_logger(name: str) -> logging.Logger:
|
|||
|
||||
|
||||
def get_stream_logger(level: int) -> logging.Logger:
|
||||
"""Return a stream logger."""
|
||||
"""Return a stream logger.
|
||||
|
||||
:param level: filtering level
|
||||
:type level: int
|
||||
:rtype: logging.Logger
|
||||
"""
|
||||
# handler
|
||||
out_handler = logging.StreamHandler(stream=sys.stdout)
|
||||
out_handler.setLevel(level)
|
||||
|
|
|
@ -8,7 +8,12 @@ from .debian import Debian
|
|||
|
||||
|
||||
def from_path(path: Path) -> OS:
|
||||
"""Initialize from an already existing path."""
|
||||
"""Initialize from an already existing path.
|
||||
|
||||
:param path: source root directory
|
||||
:type path: Path
|
||||
:rtype: OS
|
||||
"""
|
||||
return Debian(path)
|
||||
|
||||
|
||||
|
|
|
@ -10,10 +10,17 @@ class OS(Class, ABC):
|
|||
"""Operating System."""
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
"""Set root."""
|
||||
"""Set root.
|
||||
|
||||
:param path: root directory
|
||||
:type path: Path
|
||||
"""
|
||||
self.root = path
|
||||
self.name = self.get_name()
|
||||
|
||||
@abstractmethod
|
||||
def get_name(self) -> str:
|
||||
"""Return mandatory name."""
|
||||
"""Return mandatory name.
|
||||
|
||||
:rtype: str
|
||||
"""
|
||||
|
|
|
@ -16,8 +16,14 @@ class PM(Class, ABC):
|
|||
|
||||
@abstractmethod
|
||||
def get_clean_command(self) -> Command:
|
||||
"""Command to clean packages cache."""
|
||||
"""Command to clean packages cache.
|
||||
|
||||
:rtype: Command
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_install_command(self) -> Command:
|
||||
"""Command to install package(s)."""
|
||||
"""Command to install package(s).
|
||||
|
||||
:rtype: Command
|
||||
"""
|
||||
|
|
|
@ -8,9 +8,15 @@ class APT(PM):
|
|||
"""Advanced Package Tool."""
|
||||
|
||||
def get_clean_command(self) -> Command:
|
||||
"""Return clean command."""
|
||||
"""Return clean command.
|
||||
|
||||
:rtype: Command
|
||||
"""
|
||||
return Command()
|
||||
|
||||
def get_install_command(self) -> Command:
|
||||
"""Return install command."""
|
||||
"""Return install command.
|
||||
|
||||
:rtype: Command
|
||||
"""
|
||||
return Command()
|
||||
|
|
|
@ -9,13 +9,22 @@ class Command(Class):
|
|||
"""Command to run."""
|
||||
|
||||
def __init__(self, *arguments: str | tuple[str, ...]) -> None:
|
||||
"""Set raw & flat arguments."""
|
||||
"""Set raw & flat arguments.
|
||||
|
||||
:param *arguments: single argument or grouped ones
|
||||
:type *arguments: str | tuple[str, ...]
|
||||
"""
|
||||
self.raw = arguments
|
||||
self.flat: list[str] = []
|
||||
|
||||
|
||||
def get_tuples_args(*items: str | tuple[str, ...]) -> list[str]:
|
||||
"""Turn arguments tuples into an arguments list."""
|
||||
"""Turn arguments tuples into an arguments list.
|
||||
|
||||
:param *items: single item or grouped ones
|
||||
:type *items: str | tuple[str, ...]
|
||||
:rtype: list[str]
|
||||
"""
|
||||
args: list[str] = []
|
||||
for item in items:
|
||||
match item:
|
||||
|
@ -27,14 +36,26 @@ def get_tuples_args(*items: str | tuple[str, ...]) -> list[str]:
|
|||
|
||||
|
||||
def run(*items: str | tuple[str, ...]) -> subprocess.CompletedProcess:
|
||||
"""Run from a list of arguments tuples."""
|
||||
"""Run from a list of arguments tuples.
|
||||
|
||||
:param *items: single item or grouped ones
|
||||
:type *items: str | tuple[str, ...]
|
||||
:rtype: subprocess.CompletedProcess
|
||||
"""
|
||||
return subprocess.run(
|
||||
get_tuples_args(*items), capture_output=False, check=True
|
||||
)
|
||||
|
||||
|
||||
def run_line(*items: str | tuple[str, ...], charset: str = txt.CHARSET) -> str:
|
||||
"""Run and return output line."""
|
||||
"""Run and return output line.
|
||||
|
||||
:param *items: single item or grouped ones
|
||||
:type *items: str | tuple[str, ...]
|
||||
:param charset: charset to use for decoding binary output
|
||||
:type charset: str
|
||||
:rtype: str
|
||||
"""
|
||||
line, *_ = run_lines(*items, charset=charset)
|
||||
return line
|
||||
|
||||
|
@ -42,7 +63,14 @@ def run_line(*items: str | tuple[str, ...], charset: str = txt.CHARSET) -> str:
|
|||
def run_lines(
|
||||
*items: str | tuple[str, ...], charset: str = txt.CHARSET
|
||||
) -> list[str]:
|
||||
"""Run and return output lines."""
|
||||
"""Run and return output lines.
|
||||
|
||||
:param *items: single item or grouped ones
|
||||
:type *items: str | tuple[str, ...]
|
||||
:param charset: charset to use for decoding binary output
|
||||
:type charset: str
|
||||
:rtype: list[str]
|
||||
"""
|
||||
process = subprocess.run(
|
||||
get_tuples_args(*items), capture_output=True, check=True
|
||||
)
|
||||
|
|
Loading…
Reference in a new issue