/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *  
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.shardingsphere.elasticjob.cloud.executor.prod;

import com.google.protobuf.ByteString;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.mesos.ExecutorDriver;
import org.apache.mesos.Protos;
import org.apache.mesos.Protos.TaskID;
import org.apache.mesos.Protos.TaskInfo;
import org.apache.mesos.Protos.TaskState;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.api.listener.ShardingContexts;
import org.apache.shardingsphere.elasticjob.cloud.executor.fixture.TestSimpleJob;
import org.apache.shardingsphere.elasticjob.cloud.executor.prod.TaskExecutor.TaskThread;
import org.apache.shardingsphere.elasticjob.infra.context.ExecutionType;
import org.apache.shardingsphere.elasticjob.infra.exception.JobSystemException;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.infra.yaml.YamlEngine;
import org.apache.shardingsphere.elasticjob.script.props.ScriptJobProperties;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.util.Collections;
import java.util.LinkedHashMap;

import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.verify;

@RunWith(MockitoJUnitRunner.class)
public final class TaskExecutorThreadTest {
    
    @Mock
    private ExecutorDriver executorDriver;
    
    private final String taskId = String.format("%s@-@0@-@%s@-@fake_slave_id@-@0", "test_job", ExecutionType.READY);
    
    @Test
    public void assertLaunchTaskWithDaemonTaskAndJavaSimpleJob() {
        TaskInfo taskInfo = buildJavaTransientTaskInfo();
        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);
        taskThread.run();
        verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_RUNNING).build());
        verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_FINISHED).build());
    }
    
    @Test
    public void assertLaunchTaskWithDaemonTaskAndJavaScriptJob() {
        TaskInfo taskInfo = buildSpringScriptTransientTaskInfo();
        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);
        taskThread.run();
        verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_RUNNING).build());
        verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_FINISHED).build());
    }
    
    // TODO the test case is not reach to catch block
    @Test
    //(expected = JobSystemException.class)
    public void assertLaunchTaskWithWrongElasticJobClass() {
        TaskInfo taskInfo = buildWrongElasticJobClass();
        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);
        try {
            taskThread.run();
        } catch (final JobSystemException ex) {
            assertTrue(ex.getMessage().startsWith(String.format("ElasticJob: Class '%s' must implements ElasticJob interface.", TaskExecutorThreadTest.class.getSimpleName())));
            throw ex;
        }
    }
    
    @Test
    public void assertLaunchTaskWithWrongClass() {
        TaskInfo taskInfo = buildWrongClass();
        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);
        try {
            taskThread.run();
        } catch (final JobSystemException ex) {
            assertTrue(ex.getMessage().startsWith("ElasticJob: Class 'WrongClass' initialize failure, the error message is 'WrongClass'."));
        }
    }
    
    private TaskInfo buildWrongClass() {
        return buildTaskInfo(buildJobConfigurationYaml()).build();
    }
    
    private TaskInfo buildWrongElasticJobClass() {
        return buildTaskInfo(buildJobConfigurationYaml()).build();
    }
    
    private TaskInfo buildJavaTransientTaskInfo() {
        return buildTaskInfo(buildJobConfigurationYaml()).build();
    }
    
    private TaskInfo buildSpringScriptTransientTaskInfo() {
        return buildTaskInfo(buildJobConfigurationYaml()).build();
    }
    
    private TaskInfo.Builder buildTaskInfo(final String jobConfigurationYaml) {
        return TaskInfo.newBuilder().setData(ByteString.copyFrom(serialize(jobConfigurationYaml)))
                .setName("test_job").setTaskId(TaskID.newBuilder().setValue(taskId)).setSlaveId(Protos.SlaveID.newBuilder().setValue("slave-S0"));
    }
    
    private byte[] serialize(final String jobConfigurationYaml) {
        // CHECKSTYLE:OFF
        LinkedHashMap<String, Object> result = new LinkedHashMap<>(2, 1);
        // CHECKSTYLE:ON
        ShardingContexts shardingContexts = new ShardingContexts(taskId, "test_job", 1, "", Collections.singletonMap(1, "a"));
        result.put("shardingContext", shardingContexts);
        result.put("jobConfigContext", jobConfigurationYaml);
        return SerializationUtils.serialize(result);
    }
    
    private String buildJobConfigurationYaml() {
        return YamlEngine.marshal(JobConfigurationPOJO.fromJobConfiguration(JobConfiguration.newBuilder("test_job", 1).setProperty(ScriptJobProperties.SCRIPT_KEY, "echo test").build()));
    }
}

@@ -1539,88 +1539,8 @@
 pe;%0A
-import org.apache.shardingsphere.elasticjob.infra.exception.JobSystemException;%0A
 impo
@@ -1612,16 +1612,16 @@
 onPOJO;%0A
+
 import o
@@ -1951,51 +1951,8 @@
 p;%0A%0A
-import static org.junit.Assert.assertTrue;%0A
 impo
@@ -3274,32 +3274,32 @@
 NING).build());%0A
+
         verify(e
@@ -3452,1380 +3452,8 @@
    %0A
-    // TODO the test case is not reach to catch block%0A    @Test%0A    //(expected = JobSystemException.class)%0A    public void assertLaunchTaskWithWrongElasticJobClass() %7B%0A        TaskInfo taskInfo = buildWrongElasticJobClass();%0A        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);%0A        try %7B%0A            taskThread.run();%0A        %7D catch (final JobSystemException ex) %7B%0A            assertTrue(ex.getMessage().startsWith(String.format(%22ElasticJob: Class '%25s' must implements ElasticJob interface.%22, TaskExecutorThreadTest.class.getSimpleName())));%0A            throw ex;%0A        %7D%0A    %7D%0A    %0A    @Test%0A    public void assertLaunchTaskWithWrongClass() %7B%0A        TaskInfo taskInfo = buildWrongClass();%0A        TaskThread taskThread = new TaskExecutor(new TestSimpleJob()).new TaskThread(executorDriver, taskInfo);%0A        try %7B%0A            taskThread.run();%0A        %7D catch (final JobSystemException ex) %7B%0A            assertTrue(ex.getMessage().startsWith(%22ElasticJob: Class 'WrongClass' initialize failure, the error message is 'WrongClass'.%22));%0A        %7D%0A    %7D%0A    %0A    private TaskInfo buildWrongClass() %7B%0A        return buildTaskInfo(buildJobConfigurationYaml()).build();%0A    %7D%0A    %0A    private TaskInfo buildWrongElasticJobClass() %7B%0A        return buildTaskInfo(buildJobConfigurationYaml()).build();%0A    %7D%0A    %0A
     

Remove expired test cases (#1248)