package com.linkedin.camus.etl.kafka.coders;

import com.linkedin.camus.coders.CamusWrapper;
import org.junit.Test;
import static org.junit.Assert.assertEquals;

import java.util.Properties;

public class TestJsonStringMessageDecoder {

    @Test
    public void testDecodeUnixMilliseconds() {

        // Test that the decoder extracts unix_milliseconds
        // It should take and return milliseconds

        long testTimestamp = 1406947271534L;
        long expectedTimestamp = testTimestamp;
        
        Properties testProperties = new Properties();
        testProperties.setProperty("camus.message.timestamp.format", "unix_milliseconds");

        JsonStringMessageDecoder testDecoder = new JsonStringMessageDecoder();
        testDecoder.init(testProperties, "testTopic");
        String payload = "{\"timestamp\":  "+ testTimestamp + ", \"myData\": \"myValue\"}";
        byte[] bytePayload = payload.getBytes();

        CamusWrapper actualResult = testDecoder.decode(bytePayload);
        long actualTimestamp = actualResult.getTimestamp();
        assertEquals(expectedTimestamp, actualTimestamp);
    }

    @Test
    public void testDecodeUnixSeconds() {

        // Test that the decoder extracts unix_seconds
        // It should receive seconds and return milliseconds

        long testTimestamp = 140694727L;
        long expectedTimestamp = 140694727000L;

        Properties testProperties = new Properties();
        testProperties.setProperty("camus.message.timestamp.format", "unix_seconds");

        JsonStringMessageDecoder testDecoder = new JsonStringMessageDecoder();
        testDecoder.init(testProperties, "testTopic");
        String payload = "{\"timestamp\":  "+ testTimestamp + ", \"myData\": \"myValue\"}";
        byte[] bytePayload = payload.getBytes();
        CamusWrapper actualResult = testDecoder.decode(bytePayload);
        long actualTimestamp = actualResult.getTimestamp();

        assertEquals(expectedTimestamp, actualTimestamp);
    }

    @Test public void testDecodeWithTimestampFormat() {

        // Test that we can specify a date and a pattern and
        // get back unix timestamp milliseconds

        String testFormat = "yyyy-MM-dd HH:mm:ss Z";
        String testTimestamp = "2014-02-01 01:15:27 UTC";
        long expectedTimestamp = 1391217327000L;

        Properties testProperties = new Properties();
        testProperties.setProperty("camus.message.timestamp.format", testFormat);

        JsonStringMessageDecoder testDecoder = new JsonStringMessageDecoder();
        testDecoder.init(testProperties, "testTopic");
        String payload = "{\"timestamp\":  \""+ testTimestamp + "\", \"myData\": \"myValue\"}";
        byte[] bytePayload = payload.getBytes();
        CamusWrapper actualResult = testDecoder.decode(bytePayload);
        long actualTimestamp = actualResult.getTimestamp();

        assertEquals(expectedTimestamp, actualTimestamp);


    }

}

@@ -1997,16 +1997,20 @@
    @Test
+%0A   
  public 
@@ -2925,16 +2925,309 @@
 tamp);%0A%0A
+    %7D%0A%0A    @Test(expected=RuntimeException.class)%0A    public void testBadJsonInput() %7B%0A        byte%5B%5D bytePayload = %22%7B%5C%22key: value%7D%22.getBytes();%0A%0A        JsonStringMessageDecoder testDecoder = new JsonStringMessageDecoder();%0A        CamusWrapper actualResult = testDecoder.decode(bytePayload);
 %0A    %7D%0A%0A

test for bad json input