@@ 15,7 15,6 @@ To get around OverlayFS limitation
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 shutil
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 @@ try:
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 @@ WORKSDIR = OVERLAYDIR / 'work'
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 @@ def _name2mountdir(name):
"""
if os.path.sep in name:
return Path(name)
- else:
- return MOUNTDIR / name
+ return MOUNTDIR / name
def _name2unit(name, _cache={}):
@@ 130,8 126,7 @@ def _name2lowerdir(name):
upper = _name2upperdir(name)
if upper.exists():
return upper
- else:
- return _name2mountdir(name)
+ return _name2mountdir(name)
def _name2infopath(name):
@@ 236,7 231,7 @@ def _iter_overlays_info(ordered=None):
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 @@ def _remove_units(info):
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 @@ def c3_merge(graph):
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 @@ def creater(name, lowers, start, permane
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 editer(name, appended=None, prepande
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 @@ def starter(name, preserve_others=False,
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 @@ def deplacer(old, new):
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 _setup_logger(level):
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 @@ def main():
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 @@ def main():
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 @@ def main():
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 @@ def main():
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 @@ def main():
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):