class Matrix(object):
    """Matrix handles multiple pipelines (ordered or in parallel)."""

    def __init__(self, matrix, parallel=False):
        """Initialize pipeline with matrix data."""
        self.event = Event.create(__name__)
        self.logger = Logger.get_logger(__name__)
        self.matrix = matrix
        self.parallel = parallel

    @staticmethod
    def can_process_matrix(entry, matrix_tags):
        """
        Check given matrix tags to be in the given list of matric tags.

        Args:
            entry (dict): matrix item (in yaml).
            matrix_tags (list): represents --matrix-tags defined by user in command line.
        Returns:
            bool: True when matrix entry can be processed.
        """
        if len(matrix_tags) == 0:
            return True

        count = 0
        if 'tags' in entry:
            for tag in matrix_tags:
                if tag in entry['tags']:
                    count += 1

        return count > 0

    def run_matrix_ordered(self, process_data):
        """
        Running pipelines one after the other.

        Returns
            dict: with two fields: success True/False and captured output (list of str).
        """
        output = []
        for entry in self.matrix:
            env = entry['env'].copy()
            env.update({'PIPELINE_MATRIX': entry['name']})

            if Matrix.can_process_matrix(entry, process_data.options.matrix_tags):
                self.logger.info("Processing pipeline for matrix entry '%s'", entry['name'])
                pipeline = Pipeline(model=process_data.model, env=env,
                                    options=process_data.options)
                pipeline.hooks = process_data.hooks
                result = pipeline.process(process_data.pipeline)
                output += result['output']
                if not result['success']:
                    return {'success': False, 'output': output}
        return {'success': True, 'output': output}

    def run_matrix_in_parallel(self, process_data):
        """Running pipelines in parallel."""
        worker_data = [{'matrix': entry, 'pipeline': process_data.pipeline,
                        'model': process_data.model, 'options': process_data.options,
                        'hooks': process_data.hooks} for entry in self.matrix
                       if Matrix.can_process_matrix(entry, process_data.options.matrix_tags)]
        output = []
        success = True
        with closing(multiprocessing.Pool(multiprocessing.cpu_count())) as pool:
            for result in pool.map(matrix_worker, worker_data):
                output += result['output']
                if not result['success']:
                    success = False
        return {'success': success, 'output': output}

    def process(self, process_data):
        """Process the pipeline per matrix item."""
        if self.parallel and not process_data.options.dry_run:
            return self.run_matrix_in_parallel(process_data)
        return self.run_matrix_ordered(process_data) class MatrixProcessData(object):
    """Matrix process parameter."""

    def __init__(self):
        """Initializing defaults."""
        self.__pipeline = {}
        self.__model = {}
        self.__options = None
        self.__hooks = None

    @property
    def pipeline(self):
        """Get pipeline definition."""
        return self.__pipeline

    @pipeline.setter
    def pipeline(self, value):
        """Set pipeline definition."""
        self.__pipeline = value

    @property
    def model(self):
        """Get model data."""
        return self.__model

    @model.setter
    def model(self, value):
        """Set model data."""
        self.__model = value

    @property
    def options(self):
        """Get application options."""
        return self.__options

    @options.setter
    def options(self, value):
        """Set application options."""
        self.__options = value

    @property
    def hooks(self):
        """Get hooks."""
        return self.__hooks

    @hooks.setter
    def hooks(self, value):
        """Set hooks."""
        self.__hooks = value class Pipeline(object):
    """Class for processing a pipeline definition."""

    def __init__(self, model=None, env=None, options=None):
        """
        Initializing pipeline with definition (loaded from a yaml file).

        Args:
            model (dict): if you have a model defined in your pipeline definition (yaml)
            env (dict): the env as defined (if) per matrix
            options (dict): command line options for spline
        """
        self.event = Event.create(__name__)
        self.options = options
        self.model = {} if not isinstance(model, dict) else model
        self.data = PipelineData()
        self.data.env_list[0].update([] if env is None else env)
        self.logger = Logger.get_logger(__name__)
        self.variables = {}

    @property
    def hooks(self):
        """Get hooks."""
        return self.data.hooks

    @hooks.setter
    def hooks(self, value):
        """Set hooks."""
        self.data.hooks = value

    def cleanup(self):
        """Run cleanup script of pipeline when hook is configured."""
        if self.data.hooks and len(self.data.hooks.cleanup) > 0:
            env = self.data.env_list[0].copy()
            env.update({'PIPELINE_RESULT': 'SUCCESS', 'PIPELINE_SHELL_EXIT_CODE': '0'})
            config = ShellConfig(script=self.data.hooks.cleanup, model=self.model,
                                 env=env, dry_run=self.options.dry_run,
                                 debug=self.options.debug, strict=self.options.strict,
                                 temporary_scripts_path=self.options.temporary_scripts_path)
            cleanup_shell = Bash(config)
            for line in cleanup_shell.process():
                yield line

    def process(self, pipeline):
        """Processing the whole pipeline definition."""
        output = []
        for entry in pipeline:
            key = list(entry.keys())[0]
            # an environment block can be repeated
            if key == "env":
                self.data.env_list[0].update(entry[key])
                self.logger.debug("Updating environment at level 0 with %s",
                                  self.data.env_list[0])
                continue

            # after validation it can't be anything else but a stage
            # and the title is inside the round brackets:
            stage = Stage(self, re.match(r"stage\((?P<title>.*)\)", key).group("title"))
            result = stage.process(entry[key])
            output += result['output']
            if not result['success']:
                return {'success': False, 'output': output}

        # logging the output of the cleanup shell when registered
        for line in self.cleanup():
            output.append(line)
            self.logger.info(" | %s", line)

        self.event.succeeded()
        return {'success': True, 'output': output} class Hooks(object):
    """Pipeline hooks."""

    def __init__(self, document=None):
        """Initialize hooks."""
        self.cleanup = ""

        if document is not None and 'hooks' in document:

            if 'cleanup' in document['hooks'] and 'script' in document['hooks']['cleanup']:
                self.cleanup = document['hooks']['cleanup']['script'] class ApplicationOptions(object):
    """Application wide configuation (usually via command line options)."""

    SCHEMA = {
        'definition': And(str, len),
        Optional('matrix_tags', default=''): Or(type(''), type(u'')),
        Optional('tags', default=''): Or(type(''), type(u'')),
        Optional('validate_only', default=False): bool,
        Optional('dry_run', default=False): bool,
        Optional('event_logging', default=False): bool,
        Optional('logging_config', default=''): Or(type(''), type(u'')),
        Optional('debug', default=False): bool,
        Optional('strict', default=False): bool,
        Optional('report', default='off'):
            And(str, lambda s: s in ['off', 'json', 'html']),
        Optional('temporary_scripts_path', default=''): Or(type(''), type(u''))
    }

    def __init__(self, **kwargs):
        """
        Initializing and validating fields.

        Args:
            kwargs (dict): application command line options.
        """
        try:
            arguments = Adapter(Schema(ApplicationOptions.SCHEMA).validate(kwargs))
            self.definition = arguments.definition
            self.matrix_tags = [entry for entry in arguments.matrix_tags.split(',') if len(entry) > 0]
            self.tags = [entry for entry in arguments.tags.split(',') if len(entry) > 0]
            self.validate_only = arguments.validate_only
            self.dry_run = arguments.dry_run
            self.event_logging = arguments.event_logging
            self.logging_config = arguments.logging_config
            self.debug = arguments.debug
            self.strict = arguments.strict
            self.report = arguments.report
            self.temporary_scripts_path = arguments.temporary_scripts_path
        except SchemaError as exception:
            logging.getLogger(__name__).error(exception)
            raise RuntimeError(str(exception)) class Loader(SafeLoader):
    """Yaml specific loader for including other yaml files."""

    def __init__(self, stream):
        """Initialize loader for specific includes."""
        super(Loader, self).__init__(stream)
        Loader.add_constructor('!include', Loader.include)

    def include(self, node):
        """Include the defined yaml file."""
        result = None
        if isinstance(node, ScalarNode):
            result = Loader.include_file(self.construct_scalar(node))
        else:
            raise RuntimeError("Not supported !include on type %s" % type(node))
        return result

    @staticmethod
    def include_file(filename):
        """Load another yaml file (no recursion)."""
        if os.path.isfile(filename):
            with open(filename) as handle:
                return safe_load(handle)
        raise RuntimeError("Include file %s doesn't exist!" % filename)

    @staticmethod
    def load(filename):
        """"Load yaml file with specific include loader."""
        if os.path.isfile(filename):
            with open(filename) as handle:
                return yaml_load(handle, Loader=Loader)  # nosec
        raise RuntimeError("File %s doesn't exist!" % filename) class Logger(object):
    """Wrapper for logging calls."""

    use_external_configuration = False

    @staticmethod
    def configure_by_file(filename):
        """Read logging configuration from an external file."""
        logging.config.fileConfig(filename)
        Logger.use_external_configuration = True

    @staticmethod
    def configure_default(logging_format, level):
        """Default configuration."""
        logging.basicConfig(format=logging_format, level=level)

    @staticmethod
    def get_logger(name):
        """Get a logger by name."""
        if name is None:
            return NoLogger()

        logger = logging.getLogger(name)
        if Logger.use_external_configuration:
            logger.propagate = False
        return logger def find_matrix(document):
    """
    Find  **matrix** in document.

    The spline syntax allows following definitions:
        - **'matrix'** - ordered execution of each pipeline (short form)
        - **'matrix(ordered)'** - ordered execution of each pipeline (more readable form)
        - **'matrix(parallel)'** - parallel execution of each pipeline

    Args:
        document (dict): validated spline document loaded from a yaml file.

    Returns:
        list: matrix as a part of the spline document or an empty list if not given.

    >>> find_matrix({})
    []
    >>> find_matrix({'matrix': [1]})
    [1]
    >>> find_matrix({'matrix(ordered)': [2]})
    [2]
    >>> find_matrix({'matrix(parallel)': [3]})
    [3]
    """
    return document['matrix'] if 'matrix' in document \
        else document['matrix(ordered)'] if 'matrix(ordered)' in document \
        else document['matrix(parallel)'] if 'matrix(parallel)' in document \
        else [] class Event(object):
    """Event mechanism for pipeline."""

    is_logging_enabled = False
    collector_queue = None

    def __init__(self, context, timestamp, **kwargs):
        """Initialize event with optional additional information."""
        self.context = context
        self.created = timestamp
        self.finished = timestamp
        self.status = 'started'
        self.information = {}
        self.information.update(kwargs)
        self.update_report_collector(int(time.mktime(self.created.timetuple())))

        if Event.is_logging_enabled:
            self.logger = Logger.get_logger(context + ".event")
        else:
            self.logger = Logger.get_logger(None)

    @staticmethod
    def configure(**kwargs):
        """Global configuration for event handling."""
        for key in kwargs:
            if key == 'is_logging_enabled':
                Event.is_logging_enabled = kwargs[key]
            elif key == 'collector_queue':
                Event.collector_queue = kwargs[key]
            else:
                Logger.get_logger(__name__).error("Unknown key %s in configure or bad type %s",
                                                  key, type(kwargs[key]))

    @staticmethod
    def create(context, **kwargs):
        """Create event with optional additional information."""
        return Event(context, datetime.now(), **kwargs)

    def delegate(self, success, **kwargs):
        """Delegate success/failure to the right method."""
        if success:
            self.succeeded(**kwargs)
        else:
            self.failed(**kwargs)

    def failed(self, **kwargs):
        """Finish event as failed with optional additional information."""
        self.finished = datetime.now()
        self.status = 'failed'
        self.information.update(kwargs)
        self.logger.info("Failed - took %f seconds.", self.duration())
        self.update_report_collector(int(time.mktime(self.finished.timetuple())))

    def succeeded(self, **kwargs):
        """Finish event as succeeded with optional additional information."""
        self.finished = datetime.now()
        self.status = 'succeeded'
        self.information.update(kwargs)
        self.logger.info("Succeeded - took %f seconds.", self.duration())
        self.update_report_collector(int(time.mktime(self.finished.timetuple())))

    def duration(self):
        """Calculate event duration."""
        return (self.finished - self.created).total_seconds()

    def update_report_collector(self, timestamp):
        """Updating report collector for pipeline details."""
        report_enabled = 'report' in self.information and self.information['report'] == 'html'
        report_enabled = report_enabled and 'stage' in self.information
        report_enabled = report_enabled and Event.collector_queue is not None

        if report_enabled:
            Event.collector_queue.put(CollectorUpdate(
                matrix=self.information['matrix'] if 'matrix' in self.information else 'default',
                stage=self.information['stage'],
                status=self.status,
                timestamp=timestamp,
                information=self.information
            )) class Collector(Process):  # pylint: disable=too-many-ancestors
    """Process that does collect updates updating the report data."""

    def __init__(self):
        """Initialize collector with the store and a queue."""
        super(Collector, self).__init__()
        self.store = Store()
        self.queue = Queue()

    def run(self):
        """Collector main loop."""
        while True:
            data = self.queue.get()
            if data is None:
                Logger.get_logger(__name__).info("Stopping collector process ...")
                break

            # updating the report data
            self.store.update(data)
            # writing the report
            generate(self.store, 'html', os.getcwd()) class VersionsCheck(object):
    """Evaluating versions of required tools."""

    LOGGER = Logger.get_logger(__name__)
    """Logger instance for this class."""

    BASH_VERSION = r'''bash --version|head -1'''
    """Find Bash version."""

    DOCKER_VERSION = r'''docker version|grep "Version:"|head -1'''
    """Find Docker version."""

    PACKER_VERSION = r'''packer -version'''
    """Find Packer version."""

    ANSIBLE_VERSION = r'''ansible --version'''
    """Find Ansible version."""

    def __init__(self):
        """Do nothing the moment."""
        pass

    def process(self, document):
        """Logging versions of required tools."""
        content = json.dumps(document)

        versions = {}
        versions.update({'Spline': Version(VERSION)})
        versions.update(self.get_version("Bash", self.BASH_VERSION))

        if content.find('"docker(container)":') >= 0 or content.find('"docker(image)":') >= 0:
            versions.update(VersionsCheck.get_version("Docker", self.DOCKER_VERSION))
        if content.find('"packer":') >= 0:
            versions.update(VersionsCheck.get_version("Packer", self.PACKER_VERSION))
        if content.find('"ansible(simple)":') >= 0:
            versions.update(VersionsCheck.get_version('Ansible', self.ANSIBLE_VERSION))

        return versions

    @staticmethod
    def get_version(tool_name, tool_command):
        """
        Get name and version of a tool defined by given command.

        Args:
            tool_name (str): name of the tool.
            tool_command (str): Bash one line command to get the version of the tool.

        Returns:
            dict: tool name and version or empty when no line has been found
        """
        result = {}
        for line in Bash(ShellConfig(script=tool_command, internal=True)).process():
            if line.find("command not found") >= 0:
                VersionsCheck.LOGGER.error("Required tool '%s' not found (stopping pipeline)!", tool_name)
                sys.exit(1)
            else:
                version = list(re.findall(r'(\d+(\.\d+)+)+', line))[0][0]
                result = {tool_name: Version(str(version))}
            break
        return result class VersionsReport(object):
    """Logging versions."""

    LOGGER = Logger.get_logger(__name__)
    """Logger instance for this class."""

    def process(self, versions):
        """Logging version sorted ascending by tool name."""
        for tool_name in sorted(versions.keys()):
            version = versions[tool_name]
            self._log("Using tool '%s', %s" % (tool_name, version))

    def _log(self, message):
        """Logging a message."""
        self.LOGGER.info(message) class Validator(object):
    """
    Validator for the Pythonic pipeline definition.

     - validates matrix, hooks, model and pipeline as main items in yaml.
     - validates parallel or ordered in matrix and tasks
     - support for Bash, Python, Docker, Packer and Ansible tasks
    """

    SCHEMA = {
        # optional matrix part of the schema
        Optional(Regex(r'((matrix\(parallel\)|matrix\(ordered\)|matrix))')): And(len, [{
            'name': And(str, len),
            Optional('env'): And(len, {
                Regex(r'([a-zA-Z][_a-zA-Z]*)'): And(str, len)
            }),
            Optional('tags'): And(len, [And(str, len)])
        }]),
        # optional model part of the schema
        # you may have any value (any yaml structure as value)
        Optional('model'): And(len, {
            Regex(r'([a-z][_a-z]*)'): object
        }),
        # optional hooks part of the schema
        Optional('hooks'): {
            'cleanup': {
                'script': And(str, len)
            }
        },
        # mandatory pipeline part of the schema
        'pipeline': And(len, [Or(
            # optional environment variables in a pipeline
            {'env': And(len, {
                Regex(r'([a-zA-Z][_a-zA-Z]*)'): And(str, len)
            })},
            {Regex(r'(stage\(.+\))'): And(len, [Or(
                # optional environment variables in a stage
                {'env': And(len, {
                    Regex(r'([a-zA-Z][_a-zA-Z]*)'): And(str, len)
                })},
                # optional tasks
                {Regex(r'((tasks\(parallel\)|matrix\(tasks\)|tasks))'): And(len, [
                    # optional environment variables
                    {Optional('env'): And(len, {
                        Regex(r'([a-zA-Z][_a-zA-Z]*)'): And(str, len)
                    })},
                    # optional shell task
                    {Optional('shell'): {
                        'script': And(Or(type(' '), type(u' ')), len),
                        Optional('title'): And(str, len),
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, Or(type(' '), type(u' '), [object])),
                        # when set store output of shell task into a variable with this name
                        # can be referenced in jinja templating with {{ variables.name }}
                        Optional('variable'):
                            And(Or(type(' '), type(u' ')), len, Regex(r'([a-zA-Z][_a-zA-Z]*)')),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }},
                    # optional Docker container task
                    {Optional('docker(container)'): {
                        'script': And(str, len),
                        # docker image name and docker tag (version)
                        Optional('image'): And(str, len),
                        # the title is printed to the logs (when given)
                        Optional('title'): And(str, len),
                        # when set the mount will be -v $PWD:/mnt/host
                        Optional('mount', default=False): bool,
                        # when set using -d option (docker run)
                        Optional('background', default=False): bool,
                        # when set using --rm option (docker run)
                        Optional('remove', default=True): bool,
                        # when set using --network= option (docker run)
                        Optional('network', default=''): And(str, len),
                        # when set using --labels= option (docker run)
                        Optional('labels'): And(len, {
                            Regex(r'(UL[_A-Z]*)'): And(str, len)
                        }),
                        # when defined used for --tags option (spline --tags=)
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, [object]),
                        # when set store output of docker container task into a variable with this name
                        # can be referenced in jinja templating with {{ variables.name }}
                        Optional('variable'):
                            And(Or(type(' '), type(u' ')), len, Regex(r'([a-zA-Z][_a-zA-Z]*)')),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }},
                    # optional Docker image task
                    {Optional('docker(image)'): {
                        'script': And(str, len),
                        'name': And(str, len),
                        'tag': And(str, len),
                        Optional('unique', default=True): bool,
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, [object]),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }},
                    # optional Python script
                    {Optional('python'): {
                        'script': And(str, len),
                        Optional('title'): And(str, len),
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, [object]),
                        # when set store output of python task into a variable with this name
                        # can be referenced in jinja templating with {{ variables.name }}
                        Optional('variable'):
                            And(Or(type(' '), type(u' ')), len, Regex(r'([a-zA-Z][_a-zA-Z]*)')),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }},
                    # optional Packer task
                    {Optional('packer'): {
                        'script': And(str, len),
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, [object]),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }},
                    # optional Ansible task
                    {Optional('ansible(simple)'): {
                        'script': And(str, len),
                        'inventory': And(str, len),
                        Optional('limit', default=''): And(str, len),
                        Optional('tags'): And([And(str, len)], len),
                        Optional('with'): And(len, [object]),
                        # condition when to run the task
                        Optional('when', default=''): And(str, Condition.is_valid)
                    }}
                ])}  # end of tasks
            )])},  # end of stage
        )])  # end of pipeline
    }  # end of schema

    @staticmethod
    def validate(data):
        """
        Validate data against the schema.

        Args:
            data(dict): data structure to validate.

        Returns:
            dict: data as provided and defaults where defined in schema.
        """
        try:
            return Schema(Validator.SCHEMA).validate(data)
        except SchemaError as exception:
            logging.getLogger(__name__).error(exception)
            return None
import sys
import platform
import os
import logging
import multiprocessing
import click
from spline.matrix import Matrix, MatrixProcessData
from spline.pipeline import Pipeline
from spline.components.hooks import Hooks
from spline.components.config import ApplicationOptions
from spline.tools.loader import Loader
from spline.tools.logger import Logger
from spline.tools.filters import find_matrix
from spline.tools.event import Event
from spline.tools.report.collector import Collector
from spline.tools.version import VersionsCheck, VersionsReport
from spline.validation import Validator
"""Represent the main entry point for the pipeline tool."""
# Copyright (c) 2017 Thomas Lehmann
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons
# to whom the Software is furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all copies
# or substantial portions of the Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
# pylint: too-many-instance-attributes




class Application(object):
    """Pipeline application."""

    def __init__(self, options):
        """
        Initialize application with command line options.

        Args:
            options (ApplicationOptions): given command line options.
        """
        self.event = Event.create(__name__)
        self.options = options
        self.logging_level = logging.DEBUG
        self.setup_logging()

        self.logger = Logger.get_logger(__name__)