Method test_trigger

class/crontabModel/triggerModel.py:786–937

@name 测试指定任务 @author hwliang @param args { trigger_id 任务ID } @return dict

(self,args)

        return page

    def test_trigger(self,args):
        '''
            @name 测试指定任务
            @author hwliang
            @param args<dict_obj>{
                trigger_id<int> 任务ID
            }
            @return dict
        ''&#x27;
        if not args or not 'trigger_id' in args:
            return public.returnMsg(False,'参数错误')


        trigger_info = self._sql.table('trigger').where('trigger_id=?',(args.trigger_id,)).find()
        if not trigger_info:
            return public.returnMsg(False,'任务不存在')
        script_body = ''
        script_type = 'bash'
        return_type = 'string'

        script_exts = {'bash':'sh','python':'py','php':'php','node':'js','ruby':'rb','perl':'pl'}
        script_args = ''
        #检查脚本是否存在
        if trigger_info['script_id']:
            script_info = self._sql.table('scripts').where('script_id=?',(trigger_info['script_id'],)).find()
            if not script_info:
                return public.returnMsg(False,'脚本不存在')
            script_body = script_info['script']
            script_type = script_info['script_type']
            return_type = script_info['return_type']
            if 'args' in trigger_info and trigger_info['args']: script_args = trigger_info['args']
        else:
            script_body = trigger_info['script_body']

        trigger_start_time = int(time.time())
        result_msg = ['正在执行任务...']
        if script_args: script_args = ' {}'.format(script_args)

        tmp_file = '{}/tmp/trigger_{}.{}'.format(public.get_panel_path(),trigger_info['trigger_id'],script_exts[script_type])
        public.writeFile(tmp_file,script_body)

        #执行脚本
        if script_type == 'bash':
            result = public.ExecShell("bash {}{}".format(tmp_file,script_args))
        else:
            result = public.ExecShell("{} {}{}".format(public.get_python_bin(),tmp_file,script_args))

        if os.path.exists(tmp_file):
            os.remove(tmp_file)

        if not result[0]:
            self.add_task_log(trigger_info['script_id'],trigger_info['trigger_id'],0,0,result[0],result[1],trigger_start_time,int(time.time()))
            return public.returnMsg(False,'脚本执行失败,错误信息:\n{}'.format(result[1]))


        result_msg.append("任务脚本已执行完成,返回结果:\n{}".format(result[0]))
        #检查任务事件
        operator_where = self._sql.table('operator_where').where('trigger_id=?',(args.trigger_id,)).select()