/**
 * 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.oozie;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.service.Services;
import org.apache.oozie.store.StoreException;
import org.apache.oozie.test.XTestCase;
import org.apache.oozie.util.XConfiguration;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class TestCoordinatorEngine extends XTestCase {
    private Services services;

    protected void setUp() throws Exception {
        super.setUp();
        services = new Services();
        services.init();
    }

    protected void tearDown() throws Exception {
        services.destroy();
        super.tearDown();
    }

    public void testEngine() throws Exception {
        String appPath = getTestCaseFileUri("coordinator.xml");
        String jobId = _testSubmitJob(appPath);
        _testGetJob(jobId, appPath);
        _testGetJobs(jobId);
        _testStatus(jobId);
        _testGetDefinition(jobId);
        _testSubsetActions(jobId);
    }

    /**
     * Test Missing Dependencies with No Done Flag in Schema
     *
     * @throws Exception
     */
    public void testDoneFlag() throws Exception {
        Configuration conf = new XConfiguration();
        String appPath = getTestCaseFileUri("coordinator.xml");

        String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:days(1)}\" start=\"2009-02-01T01:00Z\" end="
                + "\"2009-02-01T02:00Z\" timezone=\"UTC\" "
                + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>2</concurrency> "
                + "<execution>LIFO</execution> </controls> <datasets> "
                + "<dataset name=\"local_a\" frequency=\"${coord:days(1)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("workflows/${YEAR}/${DAY}") + "</uri-template> "
                + "</dataset>"
                + "</datasets> <input-events> "
                + "<data-in name=\"A\" dataset=\"local_a\"> <instance>${coord:current(0)}</instance> </data-in>  "
                + "</input-events> "
                + "<action> <workflow> <app-path>hdfs:///tmp/workflows2/</app-path> "
                + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                + "</configuration> </workflow> </action> </coordinator-app>";

        writeToFile(appXml, appPath);
        conf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
        conf.set(OozieClient.USER_NAME, getTestUser());


        final CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        final String jobId = ce.submitJob(conf, true);
        waitFor(5000, new Predicate() {
            public boolean evaluate() throws Exception {
                try {
                    List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
                    for (CoordinatorAction action : actions) {
                        CoordinatorAction.Status actionStatus = action.getStatus();
                        if (actionStatus == CoordinatorAction.Status.WAITING) {
                            return true;
                        }
                    }
                }
                catch (Exception ex) {
                    return false;
                }
                return false;
            }
        });

        List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
        assertTrue(actions.size() > 0);
        CoordinatorAction action = actions.get(0);
        String missingDeps = action.getMissingDependencies();
        System.out.println("Missing deps=" + missingDeps);
        //done flag is not added to the missing dependency list
        assertEquals(getTestCaseFileUri("workflows/2009/01/_SUCCESS"), missingDeps);
    }

    /**
     * Test Missing Dependencies with Done Flag in Schema
     *
     * @throws Exception
     */
    public void testCustomDoneFlag() throws Exception {
        Configuration conf = new XConfiguration();
        String appPath = getTestCaseFileUri("coordinator.xml");
        String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:days(1)}\" start=\"2009-02-01T01:00Z\" end="
                + "\"2009-02-01T02:00Z\" timezone=\"UTC\" "
                + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>2</concurrency> "
                + "<execution>LIFO</execution> </controls> <datasets> "
                + "<dataset name=\"local_a\" frequency=\"${coord:days(1)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("workflows/${YEAR}/${MONTH}/${DAY}") + "</uri-template> "
                + "<done-flag>consume_me</done-flag> </dataset>"
                + "</datasets> <input-events> "
                + "<data-in name=\"A\" dataset=\"local_a\"> <instance>${coord:current(0)}</instance> </data-in>  "
                + "</input-events> "
                + "<action> <workflow> <app-path>hdfs:///tmp/workflows2/</app-path> "
                + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                + "</configuration> </workflow> </action> </coordinator-app>";
        writeToFile(appXml, appPath);
        conf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
        conf.set(OozieClient.USER_NAME, getTestUser());


        final CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        final String jobId = ce.submitJob(conf, true);

        waitFor(5000, new Predicate() {
            public boolean evaluate() throws Exception {
                try {
                    List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
                    for (CoordinatorAction action : actions) {
                        CoordinatorAction.Status actionStatus = action.getStatus();
                        if (actionStatus == CoordinatorAction.Status.WAITING) {
                            return true;
                        }
                    }
                }
                catch (Exception ex) {
                    return false;
                }
                return false;
            }
        });

        List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
        assertTrue(actions.size() > 0);
        CoordinatorAction action = actions.get(0);
        String missingDeps = action.getMissingDependencies();
        System.out.println("..Missing deps=" + missingDeps);
        assertEquals(new URI(getTestCaseFileUri("workflows/2009/02/01/consume_me")), new URI(missingDeps));
    }

    /**
     * Test Missing Dependencies with Empty Done Flag in Schema
     *
     * @throws Exception
     */
    public void testEmptyDoneFlag() throws Exception {
        Configuration conf = new XConfiguration();
        String appPath = getTestCaseFileUri("coordinator.xml");
        String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:days(1)}\" start=\"2009-02-01T01:00Z\" end="
                + "\"2009-02-01T02:00Z\" timezone=\"UTC\" "
                + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>2</concurrency> "
                + "<execution>LIFO</execution> </controls> <datasets> "
                + "<dataset name=\"local_a\" frequency=\"${coord:days(1)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("workflows/${YEAR}/${MONTH}/${DAY}") + "</uri-template> "
                + "<done-flag></done-flag> </dataset>"
                + "</datasets> <input-events> "
                + "<data-in name=\"A\" dataset=\"local_a\"> <instance>${coord:current(0)}</instance> </data-in>  "
                + "</input-events> "
                + "<action> <workflow> <app-path>hdfs:///tmp/workflows2/</app-path> "
                + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                + "</configuration> </workflow> </action> </coordinator-app>";
        writeToFile(appXml, appPath);
        conf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
        conf.set(OozieClient.USER_NAME, getTestUser());


        final CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        final String jobId = ce.submitJob(conf, true);

        waitFor(5000, new Predicate() {
            public boolean evaluate() throws Exception {
                try {
                    List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
                    for (CoordinatorAction action : actions) {
                        CoordinatorAction.Status actionStatus = action.getStatus();
                        if (actionStatus == CoordinatorAction.Status.WAITING) {
                            return true;
                        }
                    }
                }
                catch (Exception ex) {
                    return false;
                }
                return false;
            }
        });

        List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
        assertTrue(actions.size() > 0);
        CoordinatorAction action = actions.get(0);
        String missingDeps = action.getMissingDependencies();
        System.out.println("..Missing deps=" + missingDeps);
        assertEquals(getTestCaseFileUri("workflows/2009/02/01"), missingDeps);
    }

    /**
     * Test Missing Dependencies with Done Flag in Schema
     *
     * @throws Exception
     */
    public void testDoneFlagCreation() throws Exception {
        Configuration conf = new XConfiguration();
        String appPath = getTestCaseFileUri("coordinator.xml");
        String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:days(1)}\" start=\"2009-02-01T01:00Z\" end="
                + "\"2009-02-01T02:00Z\" timezone=\"UTC\" "
                + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>2</concurrency> "
                + "<execution>LIFO</execution> </controls> <datasets> "
                + "<dataset name=\"local_a\" frequency=\"${coord:days(1)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("workflows/${YEAR}/${MONTH}/${DAY}") + "</uri-template> "
                + "<done-flag>consume_me</done-flag> </dataset>"
                + "</datasets> <input-events> "
                + "<data-in name=\"A\" dataset=\"local_a\"> <instance>${coord:current(0)}</instance> </data-in>  "
                + "</input-events> "
                + "<action> <workflow> <app-path>hdfs:///tmp/workflows2/</app-path> "
                + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                + "</configuration> </workflow> </action> </coordinator-app>";
        writeToFile(appXml, appPath);
        conf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
        conf.set(OozieClient.USER_NAME, getTestUser());


        final CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        final String jobId = ce.submitJob(conf, true);

        //create done flag
        new File(getTestCaseDir(), "workflows/2009/02/01/consume_me").mkdirs();

        waitFor(10000, new Predicate() {
            public boolean evaluate() throws Exception {
                try {
                    List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
                    for (CoordinatorAction action : actions) {
                        CoordinatorAction.Status actionStatus = action.getStatus();
                        if (actionStatus == CoordinatorAction.Status.SUBMITTED) {
                            return true;
                        }
                    }
                }
                catch (Exception ex) {
                    return false;
                }
                return false;
            }
        });

        List<CoordinatorAction> actions = ce.getCoordJob(jobId).getActions();
        assertTrue(actions.size() > 0);
        CoordinatorAction action = actions.get(0);
        System.out.println("status=" + action.getStatus());
        String missingDeps = action.getMissingDependencies();
        System.out.println("..Missing deps=" + missingDeps);
        if (!(missingDeps == null || missingDeps.equals(""))) {
            fail();
        }
    }

    private String _testSubmitJob(String appPath) throws Exception {
        Configuration conf = new XConfiguration();

        String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:minutes(20)}\" start=\"2009-02-01T01:00Z\" end="
                + "\"2009-02-03T23:59Z\" timezone=\"UTC\" "
                + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>2</concurrency> "
                + "<execution>LIFO</execution> </controls> <datasets> "
                + "<dataset name=\"a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>file:///tmp/coord/workflows/${YEAR}/${DAY}</uri-template> </dataset> "
                + "<dataset name=\"local_a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
                + "timezone=\"UTC\"> <uri-template>file:///tmp/coord/workflows/${YEAR}/${DAY}</uri-template> </dataset> "
                + "</datasets> <input-events> "
                + "<data-in name=\"A\" dataset=\"a\"> <instance>${coord:latest(0)}</instance> </data-in>  "
                + "</input-events> "
                + "<output-events> <data-out name=\"LOCAL_A\" dataset=\"local_a\"> "
                + "<instance>${coord:current(-1)}</instance> </data-out> </output-events> <action> <workflow> <app-path>hdfs:"
                + "///tmp/workflows/</app-path> "
                + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                + "<property> <name>inputB</name> <value>${coord:dataOut('LOCAL_A')}</value> "
                + "</property></configuration> </workflow> </action> </coordinator-app>";
        writeToFile(appXml, appPath);
        conf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
        conf.set(OozieClient.USER_NAME, getTestUser());


        final CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        final String jobId = ce.submitJob(conf, true);
        waitFor(5000, new Predicate() {
            public boolean evaluate() throws Exception {
                try {
                    ce.getJob(jobId).getStatus();
                }
                catch (Exception ex) {
                    return false;
                }
                return true;
            }
        });
        assertEquals(jobId.substring(jobId.length() - 2), "-C");
        checkCoordJob(jobId);
        return jobId;
    }

    private void _testGetJob(String jobId, String appPath) throws Exception {
        CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        CoordinatorJob job = ce.getCoordJob(jobId);
        assertEquals(jobId, job.getId());
        assertEquals(job.getAppPath(), appPath);
    }

    /**
     * Test to validate frequency and time unit filters for jobs
     *
     * @throws Exception
     */
    public void _testGetJobs(String jobId) throws Exception {
        CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        // Test with no job filter specified
        CoordinatorJobInfo jobInfo = ce.getCoordJobs("", 1, 10);
        assertEquals(1, jobInfo.getCoordJobs().size());
        CoordinatorJob job = jobInfo.getCoordJobs().get(0);
        assertEquals(jobId, job.getId());

        // Test specifying the value for unit but leaving out the value for frequency
        try {
            jobInfo = ce.getCoordJobs("unit=minutes", 1, 10);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals("E0420: Invalid jobs filter [unit=minutes], time unit should be added only when "
                    + "frequency is specified. Either specify frequency also or else remove the time unit", ex
                    .getMessage());
        }

        // Test for invalid frequency value(Non-numeric value)
        try {
            jobInfo = ce.getCoordJobs("frequency=ghj;unit=minutes", 1, 10);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals("E0420: Invalid jobs filter [frequency=ghj;unit=minutes], "
                    + "invalid value [ghj] for frequency. A numerical value is expected", ex.getMessage());
        }

        // Test for invalid unit value(Other than months, days, minutes or hours)
        try {
            jobInfo = ce.getCoordJobs("frequency=60;unit=min", 1, 10);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals("E0420: Invalid jobs filter [frequency=60;unit=min], invalid value [min] for time unit. "
                    + "Valid value is one of months, days, hours or minutes", ex.getMessage());
        }
    }

    private void _testGetDefinition(String jobId) throws Exception {
        CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        CoordinatorJobBean job = ce.getCoordJob(jobId);
        System.out.println("JOBXML=" + job.getOrigJobXml());
        assertNotNull(job.getOrigJobXml());
    }

    /**
     * Helper methods
     *
     * @param jobId
     * @throws StoreException
     */
    private void checkCoordJob(String jobId) throws StoreException {
        try {
            CoordinatorJobBean job = CoordJobQueryExecutor.getInstance()
                    .get(CoordJobQueryExecutor.CoordJobQuery.GET_COORD_JOB, jobId);
        }
        catch (JPAExecutorException se) {
            se.printStackTrace();
            fail("Job ID " + jobId + " was not stored properly in db");
        }
    }

    private void writeToFile(String appXml, String appPath) throws Exception {
        File wf = new File(new URI(appPath).getPath());
        PrintWriter out = null;
        try {
            out = new PrintWriter(new OutputStreamWriter(new FileOutputStream(wf), StandardCharsets.UTF_8));
            out.println(appXml);
        }
        catch (IOException iex) {
            throw iex;
        }
        finally {
            if (out != null) {
                out.close();
            }
        }

    }

    private void _testStatus(final String jobId) throws Exception {
        waitFor(6000, new Predicate() {
            public boolean evaluate() throws Exception {
                CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
                CoordinatorJob job = ce.getCoordJob(jobId);
                return !job.getStatus().equals(CoordinatorJob.Status.PREP);
            }
        });

        CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        CoordinatorJob job = ce.getCoordJob(jobId);
        assertFalse(job.getStatus().equals(CoordinatorJob.Status.PREP));
    }

    private void _testSubsetActions(final String jobId) throws Exception {
        CoordinatorEngine ce = new CoordinatorEngine(getTestUser());
        // Check for WAITING filter
        CoordinatorJob job = ce.getCoordJob(jobId, "status=WAITING", 1, 2, false);
        // As both actions are waiting, expected result size is 2
        assertEquals(job.getActions().size(), 2);

        job = ce.getCoordJob(jobId, "status=WAITING", 1, 0, false);
        // Since length is 0, number of actions returned should be 0.
        assertEquals(job.getActions().size(), 0);

        job = ce.getCoordJob(jobId, "status=RUNNING", 1, 2, false);
        assertEquals(job.getActions().size(), 0);

        //Check for actions WAITING OR RUNNING
        job = ce.getCoordJob(jobId, "status=RUNNING;status=WAITING", 1, 2, false);
        assertEquals(job.getActions().size(), 2);

        //Check without filters
        job = ce.getCoordJob(jobId, null, 1, 2, false);
        assertEquals(job.getActions().size(), 2);

        //Check for empty filter list
        job = ce.getCoordJob(jobId, "", 1, 2, false);
        assertEquals(job.getActions().size(), 2);

        //Check for negative filter
        job = ce.getCoordJob(jobId, "status!=RUNNING", 1, 2, false);
        assertEquals(job.getActions().size(), 2);

        //Check for multiple negative filter
        job = ce.getCoordJob(jobId, "status!=RUNNING;status!=WAITING", 1, 2, false);
        assertEquals(job.getActions().size(), 0);

        //Check for combination of positive and negative filter
        try {
            job = ce.getCoordJob(jobId, "status=WAITING;status!=WAITING", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals(
                    "E0421: Invalid job filter [status=WAITING;status!=WAITING], the status [WAITING] "
                    + "specified in both positive and negative filters",
                    ex.getMessage());
        }

        //Check for missing "="
        try {
            job = ce.getCoordJob(jobId, "statusRUNNING", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals("E0421: Invalid job filter [statusRUNNING], " +
                    "filter should be of format <key><comparator><value> pairs", ex.getMessage());
        }

        //Check for missing value after "="
        try {
            job = ce.getCoordJob(jobId, "status=", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals("E0421: Invalid job filter [status=], invalid status value []. Valid status values are: ["
                + StringUtils.join(CoordinatorAction.Status.values(), ", ") + "]", ex.getMessage());
        }

        // Check for invalid status value
        try {
            job = ce.getCoordJob(jobId, "status=blahblah", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals("E0421: Invalid job filter [status=blahblah], invalid status value [blahblah]."
                + " Valid status values are: ["
                + StringUtils.join(CoordinatorAction.Status.values(), ", ") + "]", ex.getMessage());
        }

        // Check for empty status value
        try {
            job = ce.getCoordJob(jobId, "status=\"\"", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals("E0421: Invalid job filter [status=\"\"], invalid status value [\"\"]."
                + " Valid status values are: ["
                + StringUtils.join(CoordinatorAction.Status.values(), ", ") + "]", ex.getMessage());
        }

        // Check for invalid filter option
        try {
            job = ce.getCoordJob(jobId, "blahblah=blahblah", 1, 2, false);
        }
        catch (CoordinatorEngineException ex) {
            assertEquals(ErrorCode.E0421, ex.getErrorCode());
            assertEquals("E0421: Invalid job filter [blahblah=blahblah], invalid filter [blahblah]. " +
                "Valid filters [" + StringUtils.join(CoordinatorEngine.VALID_JOB_FILTERS, ", ") + "]", ex.getMessage());
        }
    }
}