@@ 21,11 21,12 @@ import os
import re
import shutil
import sys
+from argparse import ArgumentParser
from collections import OrderedDict, defaultdict
from configparser import ConfigParser
from contextlib import contextmanager
from functools import lru_cache
-from itertools import chain
+from itertools import chain, zip_longest
from pathlib import Path
from subprocess import PIPE, Popen
from tempfile import NamedTemporaryFile
@@ 587,19 588,45 @@ def _stop_branch(current, interrupt=Fals
layer.start()
-def creater(name, lowers, start=None, permanent=None):
- """create an overlay at `name` that inherites from `lowerdir`.
+class command_registry:
+ def __init__(self):
+ self._commands = []
+
+ def iter_commands(self):
+ return iter(self._commands)
- :name: the overlay path.
- If it doesn't contains a path separator (a.k.a. '/'),
- /var/lib/machines/{name} is used
+ def __call__(self, func):
+ self._commands.append(func)
+ return func
+
+
+command = command_registry()
+
+
+@command
+def create(mountdir, lowerdir=(), *, start=False, permanent=False, **globaloptions):
+ """create a new overlay
- :lowers: overlayed directories
- If a lower path doesn't contains a path separator (a.k.a. '/'),
- /var/lib/machines/{lower} is used
+ Create a new overlay at MOUNTDIR (copy-on-write)
+ on top of LOWERDIR (read-only). MOUNTDIR can be a name
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+ :lowerdir: Director[y,ies] used as base FS.
+ If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<lowerdir>.
+ Can be specified multiple times to combine lower directories
+ :start: Start the overlay immediatly
+ :permanent: Make changes permanent on reboot
+
+ Upper and work directory are placed in {OVERLAYDIR}.
+ Systemd unit files are created and enabled in {UNITSDIR}
+ allowing auto mounting the overlay on any acces to the
+ overlay mount point.
"""
- layer = Layer(name)
- layer.lowers = [build_layer(lowername) for lowername in lowers]
+
+ layer = Layer(mountdir)
+ layer.lowers = [build_layer(lowername) for lowername in lowerdir]
layer.dump()
systemctl.daemon_reload()
if start:
@@ 622,36 649,76 @@ def _delete_layer(layer):
layer.delete()
-def deleter(name):
- """Delete a managed overlay and related folder/files"""
- _delete_layer(build_layer(name))
+@command
+def delete(mountdir, **globaloptions):
+ """Delete an existing overlay
+
+ Delete the overlay at MOUNTDIR. MOUNTDIR can be a name.
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+
+ upperdir, workdir, mountdir, systemd units are removed.
+ So, overall data will be lost.
+ """
+ _delete_layer(build_layer(mountdir))
systemctl.daemon_reload()
- logger.info("\"%s\" deleted", name)
+ logger.info("\"%s\" deleted", mountdir)
-def reseter(name):
- """Delete then re-create an overlay preserving the configuration."""
- layer = build_layer(name)
+@command
+def reset(mountdir, **globaloptions):
+ """reset the overlay
+
+ Delete then re-create an overlay preserving the configuration
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+
+ Quick alias for `delete`, `create` then `start` preserving the configuration.
+ """
+ layer = build_layer(mountdir)
_delete_layer(layer)
layer.dump()
systemctl.daemon_reload()
layer.start()
-
def _get_editor_on_file(filename):
cmd = '%s "%s"' % (os.environ.get('EDITOR', 'vim'), filename)
Popen(cmd, shell=True).wait()
-def editer(name, appended=None, prepended=None, removed=None, interrupt=False, preserve=False):
- """Ask user to edit lowers of the given overlay name, rewrite
- the metadata and unit files accordingly, finaly retart systemd units
- of the edited overlay.
+@command
+def edit(
+ mountdir,
+ *,
+ prepend=(),
+ append=(),
+ delete=(),
+ interrupt: '-I' = False,
+ preserve: '-P' = False,
+ **globaloptions,
+):
+ """edit existing overlay
+
+ edit an overlay created by this tool.
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+ :prepend: Prepend the following layers (can be used multiple times)
+ :append: Append the following layers (can be used multiple times)
+ :delete: Delete the following layers (can be used multiple times)
+ :interrupt: Automatically interrupt descendants to prevent broken mount point.
+ Restart them afterward.
+ :preserve: Do not check if descendant overlays are started
+ (may produce inconsistant behaviours)
+
+ An editor is started if --prepend nor --append nor --delete is provided.
"""
- layer = build_layer(name)
- with _stop_branch(layer, interrupt=interrupt, preserve=preserve, restart=interrupt):
- if not appended and not prepended and not removed:
+ layer = build_layer(mountdir)
+ with _stop_branch(layer, interrupt, preserve, restart=interrupt):
+ if not append and not prepend and not delete:
with NamedTemporaryFile() as fobj:
fobj.write(b'# -*- encoding: utf-8 -*-\n')
fobj.write(b'# Note: Leave unchanged or fully empty to ignore changes\n')
@@ 668,12 735,12 @@ def editer(name, appended=None, prepende
if not fobj.read(1):
return # Ignore empty file for convenience
else:
- prepended = map(build_layer, prepended or ())
- appended = map(build_layer, appended or ())
- removed = list(map(build_layer, removed or ()))
+ prepended = map(build_layer, chain(*prepend))
+ appended = map(build_layer, chain(*append))
+ deleted = list(map(build_layer, chain(*delete)))
lowers = chain(
prepended,
- (lower for lower in layer.lowers if lower not in removed),
+ (lower for lower in layer.lowers if lower not in deleted),
appended,
)
if lowers == layer.lowers:
@@ 687,8 754,26 @@ def editer(name, appended=None, prepende
logger.info("\"%s\" updated", layer.name)
-def starter(name, interrupt=False, preserve=False, permanent=False):
- layer = build_layer(name)
+@command
+def start(
+ mountdir, *, permanent=False, interrupt: '-I' = False, preserve: '-P' = False, **globaloptions
+):
+ """Mount the overlay
+
+ Enable automounting of the layer, on first access (e.g. `ls <mountdir>`).
+ Start the systemctl automount unit for the layer.
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+ :permanent: Make changes permanent on reboot
+ :interrupt: Automatically interrupt descendants to prevent broken mount point.
+ Restart them afterward.
+ :preserve: Do not check if descendant overlays are started
+ (may produce inconsistant behaviours)
+
+ In order to prevent strang behaviours only one overlay shall be started on a branch.
+ """
+ layer = build_layer(mountdir)
if not layer.is_managed():
logger.info("\"%s\" not managed. Nothing to do.", layer.name)
return
@@ 696,18 781,59 @@ def starter(name, interrupt=False, prese
layer.start(permanent)
-def stoper(name, permanent=False):
- layer = build_layer(name)
+@command
+def stop(mountdir, *, permanent=False, **globaloptions):
+ """Unmount the overlay
+
+ Unmount the overlay and stop the systemd automount unit.
+
+ :mountdir: Overlay mount point. If just a name (without any \"{os.path.sep}\"),
+ it is assumed to be {MOUNTDIR}/<mountdir>.
+ :permanent: Make changes permanent on reboot
+
+ This is required to create a new layer on top of this one.
+ """
+ layer = build_layer(mountdir)
if not layer.is_managed():
logger.info("\"%s\" not managed. Nothing to do.", layer.name)
return
layer.stop(permanent)
-def lister(unit_name, ordered_deps, reversed_deps, no_info, regexp, depends_on, terminal):
- """List managed overlays and display overlay inheritance."""
- layers = _iter_layers(ordered_deps or reversed_deps)
- layers = reversed(list(layers)) if reversed_deps else layers
+@command
+def list_(
+ regexp='',
+ *,
+ depends_on=(),
+ unit_name=False,
+ order_deps=False,
+ order_deps_reverse: '-O' = False,
+ no_info=False,
+ terminal=False,
+ **globaloptions,
+):
+ """List existing overlays
+
+ list the overlays created by this tool.
+
+ :regexp: Optional regular expression matching overlay names to display
+ :depends_on: dsplay overlays that depends on this one.
+ "-d name1 name2" means "name1 and name2".
+ "-d name1 -d name2" means "name1 or name2".
+ :unit_name: display the full systemd unit file names instead of canonical names
+ :order_deps: order overlays from head to root of the dependencies graph (if possible)
+ :order_deps_reverse: order overlays from root to heads of the dependencies graph (if possible)
+ :no_info: do not display overlay information
+ :terminal: show terminal overlays only.
+
+ List managed overlays and display overlay inheritance.
+ '-t' is useful to list only layers that you mostly enable.
+ '-d' is useful to list the impacted layers if you change something on a lower layer.
+ '-n' is useful for automation scripts.
+ '-u' is useful for debuging.
+ """
+ layers = _iter_layers(order_deps or order_deps_reverse)
+ layers = reversed(list(layers)) if order_deps_reverse else layers
if terminal:
layers = list(layers)
lowers = set(chain(*(layer.lowers for layer in layers)))
@@ 717,6 843,7 @@ def lister(unit_name, ordered_deps, reve
layers = (layer for layer in layers if match(layer.name))
found = False
if depends_on:
+ depends_on = {frozenset(deps) for deps in depends_on}
depends_on = [{build_layer(layername) for layername in group} for group in depends_on]
for layer in layers:
descendants = layer.get_ascendants()
@@ 753,74 880,117 @@ def lister(unit_name, ordered_deps, reve
raise NoResult("No overlay matches.")
-def shower(name):
- layer = build_layer(name)
+@command
+def show(mountdir, **globaloptions):
+ """Show property of a layer.
+
+ Show detailled information about the lower configuration.
+
+ :mountdir: overlay mount point. If just a name (without any "{os.path.sep}"),
+ it is assumed to be {MOUNTDIR}/<mountdir>
+
+ You will find useful information about the layer and things to
+ manually handle the mount point.
+ """
+ layer = build_layer(mountdir)
if not layer.is_managed():
logger.warning('Not managed.')
return
lowers = ', '.join(x.name for x in layer.lowers)
-
ascendants = ', '.join(x.name for x in layer.get_ascendants())
descendants = ', '.join(x.name for x in _iter_descendants(layer))
config = layer.mountunit.load_config()
- command = ' '.join([
- 'mount', config['Mount']['What'],
- '-t', config['Mount']['Type'],
- '-o', config['Mount']['Options'],
- config['Mount']['Where']])
- print(dedent(f'''\
- {Style.BRIGHT}Name{Style.RESET_ALL}: {layer.name}
- {Style.BRIGHT}Mount dir{Style.RESET_ALL}: {layer.mountdir}
- {Style.BRIGHT}Lowers{Style.RESET_ALL}: {lowers}
- {Style.BRIGHT}Ascendants{Style.RESET_ALL}: {ascendants}
- {Style.BRIGHT}Descendants{Style.RESET_ALL}: {descendants}
- {Style.BRIGHT}Automount unit{Style.RESET_ALL}: {layer.automountunit.path}
- {Style.BRIGHT}Mount unit{Style.RESET_ALL}: {layer.mountunit.path}
- {Style.BRIGHT}Mount{Style.RESET_ALL}: {command}
- {Style.BRIGHT}Unmount{Style.RESET_ALL}: unmount {layer.mountdir}\
- '''))
+ command = ' '.join(
+ [
+ 'mount',
+ config['Mount']['What'],
+ '-t',
+ config['Mount']['Type'],
+ '-o',
+ config['Mount']['Options'],
+ config['Mount']['Where'],
+ ]
+ )
+ info = {
+ 'Name': layer.name,
+ 'Mount dir': layer.mountdir,
+ 'Lowers': lowers,
+ 'Ascendants': ascendants,
+ 'Descendants': descendants,
+ 'Automount unit': layer.automountunit.path,
+ 'Mount unit': layer.mountunit.path,
+ 'Mount': command,
+ 'Unmount': f'unmount {layer.mountdir}',
+ }
+ align = max(len(key) for key in info) + 1
+ for key, value in info.items():
+ print(f'{Style.BRIGHT}{key:>{align}s}{Style.RESET_ALL}: {value}')
-def statuser(name):
- layer = build_layer(name)
+@command
+def status(mountdir, **globaloptions):
+ """Status of the layer.
+
+ Display the effective status of the layer.
+
+ :mountdir: overlay mount point. If just a name (without any "{os.path.sep}"),
+ it is assumed to be {MOUNTDIR}/<mountdir>
+
+ A quick overview on systemd units status.
+ """
+
+ layer = build_layer(mountdir)
if not layer.is_managed():
logger.error('Not found.')
return
- Popen(['systemctl', 'status', layer.automountunit.path.name], stdout=sys.stdout).wait()
+ Popen(
+ ['systemctl', '--no-pager', 'status', layer.automountunit.path.name], stdout=sys.stdout
+ ).wait()
print('')
- Popen(['systemctl', 'status', layer.mountunit.path.name], stdout=sys.stdout).wait()
- # print(layer.mountunit.status())
-
+ Popen(
+ ['systemctl', '--no-pager', 'status', layer.mountunit.path.name], stdout=sys.stdout
+ ).wait()
-def deplacer(oldname, newname, interrupt=False, preserve=False):
- # Checks
- _new = Layer(newname)
+@command
+def move(old, new, *, interrupt: '-I' = False, preserve: '-P' = False, **globaloptions):
+ """Move existing overlay
+
+ Move a layer created by this tool.
+ :old: Existing overlay name
+ :new: New overlay name
+ :interrupt: Automatically interrupt descendants during operation to prevent broken
+ mount point.
+ :preserve: Do not check if descendant overlays are started
+ (may produce inconsistant behaviours)
+
+ Other overlays that depends on it are updated.
+ """
+ _new = Layer(new)
if _new.mountdir.exists():
- logger.error('"%s" already exists.', newname)
+ logger.error('"%s" already exists.', new)
return
-
- layer = build_layer(oldname)
-
+ layer = build_layer(old)
if layer == _new:
logger.info('Not a new unit name, nothing to do.')
return
-
- with _stop_branch(layer, interrupt=interrupt, preserve=preserve, restart=interrupt):
+ with _stop_branch(layer, interrupt, preserve, interrupt):
# Move overlay dirs.
layer.upperdir.rename(_new.upperdir)
layer.workdir.rename(_new.workdir)
-
layer.delete()
-
- layer.name = newname
-
+ layer.name = new
layer.dump()
for layer in _iter_descendants(layer):
layer.dump()
+ systemctl.daemon_reload()
- systemctl.daemon_reload()
+
+def ensure_directories_exist():
+ """Ensure required folders exist."""
+ for path in (MOUNTDIR, UPPERSDIR, WORKSDIR, INFOSDIR, UNITSDIR):
+ path.mkdir(parents=True, exist_ok=True)
def _setup_logger(level):
@@ 830,424 1000,127 @@ def _setup_logger(level):
logger.addHandler(steam_handler)
-def ensure_directories_exist():
- """Ensure required folders exist."""
- for path in (MOUNTDIR, UPPERSDIR, WORKSDIR, INFOSDIR, UNITSDIR):
- path.mkdir(parents=True, exist_ok=True)
+class _CommandBuilder:
+ """A little bit of brainfuck to fetch command information from the python AST."""
+
+ def __init__(self, func):
+ self._func = func
+
+ def declare_command(self, subparser):
+ documentations = self._extract_documentations()
+ parser = subparser.add_parser(
+ self._get_command_name(),
+ help=documentations['help'],
+ description=documentations['description'],
+ epilog=documentations['epilog'],
+ )
+ defaults = self._get_default_values()
+ for name in self._iter_positioned_arguments():
+ kwargs = {'metavar': name.upper()}
+ default = defaults.get(name)
+ if default is not None:
+ kwargs['default'] = default
+ if default == ():
+ kwargs['nargs'] = '*'
+ else:
+ kwargs['nargs'] = '?'
+ if name in documentations['arguments']:
+ kwargs['help'] = documentations['arguments'][name]
+ parser.add_argument(name, **kwargs)
+ for name in self._iter_named_arguments():
+ default = defaults.get(name)
+ args = ['--' + name.replace('_', '-')]
+ kwargs = {}
+ if default is not None:
+ kwargs['default'] = default
+ if name in documentations['arguments']:
+ kwargs['help'] = documentations['arguments'][name]
+ if name in self._func.__annotations__:
+ args.append(self._func.__annotations__[name])
+ else:
+ args.append('-' + name[0])
+ if default is False:
+ kwargs['action'] = 'store_true'
+ elif default is True:
+ kwargs['action'] = 'store_false'
+ elif default == ():
+ kwargs['action'] = 'append'
+ kwargs['nargs'] = '*'
+ kwargs['default'] = []
+ parser.add_argument(*args, **kwargs)
+ parser.set_defaults(func=self._func)
+
+ def _extract_documentations(self):
+ doc = dedent(self._func.__doc__ or '')
+ [help, description, argdoc, epilog] = (doc.split(os.linesep * 2, 3) + ['', '', ''])[:4]
+ arguments = (d.split(':') for d in re.split(r'\n\s*:', dedent(argdoc).strip(':')) if d)
+ arguments = {
+ name: dedent(doc).format(**globals())
+ for name, doc in (
+ d.split(':') for d in re.split(r'\n\s*:', dedent(argdoc).strip(':')) if d
+ )
+ }
+ return {
+ 'help': help,
+ 'description': description,
+ 'arguments': arguments,
+ 'epilog': epilog,
+ }
+
+ def _get_default_values(self):
+ code = self._func.__code__
+ arguments = code.co_varnames
+ defaults = {
+ name: value
+ for name, value in zip_longest(
+ reversed(arguments[: code.co_argcount]), reversed(self._func.__defaults__ or ())
+ )
+ if value is not None
+ }
+ defaults.update(self._func.__kwdefaults__ or {})
+ return defaults
+
+ def _iter_positioned_arguments(self):
+ code = self._func.__code__
+ return iter(code.co_varnames[: code.co_argcount])
+
+ def _iter_named_arguments(self):
+ code = self._func.__code__
+ return iter(code.co_varnames[code.co_argcount :][: code.co_kwonlyargcount])
+
+ def _get_command_name(self):
+ return self._func.__name__.rstrip('_')
def main():
- """Main entry point for the CLI"""
from argparse import ArgumentParser
ensure_directories_exist()
- parser = ArgumentParser(description="Manage images as overlays for machinectl.")
+ parser = ArgumentParser(description="Manage sytstemd-nspawn machines.")
+
parser.add_argument(
'--traceback',
action='store_true',
default=False,
- help="display traceback on errors",
+ help="Display traceback on errors",
)
parser.add_argument(
'-l',
'--loglevel',
default='INFO',
metavar='LEVEL',
- choices=(
- 'DEBUG',
- 'INFO',
- 'WARNING',
- 'ERROR',
- 'CRITICAL',
- 'debug',
- 'info',
- 'warning',
- 'error',
- 'critical',
- ),
+ choices=('debug', 'info', 'warning', 'error', 'critical'),
help=(
"Set log priority to LEVEL. The default log priority is %(default)s. "
- "Possible values are : CRITICAL, ERROR, WARNING, INFO, DEBUG"
- ),
- )
- subparser = parser.add_subparsers(
- help="available commands', description='Valid commands to manipulate overlays.",
- )
-
- def add_interrupt_preserve_arguments(parser):
- parser.add_argument(
- '--interrupt',
- '-i',
- action='store_true',
- default=False,
- help=(
- 'automatically interrupt descendants to prevent broken mount point. '
- 'Restart them afterward.'
- ),
- )
- parser.add_argument(
- '--preserve',
- '-K',
- action='store_true',
- default=False,
- help=(
- 'do not check if descendant overlays are started '
- '(may produce inconsistant behaviours)'
- ),
- )
-
- # create
- create = subparser.add_parser(
- 'create',
- help="create a new overlay",
- description=(
- "Create a new overlay at MOUNTDIR (copy-on-write) "
- "on top of LOWERDIR (read-only). MOUNTDIR can be a name"
- ),
- epilog=(
- "Upper and work directory are placed in %(od)s. "
- "Systemd unit files are created and enabled in %(ud)s "
- "allowing auto mounting the overlay on any acces to the "
- "overlay mount point."
- )
- % {'od': OVERLAYDIR, 'ud': UNITSDIR},
- )
- create.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
- create.add_argument(
- 'lowerdir',
- metavar='LOWERDIR',
- nargs='*',
- help=(
- "director{y,ies} used as base FS. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{LOWERDIR}. "
- "Can be specified multiple times to combine lower directories"
- )
- % (os.path.sep, MOUNTDIR),
- )
- create.add_argument(
- '--start',
- '-s',
- action='store_true',
- default=False,
- help='start the overlay immediatly',
- )
- create.add_argument(
- '--permanent',
- '-p',
- action='store_true',
- default=False,
- help='make changes permanent on reboot',
- )
-
- def _creater(args):
- creater(args.mountdir, args.lowerdir, args.start, args.permanent)
-
- create.set_defaults(func=_creater)
-
- # delete
- delete = subparser.add_parser(
- 'delete',
- help="delete an existing overlay",
- description='Delete the overlay at MOUNTDIR. MOUNTDIR can be a name.',
- epilog=(
- "upperdir, workdir, mountdir, systemd units are removed. "
- "So, overall data will be lost."
- ""
- ),
- )
- delete.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
-
- def _deleter(args):
- deleter(args.mountdir)
-
- delete.set_defaults(func=_deleter)
-
- # reset
- reset = subparser.add_parser(
- 'reset',
- help="Reset the overlay",
- description="Quick alias for `delete`, `create` then `start` preserving the configuration.",
- )
- reset.add_argument(
- 'mountdir', metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
-
- def _reseter(args):
- reseter(args.mountdir)
-
- reset.set_defaults(func=_reseter)
-
-
- # list
- list = subparser.add_parser(
- 'list',
- help="list existing overlays",
- description="list the overlays created by this tool.",
- )
- list.add_argument(
- 'regexp',
- metavar="REGEXP",
- default='',
- nargs='?',
- help="Optional regular expression matching overlay names to display",
- )
- list.add_argument(
- '-d',
- '--depends-on',
- metavar="NAME",
- default=[],
- nargs='*',
- action='append',
- help=(
- "dsplay overlays that depends on this one. \n"
- "\"-d name1 name2\" means \"name1 and name2\". \n"
- "\"-d name1 -d name2\" means \"name1 or name2\". \n"
+ "Possible values are : critical, error, warning, info, debug"
),
)
- list.add_argument(
- '-u',
- '--unit-name',
- action='store_true',
- default=False,
- help="display the full systemd unit file names instead of canonical names",
- )
- list.add_argument(
- '-o',
- '--order-deps',
- action='store_true',
- default=False,
- help="order overlays from head to root of the dependencies graph (if possible)",
- )
- list.add_argument(
- '-O',
- '--order-deps-reverse',
- action='store_true',
- default=False,
- help="order overlays from root to heads of the dependencies graph (if possible)",
- )
- list.add_argument(
- '-n',
- '--no-info',
- action='store_true',
- default=False,
- help="do not display overlay information",
- )
- list.add_argument(
- '-t',
- '--terminal',
- action='store_true',
- default=False,
- help="show terminal overlays only.",
- )
-
- def _lister(args):
- if args.order_deps and args.order_deps_reverse:
- parser.error('--order-deps and --orser-dep-reverse are mutually exclusive.')
- args.depends_on = {frozenset(deps) for deps in args.depends_on}
- lister(
- args.unit_name,
- args.order_deps,
- args.order_deps_reverse,
- args.no_info,
- args.regexp,
- set(args.depends_on),
- args.terminal,
- )
-
- list.set_defaults(func=_lister)
-
- # Info
- show = subparser.add_parser(
- 'show',
- help="Show property of a layer.",
- description="Show detailled information about the lower configuration.",
- )
- show.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
-
- def _shower(args):
- shower(args.mountdir)
-
- show.set_defaults(func=_shower)
-
- # status
- status = subparser.add_parser(
- 'status',
- help="Status of the overlay layer",
- description="Display the effective status of the layer."
- )
- status.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
-
- def _statuser(args):
- statuser(args.mountdir)
-
- status.set_defaults(func=_statuser)
+ subparser = parser.add_subparsers(help="Available commands to manipulate machines")
- # edit
- 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."
- ),
- )
- edit.add_argument(
- '-p',
- '--prepend',
- action='store_true',
- help="prepend the following names",
- )
- edit.add_argument(
- '-a',
- '--append',
- action='append',
- help="append the following names",
- )
- edit.add_argument(
- '-d',
- '--delete',
- action='append',
- help="append the following names",
- )
- add_interrupt_preserve_arguments(edit)
- edit.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
-
- def _editer(args):
- editer(
- args.mountdir,
- args.append,
- args.prepend,
- args.delete,
- interrupt=args.interrupt,
- preserve=args.preserve,
- )
-
- edit.set_defaults(func=_editer)
-
- # move
- 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"
- ),
- )
- add_interrupt_preserve_arguments(move)
- move.add_argument('old', metavar='OLD', help="existing overlay name.")
- move.add_argument('new', metavar='NEW', help="New overlay name.")
-
- def _deplacer(args):
- deplacer(args.old, args.new, interrupt=args.interrupt, preserve=args.preserve)
-
- move.set_defaults(func=_deplacer)
-
- # Start/Strop
- start = subparser.add_parser(
- 'start',
- help="Mount the overlay.",
- 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 strang behaviours only one overlay shall be started on a branch."
- ),
- )
- start.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
- add_interrupt_preserve_arguments(start)
- start.add_argument(
- '--permanent',
- '-p',
- action='store_true',
- default=False,
- help='make changes permanent on reboot',
- )
-
- def _start(args):
- starter(
- args.mountdir,
- interrupt=args.interrupt,
- preserve=args.preserve,
- permanent=args.permanent,
- )
-
- start.set_defaults(func=_start)
-
- stop = subparser.add_parser(
- 'stop',
- help="Unmount the overlay.",
- description=(
- "Unmount the overlay and stop the systemd automount target. "
- "This is required to create a new layer on top of this one. "
- ),
- )
- stop.add_argument(
- 'mountdir',
- metavar='MOUNTDIR',
- help=(
- "overlay mount point. If just a name (without any \"%s\"), "
- "it is assumed to be %s/{MOUNTDIR}" % (os.path.sep, MOUNTDIR)
- ),
- )
- stop.add_argument(
- '--permanent',
- '-p',
- action='store_true',
- default=False,
- help='make changes permanent on reboot',
- )
-
- def _stop(args):
- stoper(args.mountdir, args.permanent)
-
- stop.set_defaults(func=_stop)
+ for func in command.iter_commands():
+ _CommandBuilder(func).declare_command(subparser)
args = parser.parse_args()
_setup_logger(args.loglevel)
@@ 1255,7 1128,8 @@ def main():
parser.print_help()
else:
try:
- args.func(args)
+ kwargs = vars(args)
+ kwargs.pop('func')(**kwargs)
except Exception as err:
if args.traceback:
raise
@@ 9,8 9,7 @@ from unittest import TestCase, mock
# Like `import overlayctl` but it does not need a real python lib.
with io.open('overlayctl') as fobj:
- overlayctl = imp.load_module(
- 'overlayctl', fobj, 'overlayctl', ('', 'r', imp.PY_SOURCE))
+ overlayctl = imp.load_module('overlayctl', fobj, 'overlayctl', ('', 'r', imp.PY_SOURCE))
class BaseTest(TestCase):
@@ 39,10 38,7 @@ class BaseTest(TestCase):
self.systemctl.daemon_reload.assert_called()
def get_unit_name(self, systemd_name):
- name = '-'.join([
- str(self.mountdir).replace(os.path.sep, '-'),
- systemd_name
- ])
+ name = '-'.join([str(self.mountdir).replace(os.path.sep, '-'), systemd_name])
return name.lstrip('-')
def get_mount_dir(self, name):
@@ 56,7 52,8 @@ class BaseTest(TestCase):
def write_info(self, name, lowers):
(self.infosdir / self.get_unit_name(name)).write_text(
- json.dumps({'name': name, 'lowers': lowers}))
+ json.dumps({'name': name, 'lowers': lowers})
+ )
def assert_info_equal(self, name, lowers, systemd_name=''):
unitname = self.get_unit_name(systemd_name or name)
@@ 94,7 91,8 @@ class BaseTest(TestCase):
value = {
key: list(filter(None, folders.split(':')))
for group in value.split(',')
- for key, folders in [group.split('=', 1)]}
+ for key, folders in [group.split('=', 1)]
+ }
target[key] = value
return content
@@ 107,19 105,25 @@ class BaseTest(TestCase):
self.assertDictEqual(result, expected)
def check_mount_unit(self, layer_name, lower_names):
- self.assert_mount_unit_equal(layer_name, {
- 'Mount': {
- 'Options': {
- 'lowerdir': [str(lower_name) for lower_name in lower_names],
- 'upperdir': [str(self.get_upper_dir(layer_name))],
- 'workdir': [str(self.get_work_dir(layer_name))]},
- 'Type': 'overlay',
- 'What': 'overlay',
- 'Where': str(self.get_mount_dir(layer_name))},
- 'Unit': {
- 'ConditionPathExists': str(self.get_mount_dir(layer_name)),
- 'Description': '%s Container overlay' % layer_name
- }})
+ self.assert_mount_unit_equal(
+ layer_name,
+ {
+ 'Mount': {
+ 'Options': {
+ 'lowerdir': [str(lower_name) for lower_name in lower_names],
+ 'upperdir': [str(self.get_upper_dir(layer_name))],
+ 'workdir': [str(self.get_work_dir(layer_name))],
+ },
+ 'Type': 'overlay',
+ 'What': 'overlay',
+ 'Where': str(self.get_mount_dir(layer_name)),
+ },
+ 'Unit': {
+ 'ConditionPathExists': str(self.get_mount_dir(layer_name)),
+ 'Description': '%s Container overlay' % layer_name,
+ },
+ },
+ )
def assert_layer_started(self, name):
self.systemctl.start.assert_called_once_with(self.get_unit_name(name) + '.automount')
@@ 144,27 148,36 @@ class TestCreate(BaseTest):
# The layer name is escaped
name = 'test-it'
systemd_name = r'test\x2dit'
- overlayctl.creater(name, lowers=(), start=False, permanent=False)
+ overlayctl.create(name, lowers=(), start=False, permanent=False)
self.assert_daemon_reload_called()
self.assert_info_equal(name=name, lowers=[], systemd_name=systemd_name)
self.assert_overlay_dirs_exists(systemd_name, name)
- self.assert_automount_unit_equal(systemd_name, {
- 'Automount': {'Where': str(self.get_mount_dir('test-it'))},
- 'Install': {'WantedBy': 'local-fs.target'}
- })
- self.assert_mount_unit_equal(systemd_name, {
- 'Mount': {
- 'Options': {
- 'lowerdir': [],
- 'upperdir': [str(self.get_upper_dir(r'test\\x2dit'))],
- 'workdir': [str(self.get_work_dir(r'test\\x2dit'))]},
- 'Type': 'overlay',
- 'What': 'overlay',
- 'Where': str(self.get_mount_dir('test-it'))},
- 'Unit': {
- 'ConditionPathExists': str(self.get_mount_dir('test-it')),
- 'Description': 'test-it Container overlay'
- }})
+ self.assert_automount_unit_equal(
+ systemd_name,
+ {
+ 'Automount': {'Where': str(self.get_mount_dir('test-it'))},
+ 'Install': {'WantedBy': 'local-fs.target'},
+ },
+ )
+ self.assert_mount_unit_equal(
+ systemd_name,
+ {
+ 'Mount': {
+ 'Options': {
+ 'lowerdir': [],
+ 'upperdir': [str(self.get_upper_dir(r'test\\x2dit'))],
+ 'workdir': [str(self.get_work_dir(r'test\\x2dit'))],
+ },
+ 'Type': 'overlay',
+ 'What': 'overlay',
+ 'Where': str(self.get_mount_dir('test-it')),
+ },
+ 'Unit': {
+ 'ConditionPathExists': str(self.get_mount_dir('test-it')),
+ 'Description': 'test-it Container overlay',
+ },
+ },
+ )
def test_creating_a_simple_layer(self):
# With dependencies, the layer is created. The layer can
@@ 174,26 187,31 @@ class TestCreate(BaseTest):
self.write_info('lower1', [])
lower2dir = self.mountdir / 'lower2'
lower2dir.mkdir()
- overlayctl.creater(
- 'test', lowers=('lower1', str(lower2dir), 'lower3'))
+ overlayctl.create('test', ('lower1', str(lower2dir), 'lower3'))
self.assert_daemon_reload_called()
- self.assert_info_equal('test', lowers=[
- 'lower1', str(lower2dir), 'lower3'])
+ self.assert_info_equal('test', lowers=['lower1', str(lower2dir), 'lower3'])
self.assert_overlay_dirs_exists('test')
- self.assert_automount_unit_equal('test', {
- 'Automount': {'Where': str(self.get_mount_dir('test'))},
- 'Install': {'WantedBy': 'local-fs.target'}
- })
- self.check_mount_unit('test', [
- self.get_upper_dir('lower1'),
- self.get_mount_dir('lower2'),
- self.get_mount_dir('lower3')])
+ self.assert_automount_unit_equal(
+ 'test',
+ {
+ 'Automount': {'Where': str(self.get_mount_dir('test'))},
+ 'Install': {'WantedBy': 'local-fs.target'},
+ },
+ )
+ self.check_mount_unit(
+ 'test',
+ [
+ self.get_upper_dir('lower1'),
+ self.get_mount_dir('lower2'),
+ self.get_mount_dir('lower3'),
+ ],
+ )
def test_automount_started(self):
# with --start, the automount unit is automatically started.
self.systemctl.status.return_value = {'Active': 'inactive'}
self.get_mount_dir('lower').mkdir()
- overlayctl.creater('test', lowers=('lower',), start=True)
+ overlayctl.create('test', lowers=('lower',), start=True)
self.assert_overlay_dirs_exists('test')
self.assert_daemon_reload_called()
self.systemctl.start.assert_called_once_with(self.get_unit_name('test') + '.automount')
@@ 202,7 220,7 @@ class TestCreate(BaseTest):
# with --start --permanent, the automount unit is automatically started and enabled.
self.systemctl.status.return_value = {'Active': 'inactive'}
self.get_mount_dir('lower').mkdir()
- overlayctl.creater('test', lowers=('lower',), start=True, permanent=True)
+ overlayctl.create('test', lowers=('lower',), start=True, permanent=True)
self.assert_overlay_dirs_exists('test')
self.assert_daemon_reload_called()
unit = self.get_unit_name('test') + '.automount'
@@ 223,24 241,34 @@ class TestCreate(BaseTest):
#
# The lower directories are linearized using the c3 algorithm.
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('network', [])
- overlayctl.creater('nginx', ['network'])
- overlayctl.creater('postgresql', ['archbase'])
- overlayctl.creater('python', ['archbase'])
- overlayctl.creater('flask', ['python'])
- overlayctl.creater('sqlalchemy', ['postgresql', 'python'])
- overlayctl.creater('graphql', ['sqlalchemy'])
- overlayctl.creater('app', ['graphql', 'flask', 'nginx', ])
- self.check_mount_unit('app', [
- self.get_upper_dir('graphql'),
- self.get_upper_dir('flask'),
- self.get_upper_dir('nginx'),
- self.get_upper_dir('sqlalchemy'),
- self.get_upper_dir('network'),
- self.get_upper_dir('postgresql'),
- self.get_upper_dir('python'),
- self.get_mount_dir('archbase'),
- ])
+ overlayctl.create('network', [])
+ overlayctl.create('nginx', ['network'])
+ overlayctl.create('postgresql', ['archbase'])
+ overlayctl.create('python', ['archbase'])
+ overlayctl.create('flask', ['python'])
+ overlayctl.create('sqlalchemy', ['postgresql', 'python'])
+ overlayctl.create('graphql', ['sqlalchemy'])
+ overlayctl.create(
+ 'app',
+ [
+ 'graphql',
+ 'flask',
+ 'nginx',
+ ],
+ )
+ self.check_mount_unit(
+ 'app',
+ [
+ self.get_upper_dir('graphql'),
+ self.get_upper_dir('flask'),
+ self.get_upper_dir('nginx'),
+ self.get_upper_dir('sqlalchemy'),
+ self.get_upper_dir('network'),
+ self.get_upper_dir('postgresql'),
+ self.get_upper_dir('python'),
+ self.get_mount_dir('archbase'),
+ ],
+ )
class TestDelete(BaseTest):
@@ 251,14 279,14 @@ class TestDelete(BaseTest):
# info file, the .automount and the .mount systemd. unit file.
# The systemd units are stopped and disabled.
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.creater('test', lowers=('lower',), start=True, permanent=True)
+ overlayctl.create('test', lowers=('lower',), start=True, permanent=True)
automountunit = self.get_unit_name('test') + '.automount'
mountunit = self.get_unit_name('test') + '.mount'
self.assert_overlay_dirs_exists('test')
self.systemctl.start.assert_called_once_with(automountunit)
self.systemctl.enable.assert_called_once_with(automountunit)
self.systemctl.status.return_value = {'Active': 'active'}
- overlayctl.deleter('test')
+ overlayctl.delete('test')
self.assertFalse(self.get_upper_dir('test').exists())
self.assertFalse(self.get_work_dir('test').exists())
self.assertFalse(self.get_mount_dir('test').exists())
@@ 272,21 300,21 @@ class TestDelete(BaseTest):
# Unmanaged layer cannot be deleted.
archbase = self.get_mount_dir('archbase')
archbase.mkdir()
- overlayctl.deleter('archbase')
+ overlayctl.delete('archbase')
self.assertTrue(archbase.exists())
def test_cannot_delete_lower_layer(self):
# Its not allowed to delete a layer having descendants managed layer.
self.systemctl.status.return_value = {'Active': 'inactive'}
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
- overlayctl.creater('layer2', ['layer1'])
+ overlayctl.create('layer1', ['archbase'])
+ overlayctl.create('layer2', ['layer1'])
with self.assertRaises(overlayctl.DeletionError):
- overlayctl.deleter('layer1')
+ overlayctl.delete('layer1')
self.assertTrue(self.get_unit_path('layer1', 'mount').exists())
# But we can delete a layer on top of unmanaged layers.
- overlayctl.deleter('layer2')
- overlayctl.deleter('layer1')
+ overlayctl.delete('layer2')
+ overlayctl.delete('layer1')
class TestReset(BaseTest):
@@ 298,16 326,16 @@ class TestReset(BaseTest):
# Then recreate them.
self.systemctl.status.return_value = {'Active': 'inactive'}
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('lower1', lowers=('archbase',))
- overlayctl.creater('lower2', lowers=('archbase',))
- overlayctl.creater('test', lowers=['lower1', 'lower2'])
+ overlayctl.create('lower1', lowers=('archbase',))
+ overlayctl.create('lower2', lowers=('archbase',))
+ overlayctl.create('test', lowers=['lower1', 'lower2'])
lower1_file = self.get_upper_dir('lower1') / 'lower1'
lower2_file = self.get_upper_dir('lower2') / 'lower2'
test_file = self.get_upper_dir('test') / 'test'
lower1_file.write_text('1')
lower2_file.write_text('1')
test_file.write_text('1')
- overlayctl.reseter('test')
+ overlayctl.reset('test')
self.assertTrue(lower1_file.exists())
self.assertTrue(lower2_file.exists())
self.assertFalse(test_file.exists())
@@ 320,10 348,10 @@ class TestStart(BaseTest):
# Starting a layer starts the automount unit. Mounting is
# handled by systemd on first access to the folder (a.k.a `machinectl start`).
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
+ overlayctl.create('layer1', ['archbase'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.starter('layer1')
+ overlayctl.start('layer1')
self.assert_layer_started('layer1')
self.systemctl.enable.assert_not_called()
self.systemctl.stop.assert_not_called()
@@ 332,17 360,17 @@ class TestStart(BaseTest):
# The user can make change permanent after reboot using the
# --permanent option.
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
+ overlayctl.create('layer1', ['archbase'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.starter('layer1', permanent=True)
+ overlayctl.start('layer1', permanent=True)
self.assert_layer_started('layer1')
self.assert_layer_enabled('layer1')
def test_cannot_start_an_unmanaged_layer(self):
# Starting an unmanaged layer does nothing.
self.get_mount_dir('archbase').mkdir()
- overlayctl.starter('archbase')
+ overlayctl.start('archbase')
self.systemctl.start.assert_not_called()
self.systemctl.enable.assert_not_called()
@@ 351,11 379,11 @@ class TestStart(BaseTest):
# because overlayFS may broke thing.
self.systemctl.status.side_effect = [{'Active': 'active'}, {'Active': 'inactive'}]
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
- overlayctl.creater('layer2', ['layer1'])
+ overlayctl.create('layer1', ['archbase'])
+ overlayctl.create('layer2', ['layer1'])
self.systemctl.reset_mock()
with self.assertRaises(SystemExit):
- overlayctl.starter('layer1')
+ overlayctl.start('layer1')
self.systemctl.start.assert_not_called()
def test_user_can_force_to_start_a_lower_with_active_descendants(self):
@@ 364,24 392,26 @@ class TestStart(BaseTest):
# user is the master.
self.systemctl.status.side_effect = [{'Active': 'inactive'}]
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
- overlayctl.creater('layer2', ['layer1'])
+ overlayctl.create('layer1', ['archbase'])
+ overlayctl.create('layer2', ['layer1'])
self.systemctl.reset_mock()
- overlayctl.starter('layer1', preserve=True)
+ overlayctl.start('layer1', preserve=True)
self.assert_layer_started('layer1')
def test_starting_stops_descendants_if_needed(self):
# Active descendant layers can be stopped automatically before
# starting the layer with the --interrupt options.
self.systemctl.status.side_effect = [
- {'Active': 'active'}, {'Active': 'active'},
- {'Active': 'inactive'}]
+ {'Active': 'active'},
+ {'Active': 'active'},
+ {'Active': 'inactive'},
+ ]
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
- overlayctl.creater('layer2', ['layer1'])
- overlayctl.creater('layer3', ['layer2'])
+ overlayctl.create('layer1', ['archbase'])
+ overlayctl.create('layer2', ['layer1'])
+ overlayctl.create('layer3', ['layer2'])
self.systemctl.reset_mock()
- overlayctl.starter('layer1', interrupt=True)
+ overlayctl.start('layer1', interrupt=True)
self.assert_layer_stopped('layer2')
self.assert_layer_stopped('layer3')
self.assert_layer_started('layer1')
@@ 394,20 424,20 @@ class TestStop(BaseTest):
# Simply use the `stop` command to unmount the overlayfs and
# disable the automount unit.
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
+ overlayctl.create('layer1', ['archbase'])
self.systemctl.reset_mock()
self.systemctl.status.side_effect = [{'Active': 'active'}]
- overlayctl.stoper('layer1')
+ overlayctl.stop('layer1')
self.assert_layer_stopped('layer1')
self.systemctl.disable.assert_not_called()
def test_stop_a_layer_permanently(self):
# The user can disable automounting on reboot with the --permanent option.
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer1', ['archbase'])
+ overlayctl.create('layer1', ['archbase'])
self.systemctl.reset_mock()
self.systemctl.status.side_effect = [{'Active': 'active'}]
- overlayctl.stoper('layer1', permanent=True)
+ overlayctl.stop('layer1', permanent=True)
self.assert_layer_stopped('layer1')
self.assert_layer_disabled('layer1')
@@ 418,42 448,48 @@ class TestEdit(BaseTest):
def test_prepending_lowers(self):
# User can augment lowers with the --prepend option.
self.get_mount_dir('archbase').mkdir()
- overlayctl.creater('layer0', ['archbase'])
- overlayctl.creater('layer1', ['archbase'])
- overlayctl.creater('layer2', ['archbase'])
+ overlayctl.create('layer0', ['archbase'])
+ overlayctl.create('layer1', ['archbase'])
+ overlayctl.create('layer2', ['archbase'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.editer('layer2', appended=['layer0', 'layer1'])
+ overlayctl.edit('layer2', append=['layer0', 'layer1'])
self.assert_daemon_reload_called()
- self.check_mount_unit('layer2', [
- self.get_upper_dir('layer0'),
- self.get_upper_dir('layer1'),
- self.get_mount_dir('archbase'),
- ])
+ self.check_mount_unit(
+ 'layer2',
+ [
+ self.get_upper_dir('layer0'),
+ self.get_upper_dir('layer1'),
+ self.get_mount_dir('archbase'),
+ ],
+ )
def test_appending_lowers(self):
# User can augment the overlay bases with the --append option.
self.get_mount_dir('base1').mkdir()
self.get_mount_dir('base2').mkdir()
- overlayctl.creater('layer', [])
+ overlayctl.create('layer', [])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.editer('layer', appended=['base1', 'base2'])
+ overlayctl.edit('layer', append=['base1', 'base2'])
self.assert_daemon_reload_called()
- self.check_mount_unit('layer', [
- self.get_mount_dir('base1'),
- self.get_mount_dir('base2'),
- ])
+ self.check_mount_unit(
+ 'layer',
+ [
+ self.get_mount_dir('base1'),
+ self.get_mount_dir('base2'),
+ ],
+ )
def test_removing_lowers(self):
# User can reduce the lower layers with the --remove option.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('autologin', [])
- overlayctl.creater('network', [])
- overlayctl.creater('layer', ['network', 'autologin', 'base'])
+ overlayctl.create('autologin', [])
+ overlayctl.create('network', [])
+ overlayctl.create('layer', ['network', 'autologin', 'base'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.editer('layer', removed=['autologin', 'network'])
+ overlayctl.edit('layer', delete=['autologin', 'network'])
self.assert_daemon_reload_called()
self.check_mount_unit('layer', [self.get_mount_dir('base')])
@@ 461,81 497,91 @@ class TestEdit(BaseTest):
# User can use both use --remove and --append/--prepend.
self.get_mount_dir('base1').mkdir()
self.get_mount_dir('base2').mkdir()
- overlayctl.creater('layer', ['base1', 'base2'])
+ overlayctl.create('layer', ['base1', 'base2'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.editer('layer', removed=['base2'], prepended=['base2'])
+ overlayctl.edit('layer', delete=['base2'], prepend=['base2'])
self.assert_daemon_reload_called()
- self.check_mount_unit('layer', [
- self.get_mount_dir('base2'),
- self.get_mount_dir('base1'),
- ])
+ self.check_mount_unit(
+ 'layer',
+ [
+ self.get_mount_dir('base2'),
+ self.get_mount_dir('base1'),
+ ],
+ )
def test_modifications_propagated_to_descendants(self):
# Modifications are propagated to descendants.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('lower1', ['base'])
- overlayctl.creater('lower2', ['base'])
- overlayctl.creater('layer', ['lower2'])
+ overlayctl.create('lower1', ['base'])
+ overlayctl.create('lower2', ['base'])
+ overlayctl.create('layer', ['lower2'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.editer('lower2', prepended=['lower1'])
+ overlayctl.edit('lower2', prepend=['lower1'])
self.assert_daemon_reload_called()
- self.check_mount_unit('layer', [
- self.get_upper_dir('lower2'),
- self.get_upper_dir('lower1'),
- self.get_mount_dir('base'),
- ])
+ self.check_mount_unit(
+ 'layer',
+ [
+ self.get_upper_dir('lower2'),
+ self.get_upper_dir('lower1'),
+ self.get_mount_dir('base'),
+ ],
+ )
def test_with_editor(self):
# The user's editor is started with no options.
# In the editor, user can move, add and remove lower freely.
self.get_mount_dir('base1').mkdir()
self.get_mount_dir('base2').mkdir()
- overlayctl.creater('lower1', [])
- overlayctl.creater('lower2', [])
- overlayctl.creater('lower3', [])
- overlayctl.creater('layer', ['lower2', 'lower1', 'base1', 'base2'])
+ overlayctl.create('lower1', [])
+ overlayctl.create('lower2', [])
+ overlayctl.create('lower3', [])
+ overlayctl.create('layer', ['lower2', 'lower1', 'base1', 'base2'])
self.systemctl.reset_mock()
def check_and_edit_file(filename):
filepath = Path(filename)
- lowers = [line for line in filepath.read_text().splitlines()
- if not line.startswith('#')]
+ lowers = [
+ line for line in filepath.read_text().splitlines() if not line.startswith('#')
+ ]
self.assertSequenceEqual(lowers, ['lower2', 'lower1', 'base1', 'base2'])
filepath.write_text('\n'.join(['lower1', 'lower2', 'lower3', 'base1']))
self.systemctl.status.return_value = {'Active': 'inactive'}
with mock.patch('overlayctl._get_editor_on_file', new=check_and_edit_file):
- overlayctl.editer('layer')
- self.check_mount_unit('layer', [
- self.get_upper_dir('lower1'),
- self.get_upper_dir('lower2'),
- self.get_upper_dir('lower3'),
- self.get_mount_dir('base1'),
- ])
+ overlayctl.edit('layer')
+ self.check_mount_unit(
+ 'layer',
+ [
+ self.get_upper_dir('lower1'),
+ self.get_upper_dir('lower2'),
+ self.get_upper_dir('lower3'),
+ self.get_mount_dir('base1'),
+ ],
+ )
def test_cannot_edit_layer_with_active_descendant(self):
# Because OverlayFS does not like it, we protect users from
# having broken mount point.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('lower', ['base'])
- overlayctl.creater('layer', ['lower'])
+ overlayctl.create('lower', ['base'])
+ overlayctl.create('layer', ['lower'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'active'}
with self.assertRaises(SystemExit):
- overlayctl.editer('lower', removed=['base'])
+ overlayctl.edit('lower', delete=['base'])
self.check_mount_unit('lower', [self.get_mount_dir('base')])
def test_editing_layer_with_active_descendant(self):
# Because OverlayFS does not like it, we protect users from
# having broken mount point. But if the user is ok with that, let's do it.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('lower', ['base'])
- overlayctl.creater('layer', ['lower'])
+ overlayctl.create('lower', ['base'])
+ overlayctl.create('layer', ['lower'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'active'}
- overlayctl.editer('lower', removed=['base'], preserve=True)
+ overlayctl.edit('lower', delete=['base'], preserve=True)
self.check_mount_unit('lower', [])
@@ 546,48 592,57 @@ class TestMove(BaseTest):
# Simple case when the user moves a layer without descendants.
# So, we have nothing to propagate.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('lower', ['base'])
- overlayctl.creater('layer', ['lower'])
+ overlayctl.create('lower', ['base'])
+ overlayctl.create('layer', ['lower'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.deplacer('layer', 'newlayer')
+ overlayctl.move('layer', 'newlayer')
self.assert_layer_not_exist('layer')
- self.assert_layer_exists('newlayer', [
- self.get_upper_dir('lower'),
- self.get_mount_dir('base'),
- ])
+ self.assert_layer_exists(
+ 'newlayer',
+ [
+ self.get_upper_dir('lower'),
+ self.get_mount_dir('base'),
+ ],
+ )
self.assert_daemon_reload_called()
def test_descendant_of_moved_are_updated(self):
# When moving a layer, its descendants must be updated accordingly.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('layer1', ['base'])
- overlayctl.creater('layer2', ['layer1'])
- overlayctl.creater('layer3', ['layer2'])
+ overlayctl.create('layer1', ['base'])
+ overlayctl.create('layer2', ['layer1'])
+ overlayctl.create('layer3', ['layer2'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'inactive'}
- overlayctl.deplacer('layer1', 'renamed')
+ overlayctl.move('layer1', 'renamed')
self.assert_layer_not_exist('layer1')
self.assert_layer_exists('renamed', [self.get_mount_dir('base')])
- self.assert_layer_exists('layer2', [
- self.get_upper_dir('renamed'),
- self.get_mount_dir('base'),
- ])
- self.assert_layer_exists('layer3', [
- self.get_upper_dir('layer2'),
- self.get_upper_dir('renamed'),
- self.get_mount_dir('base'),
- ])
+ self.assert_layer_exists(
+ 'layer2',
+ [
+ self.get_upper_dir('renamed'),
+ self.get_mount_dir('base'),
+ ],
+ )
+ self.assert_layer_exists(
+ 'layer3',
+ [
+ self.get_upper_dir('layer2'),
+ self.get_upper_dir('renamed'),
+ self.get_mount_dir('base'),
+ ],
+ )
self.assert_daemon_reload_called()
def test_moved_layer_is_restarted(self):
# Moving an activated terminal layer is ok with --interrupt, It's still
# activated at the end.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('layer1', ['base'])
+ overlayctl.create('layer1', ['base'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'active'}
- overlayctl.deplacer('layer1', 'renamed', interrupt=True)
+ overlayctl.move('layer1', 'renamed', interrupt=True)
self.systemctl.stop.assert_any_call(self.get_unit_name('layer1') + '.mount')
self.systemctl.stop.assert_any_call(self.get_unit_name('layer1') + '.automount')
self.systemctl.start.assert_any_call(self.get_unit_name('renamed') + '.automount')
@@ 595,9 650,9 @@ class TestMove(BaseTest):
def test_cannot_rename_if_new_name_exists(self):
# Do nothing if the new name is already known.
self.get_mount_dir('lower1').mkdir()
- overlayctl.creater('lower2', [])
+ overlayctl.create('lower2', [])
self.systemctl.reset_mock()
- overlayctl.deplacer('lower2', 'lower1')
+ overlayctl.move('lower2', 'lower1')
self.assertTrue(self.get_unit_path('lower2', 'mount').exists())
self.assertTrue(self.get_mount_dir('lower2').exists())
self.assertTrue(self.get_mount_dir('lower1').exists())
@@ 606,12 661,12 @@ class TestMove(BaseTest):
# OverlayFS does not like when lower directories moves, so,
# user cannot move a layer having an active descandant.
self.get_mount_dir('base').mkdir()
- overlayctl.creater('lower', ['base'])
- overlayctl.creater('layer', ['lower'])
+ overlayctl.create('lower', ['base'])
+ overlayctl.create('layer', ['lower'])
self.systemctl.reset_mock()
self.systemctl.status.return_value = {'Active': 'active'}
with self.assertRaises(SystemExit):
- overlayctl.deplacer('lower', 'newlower')
+ overlayctl.move('lower', 'newlower')
def assert_layer_not_exist(self, name):
self.assertFalse(self.get_unit_path(name, 'mount').exists())
@@ 624,8 679,11 @@ class TestMove(BaseTest):
self.assertTrue(self.get_mount_dir(name).exists())
self.assertTrue(self.get_upper_dir(name).exists())
self.assertTrue(self.get_work_dir(name).exists())
- self.assert_automount_unit_equal(name, {
- 'Automount': {'Where': str(self.get_mount_dir(name))},
- 'Install': {'WantedBy': 'local-fs.target'}
- })
+ self.assert_automount_unit_equal(
+ name,
+ {
+ 'Automount': {'Where': str(self.get_mount_dir(name))},
+ 'Install': {'WantedBy': 'local-fs.target'},
+ },
+ )
self.check_mount_unit(name, lowers)