# HG changeset patch # User Alain Leufroy # Date 1614466901 -3600 # Sun Feb 28 00:01:41 2021 +0100 # Node ID 5818de7900a4ca91257bd3a4b90948fd9e89d51e # Parent ce9bbc83cd90ac66f26c76bd784247c8e47b1406 coding style diff --git a/overlayctl b/overlayctl --- a/overlayctl +++ b/overlayctl @@ -15,7 +15,6 @@ we stack the upper dir of lowered overlays instead of their mount point. """ -import io import json import logging import operator @@ -25,8 +24,7 @@ import sys from collections import OrderedDict, defaultdict from functools import reduce -from glob import glob -from itertools import chain, dropwhile, filterfalse +from itertools import chain from pathlib import Path from subprocess import PIPE, Popen from tempfile import NamedTemporaryFile @@ -38,7 +36,9 @@ init() except ImportError: - Fore = Back = Style = type('mock', (), {'__getattribute__': lambda *a: ''})() + Fore = Back = Style = type( + 'mock', (), + {'__getattribute__': lambda * a: ''})() logger = logging.getLogger(__name__) @@ -51,9 +51,6 @@ INFOSDIR = OVERLAYDIR / 'info' UNITSDIR = Path(os.environ.get('UNITSDIR', Path('/') / 'etc' / 'systemd' / 'system')) -for path in (MOUNTDIR, OVERLAYDIR, UPPERSDIR, WORKSDIR, INFOSDIR, UNITSDIR): - path.mkdir(parents=True, exist_ok=True) - MOUNTTMPL = ''' [Unit] @@ -96,8 +93,7 @@ """ if os.path.sep in name: return Path(name) - else: - return MOUNTDIR / name + return MOUNTDIR / name def _name2unit(name, _cache={}): @@ -130,8 +126,7 @@ upper = _name2upperdir(name) if upper.exists(): return upper - else: - return _name2mountdir(name) + return _name2mountdir(name) def _name2infopath(name): @@ -236,7 +231,7 @@ if not ordered: return infos infos = {info['name']: info for info in infos} - lowers = {name for name in chain(*[info['lowers'] for info in infos.values()])} + lowers = set(chain(*[info['lowers'] for info in infos.values()])) heads = list(set(infos) - lowers) return ( infos[name] for name in _get_linearized_lowerdirs(None, heads) if name in infos @@ -308,7 +303,6 @@ info['mountpath'].unlink() - def c3_merge(graph): """Merge overlay inheritence sequences preserving order in initial sequences. This is the merge function as described for C3, see: @@ -318,7 +312,10 @@ result = ( OrderedDict() ) # ordered uniq names (there is no OrderedSet in stdlib for now) - isheadclean = lambda current: all(current[0] not in other[1:] for other in graph) + + def isheadclean(current): + return all(current[0] not in other[1:] for other in graph) + while any(graph): branches = list(filter(isheadclean, filter(None, graph))) if not branches: @@ -374,7 +371,7 @@ sh.systemctl.start(info['automountpath'].name).wait() logger.debug("Systemd %s.automount unit started", info['unitname']) else: - logger.warning("Use `overlayctl start {name}` to start the overlay.".format(**info)) + logger.warning("Use `overlayctl start %s` to start the overlay.", info['name']) if permanent: sh.systemctl.enable(info['automountpath'].name).wait() logger.debug("Systemd %s.automount unit enabled", info['unitname']) @@ -443,7 +440,7 @@ def starter(name, preserve_others=False, stop_others=False, permanent=False): info = _read_info(name) if _is_automount_active(info): - logger.info('{name} is already started.'.format(**info)) + logger.info('%s is already started.', info['name']) return if not preserve_others: to_stop = [] @@ -453,8 +450,8 @@ if to_stop: if not stop_others: logger.error( - 'You should prefere to stop the following overlays: \n\n%s\n' - % ', '.join(i['name'] for i in to_stop)) + 'You should prefere to stop the following overlays: \n\n%s\n', + ', '.join(i['name'] for i in to_stop)) logger.info( 'You may want to use `--stop-others`. ' 'Or see --help for more options.') @@ -545,7 +542,8 @@ logger.error('New name "%s" already exists.', new) return superseeders = [x for x in existings if old in x['lowers']] - logger.debug('%s overlays affected: %s', len(superseeders), ', '.join(x['name'] for x in superseeders)) + logger.debug('%s overlays affected: %s', len(superseeders), + ', '.join(x['name'] for x in superseeders)) mounted = [info['name'] for info in superseeders if _is_mount_active(info)] if mounted: logger.error('The following units must be stopped: %s', ', '.join(mounted)) @@ -580,8 +578,12 @@ def main(): """Main entry point for the CLI""" + from argparse import ArgumentParser + for path in (MOUNTDIR, OVERLAYDIR, UPPERSDIR, WORKSDIR, INFOSDIR, UNITSDIR): + path.mkdir(parents=True, exist_ok=True) + parser = ArgumentParser(description="Manage images as overlays for machinectl.") parser.add_argument( '--traceback', @@ -773,7 +775,9 @@ edit = subparser.add_parser( 'edit', help="edit existing overlay", - description="edit an overlay created by this tool. An editor is started if --add nor --remove provided.", + description=( + "edit an overlay created by this tool. " + "An editor is started if --add nor --remove provided."), ) edit.add_argument('-p', '--prepand', action='store_true', help="prepand the following names", ) @@ -801,7 +805,9 @@ move = subparser.add_parser( 'move', help="move existing overlay", - description="move an overlay created by this tool. Other overlays that depends on it are updated", + description=( + "move an overlay created by this tool. " + "Other overlays that depends on it are updated"), ) move.add_argument('old', metavar='OLD', help="existing overlay name.") move.add_argument('new', metavar='NEW', help="New overlay name.") @@ -818,7 +824,7 @@ description=( "Start the systemctl automount of the overlay. " "The mount point will be automcally mounter on first assess (e.g. `ls {MOUNTDIR}`). " - "In order to prevent unexpected behaviours, descendant and ascendants overlays should be stopped." + "In order to prevent strang behaviours only one overlay shall be started on a branch." ), ) start.add_argument( @@ -847,6 +853,7 @@ action='store_true', default=False, help='make changes permanent on reboot' ) + def _start(args): starter(args.mountdir, args.keep_others, args.stop_others, args.permanent) @@ -873,12 +880,12 @@ action='store_true', default=False, help='make changes permanent on reboot' ) + def _stop(args): stoper(args.mountdir, args.permanent) stop.set_defaults(func=_stop) - args = parser.parse_args() _setup_logger(args.loglevel) if not getattr(args, 'func', None):