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

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

    public static final LabelSoundRecording CreateLabelSoundRecording() {
        LabelSoundRecording soundRecording = new LabelSoundRecording();
        soundRecording.name = "test";
        soundRecording.uuid = "abc123";
        soundRecording.subaccountId = 1;
        soundRecording.vendorId = 2;
        soundRecording.subaccountName = "subaccount";
        soundRecording.vendorName = "vendor";
        soundRecording.distributionFlags = new ArrayList<String>();
        soundRecording.distributionFlags.add("N");
        soundRecording.streamsStats = new StreamsStats();
        soundRecording.streamsStats.streams7Days = 7;
        soundRecording.streamsStats.streams28Days = 28;
        soundRecording.streamsStats.streamsAllTime = 100;
        return soundRecording;
    }

    @Test
    public void shouldReturnValidKafkaStruct() throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        LabelSoundRecording soundRecordingParsed = mapper.readValue(
                LabelSoundRecordingJson, LabelSoundRecording.class);

        LabelSoundRecording soundRecording = CreateLabelSoundRecording();
        Struct kafkaStruct = soundRecording.ToKafkaStruct();

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