class Pipeline(object):

    @staticmethod
    def form_dict(definition_dict):

        PIPELINES_SCHEMA.validate(definition_dict)

        if 'vars' not in definition_dict:
            definition_dict['vars'] = {}
        if 'params' not in definition_dict:
            definition_dict['params'] = {}
        if 'triggers' not in definition_dict:
            definition_dict['triggers'] = {}

        if not isinstance(definition_dict, dict):
            raise PipelineError('Unexpected argument type %s expecting dict' % type(definition_dict))

        tasks = definition_dict['actions']
        plugins = definition_dict.get('plugins', [])

        return Pipeline(tasks, plugins, context=definition_dict)


    @staticmethod
    def from_yaml(file_path, params={}):
        if not isinstance(file_path, basestring):
            raise PipelineError('Unexpected argument type %s expecting string' % type(file_path))

        if not os.path.exists(file_path):
            raise PipelineError('PipelineDefinition: Pipeline definition file does not exist: %s' % file_path)

        with open(file_path) as f:
            try:
                pipeline_def = yaml.load(f)
            except yaml.YAMLError as e:
                log.exception(e)
                err_msg = e.problem
                err_msg += ' (line: {})'.format(e.problem_mark.line)
                raise PipelineError('PipelineDefinition: Pipeline definition file is not valid YAML: %s - %s' % (file_path, err_msg))

        pipeline_def['vars'] = deepmerge(pipeline_def.get('vars', {}), params or {})

        # Substitute {{ }} variables in tasks
        # vars = pipeline_def.get('vars', {})
        # pipeline_def = substitute_variables(vars, pipeline_def)

        return Pipeline.form_dict(pipeline_def)

    def __init__(self, tasks, plugins, context):
        self.tasks = []
        self.context = context

        self.plugin_mgr = PluginManager()
        plugins.extend(DEFAULT_PLUGINS)
        self.load_plugins(plugins)

        self.load_tasks(tasks)

    def load_plugins(self, plugins):
        log.debug("Loading plugins: %s" % plugins)
        for plugin in plugins:
            self._load_plugin(plugin)

    def load_tasks(self, tasks):
        for i, task in enumerate(tasks):
            normalized_task = self._normalize_task_dict(task)
            if not self._task_executor_valid(normalized_task['type']):
                raise PipelineError('Unsupported task type: %s' % normalized_task['type'])
            task_obj = Task.from_dict(normalized_task)
            if not task_obj.name:
                task_obj.name = 'Task-{}'.format(i + 1)
            self.tasks.append(task_obj)

    def _task_executor_valid(self, task_type):
        return bool(self.plugin_mgr.get_plugin('{}.execute'.format(task_type)))


    def run(self):
        log.debug('Run pipeline. params: {}'.format(self.context))
        pipeline_context ={
            'results': [],
            'vars': self.context.get('vars', {}),
            'status': PIPELINE_STATUS_OK,
            'prev_result': None,
            'params': self.context.get('params', {})
        }

        pipeline_context = DotMap(pipeline_context)

        self.plugin_mgr.trigger('on_pipeline_start', pipeline_context)
        log.debug('Pipeline starting. context: %s' % pipeline_context)
        for task in self.tasks:
            if self._should_run(task, pipeline_context):
                result_obj = None
                try:
                    task.args = substitute_variables(pipeline_context, task.args)
                except TemplateError as e:
                    result_obj = TaskResult(EXECUTION_FAILED, e.message)

                if not result_obj:
                    result_obj = self._run_task(task)
                    if task.ignore_errors and result_obj.status == EXECUTION_FAILED:
                        log.debug('Task has ignore_errors set, overriding to successful')
                        # Override to be successful
                        result_obj['status'] = EXECUTION_SUCCESSFUL

                pipeline_context.results.append(result_obj)
                pipeline_context['prev_result'] = result_obj

                if pipeline_context['status'] == PIPELINE_STATUS_OK and result_obj.get('status') != 0:
                    pipeline_context['status'] = PIPELINE_STATUS_FAIL
            else:
                log.debug('Skipping task: {}'.format(task.name))
        log.debug('Pipeline finished. Status: {}'.format(pipeline_context['status']))
        self.plugin_mgr.trigger('on_pipeline_finish', pipeline_context)
        return pipeline_context

    def _should_run(self, task, pipeline_context):

        if task.always_run:
            return True

        if pipeline_context['status'] != PIPELINE_STATUS_OK:
            return False

        return True

    def _validate_executor_plugin(self, event_name):
        if self.plugin_mgr.get_plugin_count(event_name) > 1:
            raise PipelineError('More than one executor plugins with same name {}'.format(event_name))

        if self.plugin_mgr.get_plugin_count(event_name) == 0:
            raise PipelineError('Can not find executor plugin for {}'.format(event_name))

    def _run_task(self, task):
        log.debug('Starting to execute task {}'.format(task.name))

        self.plugin_mgr.trigger('on_task_start', task)

        event_name = '{}.execute'.format(task.executor)

        self._validate_executor_plugin(event_name)

        try:
            results = self.plugin_mgr.trigger(event_name, task.args)  # Run the task
        except PluginError as e:
            log.warning('Unexpected plugin error running task: %s' % e)
            raise
        except (KeyboardInterrupt, SystemExit) as e:
            log.warning('Unexpected system error running task: %s' % e)
            raise
        except Exception as e:
            log.warning('Unexpected error running task: %s' % e)
            log.exception(e)
            results = [TaskResult(EXECUTION_FAILED, 'Unknown Error: %s' % e)]

        result = results[0]

        if not result:
            raise PipelineError('Result still missing %s' % result)

        self.plugin_mgr.trigger('on_task_finish', task, result)
        log.debug('Task finished. Result: %s' % result)

        return result

    def _load_plugin(self, plugin_str):
        if not isinstance(plugin_str, basestring):
            raise PluginError('Plugins must be a string, got {}'.format(plugin_str))

        plugin_class = self._resolve_plugin_class(plugin_str)
        self.plugin_mgr.register_plugin(plugin_class, self.context.get('vars'))


    def _resolve_plugin_class(self, plugin_name):
        if plugin_name in builtin_plugins:
            return builtin_plugins[plugin_name]

        return _parse_class(plugin_name)

    def _normalize_task_dict(self, task):
        # Ensure dict
        if not isinstance(task, dict):
            task = {
                'type': DEFAULTS['task']['type'],
                DEFAULTS['task']['param']: task
            }

        # Ensure type
        if 'type' not in task:
            task['type'] = DEFAULTS['task']['type']

        return task PIPELINE_STATUS_OK = 'success' PIPELINE_STATUS_FAIL = 'failure' class TaskResult(dict):
    def __init__(self, status, message='', data={}, return_obj={}):
        self['status'] = status
        self['message'] = message
        self['data'] = data
        self['return_obj'] = return_obj

    @property
    def status(self):
        return self.get('status')

    @property
    def message(self):
        return self.get('message')

    @property
    def data(self):
        return self.get('data')

    @property
    def return_obj(self):
        return self.get('return_obj')

    def is_successful(self):
        return self.status == EXECUTION_SUCCESSFUL EXECUTION_SUCCESSFUL = 0 EXECUTION_FAILED = 1 class BashExecutor(BaseExecutorPlugin):
    hook_prefix = 'bash'
    hooks = ('execute',)

    def __init__(self, log_file=None, event_mgr=None):
        self.log_file = log_file
        self.event_mgr = event_mgr

        log.debug('Bash executor initiated with log_file: %s' % self.log_file)

    def _parse_args_dict(self, args_dict):
        if 'cmd' not in args_dict:
            raise PluginError('BashExecutor got incorrect arguments, got: {}'.format(
                args_dict.keys()
            ))
        timeout = args_dict.get('timeout') or DEFAULT_TIMEOUT
        if not isinstance(timeout, int):
            raise PluginError('BashExecutor got incorrect timeout argument type, got: {} expecting int'.format(
                type(timeout)
            ))

        return args_dict['cmd'], timeout

    def execute(self, args_dict):
        cmd, timeout = self._parse_args_dict(args_dict)

        if self.dry_run:
            return TaskResult(EXECUTION_SUCCESSFUL)

        try:
            output, return_obj = self._run_bash(cmd, timeout)
            status = EXECUTION_SUCCESSFUL
            msg = 'Bash task finished'
        except BashExecuteError as e:
            status = EXECUTION_FAILED
            # stdout = e.stderr
            msg = 'Bash task failed: %s' % e.msg
            output = e.data['stdout']

        return TaskResult(status, msg, data={'output': output}, return_obj=return_obj)


    def _run_bash(self, bash_input, timeout=DEFAULT_TIMEOUT):
        log.debug('Running bash command: "{}"'.format(bash_input))
        f = None
        if self.log_file:
            f = open(self.log_file, 'a+')

        output = {'stdout': '', 'last_line': ''}
        try:
            def process_line(line):
                log.debug('Got line: %s' % line)
                output['stdout'] += line

                if line and line.strip():
                    output['last_line'] = line

                if f:
                    f.write(line)

                if self.event_mgr:
                    if len(line)>0 and line[-1] == '\n':
                        line = line[:-1]
                    self.event_mgr.trigger('on_task_event', {'output': line})

            proc = bash(_in=bash_input, _out=process_line, _err=process_line, _timeout=timeout)
            proc.wait()
            log.debug('Finished: %s, %s, %s' % (proc.exit_code, proc.stdout, proc.stderr))

        except ErrorReturnCode as e:
            log.warning('BashExec failed %s' % e.message)
            raise BashExecuteError("Execution failed with code: %s" % e.exit_code, e.exit_code, data=output)
        except TimeoutException as e:
            log.debug('BashExec timed out after %s seconds' % timeout)
            raise BashExecuteError("Task Timed Out", 1, data=output)

        return_obj = None
        try:
            return_obj = json.loads(output['last_line'])
        except:
            log.debug('Failed to parse last line of bash as json: "{}"'.format(output['last_line']))

        return output['stdout'], return_obj

    @classmethod
    def from_dict(cls, conf_dict, event_mgr=None):
        return cls(conf_dict.get('bash_log_file'), event_mgr) class BashExecuteError(PluginError):
    def __init__(self, msg, code, data={}):
        self.msg = msg
        self.code = code
        self.data = data class PythonExecutor(BashExecutor):
    hook_prefix = 'python'
    hooks = ('execute',)

    def _validate_args_dict(self, args_dict):
        if 'file' not in args_dict and 'script' not in args_dict:
            raise PluginError('PythonExecutor requires either "file" or "script" arguments. Given: {}'.format(args_dict))

        if args_dict.get('file') and args_dict.get('script'):
            raise PluginError(
                'PythonExecutor got both "file" and "script" aguments. Only one is supported.')

        if args_dict.get('workspace') and args_dict.get('script'):
            raise PluginError('PythonExecutor requires either "file" or "script" arguments. Given: {}'.format(args_dict))

    def execute(self, args_dict):
        self._validate_args_dict(args_dict)
        workdir = args_dict.get('workdir')
        script = args_dict.get('script')
        file = args_dict.get('file')
        virtualenv = args_dict.get('virtualenv')

        bash_input = ''
        if workdir:
            bash_input += 'cd {}\n'.format(workdir)

        if virtualenv:
            activate_path = path.join(virtualenv, 'bin', 'activate')
            if not path.exists(activate_path) or not path.isfile(activate_path):
                raise PluginError('Python virtualenv doesn\'t exist: {}'.format(activate_path))

            bash_input += 'source {}\n'.format(activate_path)

        filename = file

        if script:
            f = NamedTemporaryFile(delete=False)

            f.write(script)
            log.debug('Wrote script to file {}, {}'.format(f.name, script))
            filename = f.name
            f.close()

        bash_input += 'python {}'.format(filename)
        log.debug('Running python script: {}'.format(bash_input))

        if self.dry_run:
            return TaskResult(EXECUTION_SUCCESSFUL)

        return_obj=None
        try:
            stdout, return_obj = self._run_bash(bash_input)
            status = EXECUTION_SUCCESSFUL
        except BashExecuteError as e:
            status = EXECUTION_FAILED
            stdout = e.data.get('stdout')
        print 'Finished, stdout: %s' % (stdout)

        return TaskResult(status, 'Execution finished', data={'output': stdout}, return_obj=return_obj) def conf_logging():
    logger = logging.getLogger('pipelines')
    logger.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    logger.propagate = False

    ch = logging.StreamHandler()
    ch.setFormatter(formatter)
    ch.setLevel(logging.DEBUG)

    logger.addHandler(ch)
