Worked on new loading and factory.

development
Shawn Davis 3 years ago
parent b2903c1490
commit 6f8bd48b64
  1. 74
      scripttease/lib/commands/base.py
  2. 46
      scripttease/lib/commands/centos.py
  3. 10
      scripttease/lib/commands/django.py
  4. 14
      scripttease/lib/commands/messages.py
  5. 12
      scripttease/lib/commands/mysql.py
  6. 29
      scripttease/lib/commands/pgsql.py
  7. 12
      scripttease/lib/commands/php.py
  8. 30
      scripttease/lib/commands/posix.py
  9. 6
      scripttease/lib/commands/python.py
  10. 54
      scripttease/lib/commands/ubuntu.py
  11. 78
      scripttease/lib/factories.py
  12. 2
      scripttease/lib/loaders/__init__.py
  13. 520
      scripttease/lib/loaders/base.py
  14. 23
      scripttease/lib/loaders/ini.py
  15. 7
      scripttease/lib/loaders/yaml.py
  16. 1
      scripttease/lib/mappings.py

@ -11,7 +11,6 @@ log = logging.getLogger(__name__)
__all__ = (
"EXCLUDED_KWARGS",
"run",
"Command",
"ItemizedCommand",
"Sudo",
@ -24,6 +23,8 @@ EXCLUDED_KWARGS = [
"cd",
"comment",
"condition",
"environment",
"name",
"prefix",
"register",
"stop",
@ -31,27 +32,17 @@ EXCLUDED_KWARGS = [
"tags",
]
# Functions
def run(statement, **kwargs):
"""Run any statement.
- statement (str): The statement to be executed.
"""
kwargs.setdefault("comment", "run statement")
return Command(statement, **kwargs)
# Classes
class Command(object):
def __init__(self, statement, cd=None, comment=None, condition=None, prefix=None, register=None, stop=False, sudo=None, tags=None, **kwargs):
def __init__(self, statement, cd=None, comment=None, condition=None, name=None, prefix=None, register=None, stop=False, sudo=None, tags=None, **kwargs):
self.cd = cd
self.comment = comment
self.condition = condition
self.name = name
self.number = None
self.prefix = prefix
self.register = register
self.statement = statement
@ -143,36 +134,10 @@ class Command(object):
return self.statement
class Sudo(object):
"""Helper class for defining sudo options."""
def __init__(self, enabled=False, user="root"):
"""Initialize the helper.
:param enabled: Indicates sudo is enabled.
:type enabled: bool
:param user: The user to be invoked.
:type user: str
"""
self.enabled = enabled
self.user = user
def __bool__(self):
return self.enabled
def __str__(self):
if self.enabled:
return "sudo -u %s" % self.user
return ""
class ItemizedCommand(object):
"""An itemized command represents multiple commands of with the same statement but different parameters."""
def __init__(self, callback, items, *args, name=None, **kwargs):
def __init__(self, callback, items, *args, **kwargs):
"""Initialize the command.
:param callback: The function to be used to generate the command.
@ -193,7 +158,6 @@ class ItemizedCommand(object):
self.callback = callback
self.items = items
self.kwargs = kwargs
self.name = name
# Set defaults for when ItemizedCommand is referenced directly before individual commands are instantiated. For
# example, when command filtering occurs.
@ -245,6 +209,32 @@ class ItemizedCommand(object):
return True
class Sudo(object):
"""Helper class for defining sudo options."""
def __init__(self, enabled=False, user="root"):
"""Initialize the helper.
:param enabled: Indicates sudo is enabled.
:type enabled: bool
:param user: The user to be invoked.
:type user: str
"""
self.enabled = enabled
self.user = user
def __bool__(self):
return self.enabled
def __str__(self):
if self.enabled:
return "sudo -u %s" % self.user
return ""
class Template(object):
PARSER_JINJA = "jinja2"

@ -3,21 +3,23 @@
from commonkit import split_csv
from .base import Command, Template
from .django import DJANGO_MAPPINGS
from .messages import MESSAGE_MAPPINGS
from .mysql import MYSQL_MAPPINGS
from .pgsql import PGSQL_MAPPINGS
from .php import PHP_MAPPINGS
from .posix import POSIX_MAPPINGS
from .python import PYTHON_MAPPINGS
# Exports
__all__ = (
"MAPPINGS",
"CENTOS_MAPPINGS",
"apache",
"apache_reload",
"apache_restart",
"apache_start",
"apache_stop",
"apache_test",
"command_exists",
"service_reload",
"service_restart",
"service_start",
@ -28,21 +30,10 @@ __all__ = (
"system_update",
"system_upgrade",
"system_uninstall",
"template",
"user",
)
def command_exists(name):
"""Indicates whether a given command exists in this overlay.
:param name: The name of the command.
:type name: str
:rtype: bool
"""
return name in MAPPINGS
# Functions
def apache(op, **kwargs):
@ -203,18 +194,6 @@ def system_upgrade(**kwargs):
return Command("yum update -y", **kwargs)
def template(source, target, backup=True, parser=None, **kwargs):
"""Create a file from a template.
- source (str): The path to the template file.
- target (str): The path to where the new file should be created.
- backup (bool): Indicates whether a backup should be made if the target file already exists.
- parser (str): The parser to use ``jinja`` (the default) or ``simple``.
"""
return Template(source, target, backup=backup, parser=parser, **kwargs)
def user(name, groups=None, home=None, op="add", password=None, **kwargs):
"""Create or remove a user.
@ -256,7 +235,7 @@ def user(name, groups=None, home=None, op="add", password=None, **kwargs):
raise NameError("Unsupported or unrecognized operation: %s" % op)
MAPPINGS = {
CENTOS_MAPPINGS = {
'apache': apache,
'install': system_install,
'reboot': system_reboot,
@ -265,15 +244,16 @@ MAPPINGS = {
'start': service_start,
'stop': service_stop,
'system': system,
'template': template,
'update': system_update,
'uninstall': system_uninstall,
'upgrade': system_upgrade,
'user': user,
}
MAPPINGS.update(COMMON_MAPPINGS)
MAPPINGS.update(DJANGO_MAPPINGS)
MAPPINGS.update(MYSQL_MAPPINGS)
MAPPINGS.update(PGSQL_MAPPINGS)
MAPPINGS.update(POSIX_MAPPINGS)
CENTOS_MAPPINGS.update(DJANGO_MAPPINGS)
CENTOS_MAPPINGS.update(MESSAGE_MAPPINGS)
CENTOS_MAPPINGS.update(MYSQL_MAPPINGS)
CENTOS_MAPPINGS.update(PHP_MAPPINGS)
CENTOS_MAPPINGS.update(PGSQL_MAPPINGS)
CENTOS_MAPPINGS.update(POSIX_MAPPINGS)
CENTOS_MAPPINGS.update(PYTHON_MAPPINGS)

@ -90,3 +90,13 @@ def django_static(**kwargs):
kwargs.setdefault("comment", "collect static files")
kwargs.setdefault("noinput", True)
return django("collectstatic", **kwargs)
DJANGO_MAPPINGS = {
'django': django,
'django.check': django_check,
'django.dump': django_dump,
'django.load': django_load,
'django.migration': django_migrate,
'django.static': django_static,
}

@ -21,8 +21,10 @@ def explain(message, heading=None, **kwargs):
return Command(message, **kwargs)
def screenshot(image, caption=None, **kwargs):
def screenshot(image, caption=None, height=None, width=None, **kwargs):
kwargs['caption'] = caption
kwargs['height'] = height
kwargs['width'] = width
return Command(image, **kwargs)
@ -49,3 +51,13 @@ def twist(message, title="Notice", url=None, **kwargs):
statement.append(url)
return Command(" ".join(statement), **kwargs)
MESSAGE_MAPPINGS = {
'dialog': dialog,
'echo': echo,
'explain': explain,
'screenshot': screenshot,
'slack': slack,
'twist': twist,
}

@ -3,6 +3,7 @@ from .base import EXCLUDED_KWARGS, Command
__all__ = (
"MYSQL_MAPPINGS",
"mysql_create",
"mysql_dump",
"mysql_exists",
@ -171,3 +172,14 @@ def mysql_user(name, admin_pass=None, admin_user="root", op="create", password=N
return command
else:
raise InvalidInput("Unrecognized or unsupported MySQL user operation: %s" % op)
MYSQL_MAPPINGS = {
'mysql.create': mysql_create,
'mysql.drop': mysql_drop,
'mysql.dump': mysql_dump,
'mysql.exists': mysql_exists,
'mysql.grant': mysql_grant,
# 'mysql.sql': mysql_exec,
'mysql.user': mysql_user,
}

@ -1,21 +1,4 @@
"""
[run django checks]
django: check
[export fixtures]
django: dump lookups.Category
[import fixtures]
django: load lookups.Category
[migrate the database]
django: migrate
[collect static files]
django: static
[create super user (ad hoc command)]
django: createsuperuser root
"""
from ...exceptions import InvalidInput
@ -23,6 +6,7 @@ from .base import EXCLUDED_KWARGS, Command
__all__ = (
"PGSQL_MAPPINGS",
"pgsql_create",
"pgsql_drop",
"pgsql_dump",
@ -31,6 +15,7 @@ __all__ = (
"pgsql_user",
)
def pgsql(command, *args, host="localhost", excluded_kwargs=None, password=None, port=5432, user="postgres", **kwargs):
# The excluded parameters (filtered below) may vary based on implementation. We do, however, need a default.
excluded_kwargs = excluded_kwargs or EXCLUDED_KWARGS
@ -153,3 +138,13 @@ def pgsql_user(name, admin_pass=None, admin_user="postgres", op="create", passwo
return command
else:
raise InvalidInput("Unrecognized or unsupported Postgres user operation: %s" % op)
PGSQL_MAPPINGS = {
'pgsql.create': pgsql_create,
'pgsql.drop': pgsql_drop,
'pgsql.dump': pgsql_dump,
'pgsql.exists': pgsql_exists,
# 'pgsql.sql': pgsql_exec,
'pgsql.user': pgsql_user,
}

@ -0,0 +1,12 @@
from .base import Command
def php_module(name, **kwargs):
statement = "phpenmod %s" % name
return Command(statement, **kwargs)
PHP_MAPPINGS = {
'php.module': php_module,
}

@ -103,7 +103,7 @@ def copy(from_path, to_path, overwrite=False, recursive=False, **kwargs):
return Command(" ".join(a), **kwargs)
def dir(path, group=None, mode=None, owner=None, recursive=True, **kwargs):
def directory(path, group=None, mode=None, owner=None, recursive=True, **kwargs):
"""Create a directory.
- path (str): The path to be created.
@ -317,6 +317,13 @@ def replace(path, backup=".b", delimiter="/", find=None, replace=None, **kwargs)
return Command(statement, **kwargs)
def run(statement, **kwargs):
"""Run any command."""
kwargs.setdefault("comment", "run statement")
return Command(statement, **kwargs)
def rsync(source, target, delete=False, exclude=None, host=None, key_file=None, links=True, port=22,
recursive=True, user=None, **kwargs):
"""Synchronize a directory structure.
@ -530,3 +537,24 @@ def write(path, content=None, **kwargs):
return Command(" ".join(a), **kwargs)
POSIX_MAPPINGS = {
'append': append,
'archive': archive,
'certbot': certbot,
'copy': copy,
'dir': directory,
'extract': extract,
'link': link,
'move': move,
'perms': perms,
'remove': remove,
'replace': replace,
'run': run,
'rysnc': rsync,
'scopy': scopy,
'sync': sync,
'touch': touch,
'wait': wait,
'write': write,
}

@ -37,3 +37,9 @@ def python_virtualenv(name, **kwargs):
kwargs.setdefault("comment", "create %s virtual environment" % name)
return Command("virtualenv %s" % name, **kwargs)
PYTHON_MAPPINGS = {
'pip': python_pip,
'virtualenv': python_virtualenv,
}

@ -2,16 +2,18 @@
from commonkit import split_csv
from .base import Command, Template
from .common import COMMON_MAPPINGS
from .django import DJANGO_MAPPINGS
from .messages import MESSAGE_MAPPINGS
from .mysql import MYSQL_MAPPINGS
from .pgsql import PGSQL_MAPPINGS
from .php import PHP_MAPPINGS
from .posix import POSIX_MAPPINGS
from .python import PYTHON_MAPPINGS
# Exports
__all__ = (
"MAPPINGS",
"UBUNTU_MAPPINGS",
"apache",
"apache_disable_module",
"apache_disable_site",
@ -22,7 +24,6 @@ __all__ = (
"apache_start",
"apache_stop",
"apache_test",
"command_exists",
"service_reload",
"service_restart",
"service_start",
@ -37,17 +38,7 @@ __all__ = (
"user",
)
def command_exists(name):
"""Indicates whether a given command exists in this overlay.
:param name: The name of the command.
:type name: str
:rtype: bool
"""
return name in MAPPINGS
# Functions
def apache(op, **kwargs):
@ -219,7 +210,7 @@ def system_install(name, **kwargs):
"""
kwargs.setdefault("comment", "install system package %s" % name)
return Command("apt-get install -y %s" % name, **kwargs)
return Command("apt install -y %s" % name, **kwargs)
def system_reboot(**kwargs):
@ -236,31 +227,19 @@ def system_uninstall(name, **kwargs):
"""
kwargs.setdefault("comment", "remove system package %s" % name)
return Command("apt-get uninstall -y %s" % name, **kwargs)
return Command("apt uninstall -y %s" % name, **kwargs)
def system_update(**kwargs):
kwargs.setdefault("comment", "update system package info")
return Command("apt-get update -y", **kwargs)
return Command("apt update -y", **kwargs)
def system_upgrade(**kwargs):
kwargs.setdefault("comment", "upgrade the system")
return Command("apt-get upgrade -y", **kwargs)
def template(source, target, backup=True, parser=None, **kwargs):
"""Create a file from a template.
- source (str): The path to the template file.
- target (str): The path to where the new file should be created.
- backup (bool): Indicates whether a backup should be made if the target file already exists.
- parser (str): The parser to use ``jinja`` (the default) or ``simple``.
"""
return Template(source, target, backup=backup, parser=parser, **kwargs)
return Command("apt upgrade -y", **kwargs)
def user(name, groups=None, home=None, op="add", password=None, **kwargs):
@ -305,7 +284,7 @@ def user(name, groups=None, home=None, op="add", password=None, **kwargs):
raise NameError("Unsupported or unrecognized operation: %s" % op)
MAPPINGS = {
UBUNTU_MAPPINGS = {
'apache': apache,
'apache.disable_module': apache_disable_module,
'apache.disable_site': apache_disable_site,
@ -323,15 +302,16 @@ MAPPINGS = {
'start': service_start,
'stop': service_stop,
'system': system,
'template': template,
'update': system_update,
'uninstall': system_uninstall,
'upgrade': system_upgrade,
'user': user,
}
MAPPINGS.update(COMMON_MAPPINGS)
MAPPINGS.update(DJANGO_MAPPINGS)
MAPPINGS.update(MYSQL_MAPPINGS)
MAPPINGS.update(PGSQL_MAPPINGS)
MAPPINGS.update(POSIX_MAPPINGS)
UBUNTU_MAPPINGS.update(DJANGO_MAPPINGS)
UBUNTU_MAPPINGS.update(MESSAGE_MAPPINGS)
UBUNTU_MAPPINGS.update(MYSQL_MAPPINGS)
UBUNTU_MAPPINGS.update(PGSQL_MAPPINGS)
UBUNTU_MAPPINGS.update(PHP_MAPPINGS)
UBUNTU_MAPPINGS.update(POSIX_MAPPINGS)
UBUNTU_MAPPINGS.update(PYTHON_MAPPINGS)

@ -0,0 +1,78 @@
import logging
from .commands.base import Command, ItemizedCommand, Template
from .commands.centos import CENTOS_MAPPINGS
from .commands.ubuntu import UBUNTU_MAPPINGS
log = logging.getLogger(__name__)
def command_exists(mappings, name):
"""Indicates whether a given command exists in this overlay.
:param mappings: A dictionary of command names and command functions.
:type mappings: dict
:param name: The name of the command.
:type name: str
:rtype: bool
"""
return name in mappings
def command_factory(loader, profile):
commands = list()
number = 1
for command_name, args, kwargs in loader.commands:
command = get_command(command_name, profile, *args, **kwargs)
if command is not None:
command.number = number
commands.append(command)
number += 1
return commands
def get_command(name, profile, *args, **kwargs):
"""Get a command instance.
:param name: The name of the command.
:type name: str
:param profile: The operating system profile name.
:type profile: str
args and kwargs are passed to the command function.
:rtype: scripttease.lib.commands.base.Command | scripttease.lib.commands.base.ItemizedCommand |
scripttease.lib.commands.base.Template
"""
if profile == "centos":
mappings = CENTOS_MAPPINGS
elif profile == "ubuntu":
mappings = UBUNTU_MAPPINGS
else:
log.error("Unsupported or unrecognized profile: %s" % profile)
return None
_args = list(args)
if name == "template":
source = _args.pop(0)
target = _args.pop(0)
return Template(source, target, **kwargs)
if not command_exists(mappings, name):
log.warning("Command does not exist: %s" % name)
return None
callback = mappings[name]
if "items" in kwargs:
items = kwargs.pop("items")
return ItemizedCommand(callback, items, *args, **kwargs)
return callback(*args, **kwargs)

@ -1,6 +1,6 @@
"""
The job of a loader is to collect commands and their arguments from a text file.
"""
from .base import filter_snippets, load_variables
from .base import load_variables
from .ini import INILoader
from .yaml import YMLLoader

@ -19,9 +19,6 @@ __all__ = (
# "filter_snippets",
"load_variables",
"BaseLoader",
"Snippet",
"Sudo",
"Template",
)
# Functions
@ -156,6 +153,7 @@ class BaseLoader(File):
be supplied as defaults for snippet processing.
"""
self.commands = list()
self.context = context
self.excluded_kwargs = excluded_kwargs or EXCLUDED_KWARGS
self.is_loaded = False
@ -171,7 +169,7 @@ class BaseLoader(File):
self.locations.insert(0, os.path.join(self.directory, "templates"))
def get_context(self):
"""Get the context for parsing command snippets.
"""Get the context for parsing command files.
:rtype: dict
@ -182,122 +180,6 @@ class BaseLoader(File):
return d
def get_snippets(self):
"""Get the snippets found in a config file.
:rtype: list[scripttease.lib.loaders.base.Snippet]
"""
a = list()
for canonical_name, args, kwargs in self.snippets:
snippet = self.find_snippet(canonical_name, *args, **kwargs)
a.append(snippet)
return a
def find_snippet(self, name, *args, **kwargs):
"""Find a named snippet that was defined in a config file.
:param name: The canonical name (or dotted name) of the snippet.
:type name: str
:rtype: scripttease.lib.loaders.base.Snippet | scripttease.lib.loaders.base.Template
``args`` and ``kwargs`` are passed to instantiate the snippet.
.. important::
The snippet may be invalid; always check ``snippet.is_valid``.
"""
# Templates require special handling.
if name == "template":
source = args[0]
target = args[1]
kwargs['locations'] = self.locations
context = kwargs.copy()
context.update(self.get_context())
return Template(source, target, context=context, **kwargs)
# Explanations have had the content split into lots of arg strings. We just need to reconstitute this as the
# snippet's content.
if name == "explain":
content = " ".join(list(args))
return Snippet("explain", content=content, kwargs=kwargs)
# Convert args to a list so we can update it below.
_args = list(args)
# The given name is not in the mappings -- which is a typo or invalid name -- but it could also be a dotted path
# to be followed down through the dictionary structure.
if name not in self.mappings[self.profile]:
if "." in name:
return self.find_snippet_by_dotted_name(name, *args, **kwargs)
log.error("Command not found in mappings: %s" % name)
return Snippet(name, args=_args, kwargs=kwargs)
# Get the global context for use in snippet instantiation below.
context = self.get_context()
# Formal or informal sub-commands exist in a dictionary.
if type(self.mappings[self.profile][name]) is dict:
try:
possible_sub_command = _args[0]
except IndexError:
log.warning("No sub-command argument for: %s" % name)
return Snippet(name)
if possible_sub_command in self.mappings[self.profile][name]:
sub = _args.pop(0)
snippet = self.mappings[self.profile][name][sub]
parser = self.mappings[self.profile][name].get('_parser', None)
_name = "%s.%s" % (name, sub)
return Snippet(_name, args=_args, content=snippet, context=context, kwargs=kwargs, parser=parser)
# Django allows pre-defined as well as adhoc commands. The name of the command is provided as the first
# argument in the config file. The following statements are only invoked if the possible_sub_command is not
# in the django dictionary. The "command" entry in the django dictionary is for handling ad hoc commands.
if name == "django":
sub = _args.pop(0)
kwargs['_name'] = sub
snippet = self.mappings[self.profile]['django']['command']
parser = self.mappings[self.profile]['django']['_parser']
return Snippet("django.%s" % sub, args=_args, content=snippet, context=context, kwargs=kwargs,
parser=parser)
log.warning("Sub-command could not be determined for: %s" % name)
return Snippet(name, args=list(args), context=context, kwargs=kwargs)
# The found snippet should just be a string.
return Snippet(name, args=list(args), content=self.mappings[self.profile][name], context=context, kwargs=kwargs)
def find_snippet_by_dotted_name(self, name, *args, **kwargs):
"""Find a snippet using it's dotted name.
:param name: The dotted name of the snippet.
:param args:
:param kwargs:
``args`` and ``kwargs`` are passed to instantiate the snippet.
.. important::
The snippet may be invalid; always check ``snippet.is_valid``.
"""
# This may not exist. If so, None is the value of the Snippet.content attribute.
snippet = pick(name, self.mappings[self.profile])
# The name of the builder callback is always the root of the given name plus _parser.
builder_name = "%s._parser" % name.split(".")[0]
parser = pick(builder_name, self.mappings[self.profile])
# Return the snippet instance.
return Snippet(name, args=list(args), parser=parser, content=snippet, context=self.get_context(), kwargs=kwargs)
def load(self):
"""Load the command file.
@ -314,7 +196,7 @@ class BaseLoader(File):
"""
if self.context is not None:
try:
return parse_jinja_template(self.path, self.context.mapping())
return parse_jinja_template(self.path, self.get_context())
except Exception as e:
log.error("Failed to process %s file as template: %s" % (self.path, e))
return None
@ -378,399 +260,3 @@ class BaseLoader(File):
_value = smart_cast(value)
return _key, _value
class Snippet(object):
"""A snippet is a pseudo-command which collects the content of the snippet as well as the parameters that may be
used to create an executable statement.
The purpose of a snippet is *not* to provide command execution, but to capture the parameters of a command defined
in a configuration file.
"""
def __init__(self, name, args=None, content=None, context=None, excluded_kwargs=None, kwargs=None, parser=None):
"""Initialize a snippet.
:param name: The canonical name of the snippet.
:type name: str
:param args: A list of arguments found in the config file.
:type args: list[str]
:param content: The content of the snippet.
:type content: str | list[str]
:param context: Additional context variables used to render the command.
:type context: dict
:param excluded_kwargs: See parameter description for BaseLoader.
:type excluded_kwargs: list[str]
:param kwargs: The keyword arguments found in the config file. These may be specific to the command or one of
the common options. They are accessible as dynamic attributes of the Snippet instance.
:type kwargs: dict
:param parser: A callback that may be used to assemble the command.
:type parser: callable
"""
self.args = args or list()
self.excluded_kwargs = excluded_kwargs or EXCLUDED_KWARGS
self.parser = parser
self.content = content
self.context = context or dict()
self.kwargs = kwargs or dict()
self.name = name
self.environments = kwargs.pop("environments", list())
self.tags = kwargs.pop("tags", list())
sudo = self.kwargs.pop("sudo", None)
if isinstance(sudo, Sudo):
self.sudo = sudo
elif type(sudo) is str:
self.sudo = Sudo(enabled=True, user=sudo)
elif sudo is True:
self.sudo = Sudo(enabled=True)
else:
self.sudo = Sudo()
def __getattr__(self, item):
return self.kwargs.get(item)
def __str__(self):
return str(self.name)
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
"""Get the command statement represented by the snippet.
:param cd: Indicates whether the change directory option should be included in the output. The ``cd`` option
must also be provided for the command in the configuration file.
:type cd: bool
:param include_comment: Indicates whether the command comment should be included in the output.
:type include_comment: bool
:param include_register: Indicates whether an additional statement to capture the result of the command should
be included in the output. The register option must also be defined for the command in
the configuration file.
:type include_register: bool
:param include_stop: Indicates whether an additional statement to exit on failure of the command should be
included in the output. The stop option must also be defined for the command in the
configuration file.
:type include_stop: bool
:rtype: str
.. note::
The boolean options allow implementers to exercise control over the output of the statement, so that the
snippet may be used in ways appropriate to the implementation.
"""
lines = list()
if self.comment and include_comment:
lines.append("# %s" % self.comment)
# Handle snippet itemization. Note that register and stop options are ignored.
if self.is_itemized:
for item in self.items:
args = list()
for arg in self.args:
args.append(arg.replace("$item", item))
if self.parser:
statement = self.parser(self, args=args, excluded_kwargs=self.excluded_kwargs)
else:
statement = self._parse(args=args)
a = list()
if cd and self.cd is not None:
a.append("( cd %s &&" % self.cd)
if self.prefix is not None:
a.append("%s &&" % self.prefix)
if self.sudo:
statement = "%s %s" % (self.sudo, statement)
a.append(statement)
if cd and self.cd is not None:
a.append(")")
lines.append(" ".join(a))
return "\n".join(lines)
# Handle normal (not itemized) snippets.
a = list()
if cd and self.cd is not None:
a.append("( cd %s &&" % self.cd)
if self.prefix is not None:
a.append("%s &&" % self.prefix)
if self.parser:
statement = self.parser(self, excluded_kwargs=self.excluded_kwargs)
else:
statement = self._parse()
if self.sudo:
statement = "%s %s" % (self.sudo, statement)
a.append(statement)
if cd and self.cd is not None:
a.append(")")
if self.condition is not None:
lines.append("if [[ %s ]]; then %s; fi;" % (self.condition, " ".join(a)))
else:
lines.append(" ".join(a))
if include_register and self.register is not None:
lines.append("%s=$?;" % self.register)
if include_stop and self.stop:
lines.append("if [[ $%s -gt 0 ]]; exit 1; fi;" % self.register)
elif include_stop and self.stop:
lines.append("if [[ $? -gt 0 ]]; exit 1; fi;")
else:
pass
return "\n".join(lines)
@property
def is_itemized(self):
"""Indicates whether the snippet includes multiple occurrences of the same command.
:rtype: bool
"""
s = " ".join(self.args)
return "$item" in s
@property
def is_valid(self):
"""Indicates whether the snippet is valid.
:rtype: bool
.. note::
This is done by determining if snippet content is not ``None``. The content is found during the loading
process when the Snippet instance is created.
"""
return self.content is not None
def _parse(self, args=None, kwargs=None):
"""Build the command statement from snippet content.
:param args: A list of arguments which override those provided by the command configuration.
:type args: list[str]
:param kwargs: A dictionary which overrides the options provided by the command configuration.
:type kwargs: dict
:rtype: str
"""
context = self.context.copy()
context['args'] = args or self.args
context.update(kwargs or self.kwargs)
if type(self.content) is list:
a = list()
for string in self.content:
output = parse_jinja_string(string, context)
if len(output) == 0:
continue
a.append(output)
return " ".join(a)
# try:
# return parse_jinja_string(self.content, context)
# except TypeError as e:
# log.error("Failed to build command statement for %s: %s" % (self.name, e))
# return None
return parse_jinja_string(self.content, context)
class Sudo(object):
"""Helper class for defining sudo options."""
def __init__(self, enabled=False, user="root"):
"""Initialize the helper.
:param enabled: Indicates sudo is enabled.
:type enabled: bool
:param user: The user to be invoked.
:type user: str
"""
self.enabled = enabled
self.user = user
def __bool__(self):
return self.enabled
def __str__(self):
if self.enabled:
return "sudo -u %s" % self.user
return ""
class Template(object):
PARSER_JINJA = "jinja2"
PARSER_PYTHON = "python"
PARSER_SIMPLE = "simple"
def __init__(self, source, target, backup=True, parser=PARSER_JINJA, **kwargs):
self.backup_enabled = backup
self.context = kwargs.pop("context", dict())
self.name = "template"
self.parser = parser
self.language = kwargs.pop("lang", None)
self.locations = kwargs.pop("locations", list())
self.source = os.path.expanduser(source)
self.target = target
sudo = kwargs.pop("sudo", None)
if isinstance(sudo, Sudo):
self.sudo = sudo
elif type(sudo) is str:
self.sudo = Sudo(enabled=True, user=sudo)
elif sudo is True:
self.sudo = Sudo(enabled=True)
else:
self.sudo = Sudo()
self.kwargs = kwargs
def __getattr__(self, item):
return self.kwargs.get(item)
def __str__(self):
return "template"
def get_content(self):
"""Parse the template.
:rtype: str | None
"""
template = self.get_template()
if self.parser == self.PARSER_SIMPLE:
content = read_file(template)
for key, value in self.context.items():
replace = "$%s$" % key
content = content.replace(replace, str(value))
return content
if self.parser == self.PARSER_PYTHON:
content = read_file(template)
return content % self.context
try:
return parse_jinja_template(template, self.context)
except TemplateNotFound:
log.error("Template not found: %s" % template)
return None
except TemplateError as e:
log.error("Could not parse %s template: %s" % (template, e))
return None
# noinspection PyUnusedLocal
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
lines = list()
if include_comment and self.comment is not None:
lines.append("# %s" % self.comment)
# TODO: Backing up a template's target is currently specific to bash.
if self.backup_enabled:
command = "%s mv %s %s.b" % (self.sudo, self.target, self.target)
lines.append('if [[ -f "%s" ]]; then %s; fi;' % (self.target, command.lstrip()))
# Get the content; e.g. parse the template.
content = self.get_content()
# Templates that are bash scripts will fail to write because of the shebang.
if content.startswith("#!"):
_content = content.split("\n")
first_line = _content.pop(0)
command = '%s echo "%s" > %s' % (self.sudo, first_line, self.target)
lines.append(command.lstrip())
command = "%s cat > %s << EOF" % (self.sudo, self.target)
lines.append(command.lstrip())
lines.append("\n".join(_content))
lines.append("EOF")
else:
command = "%s cat > %s << EOF" % (self.sudo, self.target)
lines.append(command.lstrip())
lines.append(content)
lines.append("EOF")
if include_register and self.register is not None:
lines.append("%s=$?;" % self.register)
if include_stop and self.stop:
lines.append("if [[ $%s -gt 0 ]]; exit 1; fi;" % self.register)
elif include_stop and self.stop:
lines.append("if [[ $? -gt 0 ]]; exit 1; fi;")
else:
pass
return "\n".join(lines)
def get_target_language(self):
if self.language is not None:
return self.language
if self.target.endswith(".conf"):
return "conf"
elif self.target.endswith(".ini"):
return "ini"
elif self.target.endswith(".php"):
return "php"
elif self.target.endswith(".py"):
return "python"
elif self.target.endswith(".sh"):
return "bash"
elif self.target.endswith(".yml"):
return "yaml"
else:
return "text"
def get_template(self):
"""Get the template path.
:rtype: str
"""
source = self.source
for location in self.locations:
_source = os.path.join(location, self.source)
if os.path.exists(_source):
return _source
return source
@property
def is_itemized(self):
# return "$item" in self.target
return False
@property
def is_valid(self):
return True

@ -29,14 +29,9 @@ class INILoader(BaseLoader):
log.warning("Input file does not exist: %s" % self.path)
return False
if self.context is not None:
try:
content = parse_jinja_template(self.path, self.get_context())
except Exception as e:
log.error("Failed to process %s INI file as template: %s" % (self.path, e))
return False
else:
content = read_file(self.path)
content = self.read_file()
if content is None:
return False
ini = ConfigParser()
try:
@ -59,11 +54,13 @@ class INILoader(BaseLoader):
# The first key/value pair is the command name and arguments.
if count == 0:
command_name = key
# kwargs['name'] = command_name
# Explanations aren't processed like commands, so the text need not be surrounded by double quotes.
# if command_name == "explain":
# args.append(value)
# continue
# Explanations and screenshots aren't processed like commands, so the text need not be surrounded
# by double quotes.
if command_name in ("explain", "screenshot"):
args.append(value)
continue
# Arguments surrounded by quotes are considered to be one argument. All others are split into a
# list to be passed to the parser. It is also possible that this is a call where no arguments are
@ -85,7 +82,7 @@ class INILoader(BaseLoader):
count += 1
self.snippets.append((command_name, args, kwargs))
self.commands.append((command_name, args, kwargs))
self.is_loaded = True
return True

@ -47,6 +47,7 @@ class YMLLoader(BaseLoader):
count = 0
kwargs = self.options.copy()
kwargs['comment'] = comment
kwargs['name'] = command
for key, value in tokens.items():
if key.startswith("_"):
@ -55,6 +56,12 @@ class YMLLoader(BaseLoader):
if count == 0:
command_name = key
# Explanations and screenshots aren't processed like commands, so the text need not be surrounded
# by double quotes.
if command_name in ("explain", "screenshot"):
args.append(value)
continue
try:
if value[0] == '"':
args.append(value.replace('"', ""))

@ -0,0 +1 @@
from .commands.posix import POSIX_MAPPINGS
Loading…
Cancel
Save