package com.orchard.models;

import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.ArrayList;
import java.util.List;
import org.apache.kafka.connect.data.Struct;
import org.junit.jupiter.api.Test;

public class LabelParticipantTests {

    public static final String LabelParticipantJson =
        "{\"name\":\"test\","
        + "\"distributionFlags\":[\"N\"],"
        + "\"id\":123,"
        + "\"uuid\":\"abc123\","
        + "\"globalParticipantName\":\"somegpname\","
        + "\"streamsStats\":"
        + "{\"streams28Days\":28,"
        + "\"streams7Days\":7,"
        + "\"streamsAllTime\":100},"
        + "\"subaccountName\":\"subaccount\","
        + "\"vendorName\":\"vendor\","
        + "\"vendorId\":2,"
        + "\"createdAt\":\"testdate\","
        + "\"lastModifiedAt\":\"testdate\","
        + "\"subaccountId\":1}";

    public static final LabelParticipant CreateLabelParticipant()
    {
        LabelParticipant labelParticipant = new LabelParticipant();
        labelParticipant.name = "test";
        labelParticipant.uuid = "abc123";
        labelParticipant.globalParticipantName = "somegpname";
        labelParticipant.id = 123;
        labelParticipant.subaccountId = 1;
        labelParticipant.vendorId = 2;
        labelParticipant.subaccountName = "subaccount";
        labelParticipant.vendorName = "vendor";
        labelParticipant.distributionFlags = new ArrayList<String>();
        labelParticipant.distributionFlags.add("N");

        labelParticipant.streamsStats = new StreamsStats();
        labelParticipant.streamsStats.streams7Days = 7;
        labelParticipant.streamsStats.streams28Days = 28;
        labelParticipant.streamsStats.streamsAllTime = 100;
        return labelParticipant;
    }

    @Test
    public void shouldReturnValidKafkaStruct() throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        LabelParticipant labelParticipant = CreateLabelParticipant();
        LabelParticipant labelParticipantParsed = mapper.readValue(
                LabelParticipantJson, LabelParticipant.class);
        Struct kafkaStruct = labelParticipant.ToKafkaStruct();

        assertEquals(
                labelParticipantParsed.name,
                kafkaStruct.getString("name"));
        assertEquals(
                labelParticipantParsed.uuid,
                kafkaStruct.getString("uuid"));
        assertEquals(
                labelParticipantParsed.globalParticipantName,
                kafkaStruct.getString("globalParticipantName"));
        assertEquals(
                labelParticipantParsed.id,
                kafkaStruct.getInt64("id"));
        assertEquals(
                labelParticipantParsed.subaccountName,
                kafkaStruct.getString("subaccountName"));
        assertEquals(
                labelParticipantParsed.vendorName,
                kafkaStruct.getString("vendorName"));
        assertEquals(
                labelParticipantParsed.subaccountId,
                kafkaStruct.getInt64("subaccountId"));
        assertEquals(
                labelParticipantParsed.vendorId,
                kafkaStruct.getInt64("vendorId"));
       List<String> distributionFlags = kafkaStruct.getArray("distributionFlags");
       assertArrayEquals(
               labelParticipantParsed.distributionFlags.toArray(), 
               distributionFlags.toArray());
        Struct streamStatsStruct = kafkaStruct.getStruct("streamsStats");
        assertEquals(
                labelParticipantParsed.streamsStats.streams7Days,
                streamStatsStruct.getInt64("streams7Days"));
        assertEquals(
                labelParticipantParsed.streamsStats.streams28Days,
                streamStatsStruct.getInt64("streams28Days"));
        assertEquals(
                labelParticipantParsed.streamsStats.streamsAllTime,
                streamStatsStruct.getInt64("streamsAllTime"));
    }
}