import os
import tempfile
import unittest
from unittest.case import TestCase
from datetime import datetime
from pipelines.pipeline.pipeline import Pipeline, PIPELINE_STATUS_OK, PIPELINE_STATUS_FAIL
from pipelines.pipeline.task import TaskResult, EXECUTION_SUCCESSFUL, EXECUTION_FAILED
from pipelines.plugins.bash_executor import BashExecutor, BashExecuteError
from pipelines.plugins.python_executor import PythonExecutor
from pipelines.utils import conf_logging



conf_logging()

class TestPythonExecutor(TestCase):

    # def test_basic_loading(self):
    #
    #     print 'Running test_basic_loading'
    #     executor = BashExecutor.from_dict({})
    #     self.assertIsInstance(executor, BashExecutor)
    #
    # # def test_basic_loading(self):
    # #     with tempfile.NamedTemporaryFile() as f:
    # #         print 'Running test_basic_loading'
    # #         executor = BashExecutor.from_dict({'log_file': f.name})
    # #         self.assertIsInstance(executor, BashExecutor)
    # #         executor.execute()
    # #         self.assertLess(100, len())
    #
    #
    #
    #
    #
    # def test_timeout_script_pass(self):
    #     print 'Running test_timeout_script_pass'
    #     executor = BashExecutor(timeout=3)
    #     args = {
    #         'cmd': 'echo "test" && sleep "2"'
    #     }
    #     res = executor.execute(args)
    #     self.assertFalse(res.status, EXECUTION_SUCCESSFUL)
    #     self.assertEqual(res.data['output'], 'test\n')
    #
    # def test_timeout_script_fail(self):
    #     print 'Running test_timeout_script_pass'
    #     executor = BashExecutor(timeout=2)
    #     args = {
    #         'cmd': 'echo "test" && sleep "3"'
    #     }
    #     res = executor.execute(args)
    #     self.assertIsInstance(res, TaskResult)
    #     self.assertEqual(res.status, EXECUTION_FAILED)
    #     self.assertEqual(res.data['output'], 'test\n')
    #
    # def test_timeout_in_pipeline_def(self):
    #     pipeline_def = {
    #         'actions': [
    #         {
    #             'type': 'bash',
    #             'cmd': 'echo test',
    #             'timeout': 1
    #         }
    #         ],
    #
    #     }
    #     pipeline = Pipeline.form_dict(pipeline_def)
    #     res = pipeline.run()
    #     self.assertEqual(res['status'], PIPELINE_STATUS_OK)
    def test_basic_script(self):
        print 'Running test_basic_script'
        executor = BashExecutor()
        args = {
            'cmd': 'echo "test"'
        }
        res = executor.execute(args)
        self.assertIsInstance(res, TaskResult)
        self.assertEqual(res.status, EXECUTION_SUCCESSFUL)
        self.assertEqual(res.data['output'].strip(), 'test')
        self.assertEqual(res.message.strip(), 'Bash task finished')

    def test_return_obj(self):
        print 'Testing return object parsing'
        executor = BashExecutor()
        args = {
            'cmd': 'echo \'{"test": 1}\''
        }
        res = executor.execute(args)

        self.assertIsInstance(res, TaskResult)