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 GSLabelParticipantTests {

    public static final String LabelParticipantJson =
            "{\"name\":\"test\","
                    + "\"distributionFlags\":[\"N\"],"
                    + "\"id\":123,"
                    + "\"uuid\":\"abc123\","
                    + "\"spotifyId\":\"spot123\","
                    + "\"roles\":[\"admin\"],"
                    + "\"createdAt\":\"testdate\","
                    + "\"lastModifiedAt\":\"testdate\","
                    + "\"streamsStats\":"
                    + "{\"streams28Days\":28,"
                    + "\"streams7Days\":7,"
                    + "\"streamsAllTime\":100},"
                    + "\"vendorId\":2,"
                    + "\"subaccountId\":1,"
                    + "\"globalParticipantName\":\"gpname\","
                    + "\"globalParticipantUuid\":\"gpuuid\"}";

    public static GSLabelParticipant CreateGSLabelParticipant() {
        GSLabelParticipant labelParticipant = new GSLabelParticipant();
        labelParticipant.name = "test";
        labelParticipant.globalParticipantUuid = "gpuuid";
        labelParticipant.globalParticipantName = "gpname";
        labelParticipant.uuid = "abc123";
        labelParticipant.spotifyId = "spot123";
        labelParticipant.id = 123;
        labelParticipant.subaccountId = 1;
        labelParticipant.vendorId = 2;
        labelParticipant.roles = new ArrayList<String>();
        labelParticipant.roles.add("admin");
        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();
        GSLabelParticipant labelParticipantParsed = mapper.readValue(
                LabelParticipantJson, GSLabelParticipant.class);
        GSLabelParticipant labelParticipant = CreateGSLabelParticipant();
        Struct kafkaStruct = labelParticipant.ToKafkaStruct();

        assertEquals(
                labelParticipantParsed.name,
                kafkaStruct.getString("name"));
        assertEquals(
                labelParticipantParsed.uuid,
                kafkaStruct.getString("uuid"));
        assertEquals(
                labelParticipantParsed.spotifyId,
                kafkaStruct.getString("spotifyId"));
        assertEquals(
                labelParticipantParsed.id,
                kafkaStruct.getInt64("id"));
        assertEquals(
                labelParticipantParsed.globalParticipantUuid,
                kafkaStruct.getString("globalParticipantUuid"));
        assertEquals(
                labelParticipantParsed.globalParticipantName,
                kafkaStruct.getString("globalParticipantName"));
        assertEquals(
                labelParticipantParsed.subaccountId,
                kafkaStruct.getInt64("subaccountId"));
        assertEquals(
                labelParticipantParsed.vendorId,
                kafkaStruct.getInt64("vendorId"));
        List<String> distributionFlags = kafkaStruct.getArray("distributionFlags");
        assertArrayEquals(
                labelParticipantParsed.distributionFlags.toArray(), 
                distributionFlags.toArray());
        List<String> roles = kafkaStruct.getArray("roles");
        assertArrayEquals(labelParticipantParsed.roles.toArray(), roles.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"));
    }
}