/*
 * 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.kafka.trogdor.common;

import org.apache.kafka.trogdor.fault.FilesUnreadableFaultSpec;
import org.apache.kafka.trogdor.fault.Kibosh;
import org.apache.kafka.trogdor.fault.NetworkPartitionFaultSpec;
import org.apache.kafka.trogdor.fault.ProcessStopFaultSpec;
import org.apache.kafka.trogdor.rest.AgentStatusResponse;
import org.apache.kafka.trogdor.rest.TasksResponse;
import org.apache.kafka.trogdor.rest.WorkerDone;
import org.apache.kafka.trogdor.rest.WorkerRunning;
import org.apache.kafka.trogdor.rest.WorkerStopping;
import org.apache.kafka.trogdor.workload.PartitionsSpec;
import org.apache.kafka.trogdor.workload.ProduceBenchSpec;
import org.apache.kafka.trogdor.workload.RoundTripWorkloadSpec;
import org.apache.kafka.trogdor.workload.TopicsSpec;

import org.junit.jupiter.api.Test;

import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import static org.junit.jupiter.api.Assertions.assertNotNull;

public class JsonSerializationTest {
    @Test
    public void testDeserializationDoesNotProduceNulls() throws Exception {
        verify(new FilesUnreadableFaultSpec(0, 0, null,
            null, null, 0));
        verify(new Kibosh.KiboshControlFile(null));
        verify(new NetworkPartitionFaultSpec(0, 0, null));
        verify(new ProcessStopFaultSpec(0, 0, null, null));
        verify(new AgentStatusResponse(0, null));
        verify(new TasksResponse(null));
        verify(new WorkerDone(null, null, 0, 0, null, null));
        verify(new WorkerRunning(null, null, 0, null));
        verify(new WorkerStopping(null, null, 0, null));
        verify(new ProduceBenchSpec(0, 0, null, null,
            0, 0, null, null, Optional.empty(), null, null, null, null, null, false, false));
        verify(new RoundTripWorkloadSpec(0, 0, null, null, null, null, null, null,
            0, null, null, 0));
        verify(new TopicsSpec());
        verify(new PartitionsSpec(0, (short) 0, null, null));
        Map<Integer, List<Integer>> partitionAssignments = new HashMap<>();
        partitionAssignments.put(0, Arrays.asList(1, 2, 3));
        partitionAssignments.put(1, Arrays.asList(1, 2, 3));
        verify(new PartitionsSpec(0, (short) 0, partitionAssignments, null));
        verify(new PartitionsSpec(0, (short) 0, null, null));
    }

    private <T> void verify(T val1) throws Exception {
        byte[] bytes = JsonUtil.JSON_SERDE.writeValueAsBytes(val1);
        @SuppressWarnings("unchecked")
        Class<T> clazz = (Class<T>) val1.getClass();
        T val2 = JsonUtil.JSON_SERDE.readValue(bytes, clazz);
        for (Field field : clazz.getDeclaredFields()) {
            if (Modifier.isStatic(field.getModifiers())) {
                continue;
            }
            boolean wasAccessible = field.canAccess(val2);
            field.setAccessible(true);
            assertNotNull(field.get(val2), "Field " + field + " was null.");
            field.setAccessible(wasAccessible);
        }
    }
}
