/*
 * 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.flink.runtime.messages;

import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.ApplicationState;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.messages.webmonitor.ApplicationDetails;
import org.apache.flink.runtime.messages.webmonitor.ApplicationDetailsInfo;
import org.apache.flink.runtime.messages.webmonitor.ClusterOverview;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.JobIdsWithStatusOverview;
import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
import org.apache.flink.runtime.messages.webmonitor.MultipleApplicationsDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.messages.webmonitor.RequestJobDetails;
import org.apache.flink.runtime.messages.webmonitor.RequestJobsOverview;
import org.apache.flink.runtime.messages.webmonitor.RequestJobsWithIDsOverview;
import org.apache.flink.runtime.messages.webmonitor.RequestStatusOverview;
import org.apache.flink.util.CollectionUtil;

import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

class WebMonitorMessagesTest {

    @Test
    void testStatusMessages() throws Exception {
        final Random rnd = new Random();

        GenericMessageTester.testMessageInstance(RequestJobsOverview.getInstance());
        GenericMessageTester.testMessageInstance(RequestJobsWithIDsOverview.getInstance());
        GenericMessageTester.testMessageInstance(RequestStatusOverview.getInstance());
        GenericMessageTester.testMessageInstance(RequestJobsOverview.getInstance());

        GenericMessageTester.testMessageInstance(
                GenericMessageTester.instantiateGeneric(RequestJobDetails.class, rnd));
        GenericMessageTester.testMessageInstance(
                GenericMessageTester.instantiateGeneric(ClusterOverview.class, rnd));
        GenericMessageTester.testMessageInstance(
                GenericMessageTester.instantiateGeneric(JobsOverview.class, rnd));

        GenericMessageTester.testMessageInstance(
                new JobIdsWithStatusOverview(
                        Arrays.asList(
                                new JobIdsWithStatusOverview.JobIdWithStatus(
                                        JobID.generate(), JobStatus.RUNNING),
                                new JobIdsWithStatusOverview.JobIdWithStatus(
                                        JobID.generate(), JobStatus.CANCELED),
                                new JobIdsWithStatusOverview.JobIdWithStatus(
                                        JobID.generate(), JobStatus.CREATED),
                                new JobIdsWithStatusOverview.JobIdWithStatus(
                                        JobID.generate(), JobStatus.FAILED),
                                new JobIdsWithStatusOverview.JobIdWithStatus(
                                        JobID.generate(), JobStatus.RESTARTING))));
    }

    @Test
    void testJobDetailsMessage() throws Exception {
        final Random rnd = new Random();

        int[] numVerticesPerState = new int[ExecutionState.values().length];
        int numTotal = 0;

        for (int i = 0; i < numVerticesPerState.length; i++) {
            int count = rnd.nextInt(55);
            numVerticesPerState[i] = count;
            numTotal += count;
        }

        long time = rnd.nextLong();
        long endTime = rnd.nextBoolean() ? -1L : time + rnd.nextInt();
        long lastModified = endTime == -1 ? time + rnd.nextInt() : endTime;

        String name = GenericMessageTester.randomString(rnd);
        JobID jid = GenericMessageTester.randomJobId(rnd);
        JobStatus status = GenericMessageTester.randomJobStatus(rnd);

        JobDetails msg1 =
                new JobDetails(
                        jid,
                        name,
                        time,
                        endTime,
                        endTime - time,
                        status,
                        lastModified,
                        numVerticesPerState,
                        numTotal);
        JobDetails msg2 =
                new JobDetails(
                        jid,
                        name,
                        time,
                        endTime,
                        endTime - time,
                        status,
                        lastModified,
                        numVerticesPerState,
                        numTotal);

        GenericMessageTester.testMessageInstances(msg1, msg2);
    }

    @Test
    void testMultipleJobDetails() throws Exception {
        final Random rnd = new Random();
        GenericMessageTester.testMessageInstance(new MultipleJobsDetails(randomJobDetails(rnd)));
    }

    private Collection<JobDetails> randomJobDetails(Random rnd) {
        final JobDetails[] details = new JobDetails[rnd.nextInt(10)];
        for (int k = 0; k < details.length; k++) {
            int[] numVerticesPerState = new int[ExecutionState.values().length];
            int numTotal = 0;

            for (int i = 0; i < numVerticesPerState.length; i++) {
                int count = rnd.nextInt(55);
                numVerticesPerState[i] = count;
                numTotal += count;
            }

            long time = rnd.nextLong();
            long endTime = rnd.nextBoolean() ? -1L : time + rnd.nextInt();
            long lastModified = endTime == -1 ? time + rnd.nextInt() : endTime;

            String name = new GenericMessageTester.StringInstantiator().instantiate(rnd);
            JobID jid = new JobID();
            JobStatus status = JobStatus.values()[rnd.nextInt(JobStatus.values().length)];

            details[k] =
                    new JobDetails(
                            jid,
                            name,
                            time,
                            endTime,
                            endTime - time,
                            status,
                            lastModified,
                            numVerticesPerState,
                            numTotal);
        }
        return Arrays.asList(details);
    }

    @Test
    void testApplicationDetailsMessage() throws Exception {
        final Random rnd = new Random();

        Map<String, Integer> jobInfo = new HashMap<>();

        for (int i = 0; i < JobStatus.values().length; i++) {
            int count = rnd.nextInt(55);
            jobInfo.put(JobStatus.values()[i].name(), count);
        }

        long time = rnd.nextLong();
        long endTime = rnd.nextBoolean() ? -1L : time + rnd.nextInt();

        String name = GenericMessageTester.randomString(rnd);
        ApplicationID id = GenericMessageTester.randomApplicationId(rnd);
        ApplicationState status = GenericMessageTester.randomApplicationState(rnd);

        ApplicationDetails msg1 =
                new ApplicationDetails(
                        id, name, time, endTime, endTime - time, status.toString(), jobInfo);
        ApplicationDetails msg2 =
                new ApplicationDetails(
                        id, name, time, endTime, endTime - time, status.toString(), jobInfo);

        GenericMessageTester.testMessageInstances(msg1, msg2);
    }

    @Test
    void testApplicationDetailsInfoMessage() throws Exception {
        final Random rnd = new Random();

        Map<String, Long> timestamps =
                CollectionUtil.newHashMapWithExpectedSize(ApplicationState.values().length);

        for (ApplicationState status : ApplicationState.values()) {
            timestamps.put(status.toString(), rnd.nextLong());
        }

        long time = rnd.nextLong();
        long endTime = rnd.nextBoolean() ? -1L : time + rnd.nextInt();

        String name = GenericMessageTester.randomString(rnd);
        ApplicationID id = GenericMessageTester.randomApplicationId(rnd);
        ApplicationState status = GenericMessageTester.randomApplicationState(rnd);
        Collection<JobDetails> jobs = randomJobDetails(rnd);

        ApplicationDetailsInfo msg1 =
                new ApplicationDetailsInfo(
                        id,
                        name,
                        status.toString(),
                        time,
                        endTime,
                        endTime - time,
                        timestamps,
                        jobs);
        ApplicationDetailsInfo msg2 =
                new ApplicationDetailsInfo(
                        id,
                        name,
                        status.toString(),
                        time,
                        endTime,
                        endTime - time,
                        timestamps,
                        jobs);

        GenericMessageTester.testMessageInstances(msg1, msg2);
    }

    @Test
    void testMultipleApplicationDetails() throws Exception {
        final Random rnd = new Random();
        GenericMessageTester.testMessageInstance(
                new MultipleApplicationsDetails(randomApplicationDetails(rnd)));
    }

    private Collection<ApplicationDetails> randomApplicationDetails(Random rnd) {
        final ApplicationDetails[] details = new ApplicationDetails[rnd.nextInt(10)];
        for (int k = 0; k < details.length; k++) {
            Map<String, Integer> jobInfo = new HashMap<>();

            for (int i = 0; i < JobStatus.values().length; i++) {
                int count = rnd.nextInt(55);
                jobInfo.put(JobStatus.values()[i].name(), count);
            }

            long time = rnd.nextLong();
            long endTime = rnd.nextBoolean() ? -1L : time + rnd.nextInt();

            String name = GenericMessageTester.randomString(rnd);
            ApplicationID id = GenericMessageTester.randomApplicationId(rnd);
            ApplicationState status = GenericMessageTester.randomApplicationState(rnd);

            details[k] =
                    new ApplicationDetails(
                            id, name, time, endTime, endTime - time, status.toString(), jobInfo);
        }
        return Arrays.asList(details);
    }
}
