root/ekernel
root/ekernel
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

improved mounting

Browse Source
This commit is contained in:
Florian Schroegendorfer 2024-06-04 03:39:10 +02:00
parent 0e9d9f4727
commit ed26f89aef
Signed by: root
GPG Key ID: 17625E28D4D6E339
3 changed files with 49 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
ekernel.py
View File

@ -112,6 +112,11 @@ def cli (f):
@functools.wraps(f)
def handler (argv=sys.argv[1:]):
try:
# dirty hack: set custom esp before mounting
try:
i = argv.index("-e")
Kernel.esp = pathlib.Path(argv[i + 1])
except ValueError: pass
r = f(argv)
return 0 if r is None else r
except Exception as e:
@ -119,29 +124,30 @@ def cli (f):
sys.exit(1)
return handler
def mount (path):
def mount (f):
"""Decorator ensuring a given path is mounted."""
def wrapper (f):
@functools.wraps(f)
def mounter (*args, **kwargs):
mounted = False
esp = Kernel.esp.parents[-2]
if not esp.exists() or mount.force:
try:
subprocess.run(
["mount", path],
["mount", str(esp)],
capture_output=True,
check=True
)
mounted = True
except subprocess.CalledProcessError as e:
msg = e.stderr.decode().strip()
if f"already mounted on {path}" not in msg:
if f"already mounted on {esp}" not in msg:
raise RuntimeError(e.stderr.decode().splitlines()[0])
r = f(*args, **kwargs)
if mounted:
subprocess.run(["umount", path], check=True)
subprocess.run(["umount", str(esp)], check=True)
return r
mount.force = False
return mounter
return wrapper
@cli
def configure (argv):
@ -341,7 +347,7 @@ def build (argv):
subprocess.run(["make", "modules_install"], check=True)
@cli
@mount("/boot")
@mount
def install (argv):
"""
Install a kernel.
@ -404,7 +410,7 @@ def install (argv):
help="be quiet"
)
args = parser.parse_args(argv)
if args.esp: Kernel.esp = args.esp
if args.esp: Kernel.esp = args.esp # redundant
kernel = Kernel(args.src)
out.quiet = args.quiet
@ -435,7 +441,7 @@ def install (argv):
subprocess.run(["emerge", "@module-rebuild"], check=True)
@cli
@mount("/boot")
@mount
def clean (argv):
"""
Remove unused kernel leftovers.
@ -496,7 +502,7 @@ def clean (argv):
help="be quiet"
)
args = parser.parse_args(argv)
if args.esp: Kernel.esp = args.esp
if args.esp: Kernel.esp = args.esp # redundant
out.quiet = args.quiet
if args.keep < 0:
raise ValueError("invalid int value: must be greater equal zero")
@ -512,9 +518,12 @@ def clean (argv):
# dry run
if args.dry:
if leftovers:
out.einfo("the following kernels will be removed:")
for k in leftovers:
print(f" {colorize('BAD', '✗')} {k.src.name}")
else:
out.einfo("nothing to see here")
return
# run depclean
@ -725,6 +734,7 @@ def commit (argv):
if args.msg: msg.write(f"\n\n{args.msg}")
# print changes
if removals or config_changed:
out.einfo("changes to be committed:")
for l in removals:
out.print(f" {colorize('QAWARN', '-')} {out.hilite(l)}")
@ -732,6 +742,7 @@ def commit (argv):
out.print(f" {colorize('INFO', '+')} {out.hilite(kernel.config)}")
# print message
if msg:
out.einfo("commit message:")
for l in msg.getvalue().splitlines():
out.print(f" {l}" if l else "")
@ -801,7 +812,7 @@ def update (argv):
)
args = parser.parse_args(argv)
args.jobs = ["-j", str(args.jobs)] if args.jobs else []
args.esp = ["-e", str(args.esp)] if args.esp else []
args.esp = ["-e", str(args.esp)] if args.esp else [] # redundant
args.src = ["-s", str(args.src)] if args.src else []
args.keep = ["-k", str(args.keep)] if args.keep is not None else []
args.msg = ["-m", args.msg] if args.msg else []

6
tests/test_clean.py
View File

@ -31,6 +31,8 @@ class Tests (unittest.TestCase):
k.modules.mkdir(parents=True)
if not k.efi.exists():
k.efi.touch()
# force mounting
ekernel.mount.force = True
# start interceptor
self.interceptor = Interceptor()
self.interceptor.add(subprocess.run, call=True)
@ -46,7 +48,7 @@ class Tests (unittest.TestCase):
# mount /boot
tracer, (args, kwargs) = next(trace_it)
self.assertEqual(tracer.name, "subprocess.run")
self.assertEqual(args, (["mount", "/boot"],))
self.assertEqual(args, (["mount", "/tmp"],))
self.assertEqual(kwargs, {"capture_output": True, "check": True})
# emerge -cq gentoo-sources
tracer, (args, kwargs) = next(trace_it)
@ -64,7 +66,7 @@ class Tests (unittest.TestCase):
# umount /boot
tracer, (args, kwargs) = next(trace_it)
self.assertEqual(tracer.name, "subprocess.run")
self.assertEqual(args, (["umount", "/boot"],))
self.assertEqual(args, (["umount", "/tmp"],))
self.assertEqual(kwargs, {"check": True})
def test_clean (self):

16
tests/test_install.py
View File

@ -14,6 +14,12 @@ def run (*argv):
class Tests (unittest.TestCase):
def setUp (self):
# setup test environment
data.setup()
self.kernel = Kernel.latest()
self.kernel.bzImage.touch()
# force mounting
ekernel.mount.force = True
# start interceptor
self.interceptor = Interceptor()
def run (tracer, *args, **kwargs):
@ -22,10 +28,6 @@ class Tests (unittest.TestCase):
data.linux.symlink_to(self.kernel.src.name)
self.interceptor.add(subprocess.run, call=run)
self.interceptor.start()
# setup test environment
data.setup()
self.kernel = Kernel.latest()
self.kernel.bzImage.touch()
def tearDown (self):
# stop interceptor
@ -36,7 +38,7 @@ class Tests (unittest.TestCase):
# mount /boot
tracer, (args, kwargs) = next(trace_it)
self.assertEqual(tracer.name, "subprocess.run")
self.assertEqual(args, (["mount", "/boot"],))
self.assertEqual(args, (["mount", "/tmp"],))
self.assertEqual(kwargs, {"capture_output": True, "check": True})
# eselect kernel set <name>
tracer, (args, kwargs) = next(trace_it)
@ -68,6 +70,10 @@ class Tests (unittest.TestCase):
self.assertEqual(run("-q", "-e", str(esp)), 0)
self.check_install()
def test_install_esp_missing (self):
with self.assertRaises(SystemExit):
run("-q", "-e", str(data.root / "boot/EFI/linux"))
def test_install_source (self):
self.kernel = Kernel.current()
self.assertEqual(run("-q", "-s", str(data.current)), 0)