/*
 * 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.hadoop.chukwa.extraction.demux.processor.mapper;


import junit.framework.TestCase;
import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.ChunkBuilder;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
import org.apache.hadoop.chukwa.extraction.demux.Demux;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.JobConf;

import java.util.Map;
import java.util.Date;
import java.util.Calendar;
import java.text.SimpleDateFormat;

public class TestTsProcessor extends TestCase {

  private static String DATA_TYPE = "testDataType";
  private static String DATA_SOURCE = "testDataSource";

  JobConf jobConf = null;

  Date date = null;
  Date dateWithoutMillis = null;

  protected void setUp() throws Exception {
    jobConf = new JobConf();
    Demux.jobConf = jobConf;
    date = new Date();

    //if our format doesn't contain millis, then our final record date won't
    //have them either. let's create a sample date without millis for those tests
    //so our assertions will pass
    Calendar calendar = Calendar.getInstance();
    calendar.setTime(date);
    calendar.set(Calendar.MILLISECOND, 0);
    dateWithoutMillis = calendar.getTime();
  }

  public void testDefaultFormat() {
    String record = buildSampleSimpleRecord(date, "yyyy-MM-dd HH:mm:ss,SSS");
    doTest(date, record);
  }

  public void testCustomDefaultFormat() {
    jobConf.set("TsProcessor.default.time.format", "yyyy--MM--dd HH::mm::ss SSS");

    String record = buildSampleSimpleRecord(date, "yyyy--MM--dd HH::mm::ss SSS");
    doTest(date, record);
  }

  public void testCustomDefaultFormat2() {
    // this date format produces a date that longer than the format, since z
    // expands to something like PDT
    jobConf.set("TsProcessor.default.time.format", "yyyy--MM--dd HH::mm::ss SSS,z");

    String record = buildSampleSimpleRecord(date, "yyyy--MM--dd HH::mm::ss SSS,z");
    doTest(date, record);
  }

  public void testCustomDataTypeFormat() {
    jobConf.set("TsProcessor.time.format." + DATA_TYPE, "yyyy--MM--dd HH::mm::ss SSS");

    String record = buildSampleSimpleRecord(date, "yyyy--MM--dd HH::mm::ss SSS");
    doTest(date, record);
  }

  public void testCustomDefaultFormatWithCustomDataTypeFormat() {
    jobConf.set("TsProcessor.default.time.format", "yyyy/MM/dd HH:mm:ss SSS");
    jobConf.set("TsProcessor.time.format." + DATA_TYPE, "yyyy--MM--dd HH::mm::ss SSS");

    String record = buildSampleSimpleRecord(date, "yyyy--MM--dd HH::mm::ss SSS");
    doTest(date, record);
  }

  public void testCustomApacheDefaultFormat() {
    jobConf.set("TsProcessor.default.time.format", "dd/MMM/yyyy:HH:mm:ss Z");
    jobConf.set("TsProcessor.default.time.regex",
            "^(?:[\\d.]+) \\[(\\d{2}/\\w{3}/\\d{4}:\\d{2}:\\d{2}:\\d{2} [-+]\\d{4})\\] .*");


    String record = buildSampleApacheRecord(dateWithoutMillis, "dd/MMM/yyyy:HH:mm:ss Z");
    doTest(dateWithoutMillis, record);
  }

  public void testCustomApacheDataTypeFormat() {
    jobConf.set("TsProcessor.time.format." + DATA_TYPE, "dd/MMM/yyyy:HH:mm:ss Z");
    jobConf.set("TsProcessor.time.regex." + DATA_TYPE,
            "^(?:[\\d.]+) \\[(\\d{2}/\\w{3}/\\d{4}:\\d{2}:\\d{2}:\\d{2} [-+]\\d{4})\\] .*");


    String record = buildSampleApacheRecord(dateWithoutMillis, "dd/MMM/yyyy:HH:mm:ss Z");
    doTest(dateWithoutMillis, record);
  }

  private static String buildSampleSimpleRecord(Date date, String dateFormat) {
    SimpleDateFormat sdf = new SimpleDateFormat(dateFormat);
    return "" + sdf.format(date) + " some sample record data";
  }

  private static String buildSampleApacheRecord(Date date, String dateFormat) {
    SimpleDateFormat sdf = new SimpleDateFormat(dateFormat);
    return "10.10.182.49 [" + sdf.format(date) +
            "] \"\" 200 \"-\" \"Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10.5; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3\" \"some.site.com:8076\"";
  }

  public void doTest(Date date, String recordData) {
    ChunkBuilder cb = new ChunkBuilder();
    cb.addRecord(recordData.getBytes());

    Chunk chunk = cb.getChunk();
    chunk.setDataType(DATA_TYPE);
    chunk.setSource(DATA_SOURCE);

    ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord> output =
            new ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord>();

    TsProcessor p = new TsProcessor();
    p.reset(chunk);
    p.process(null, chunk, output, Reporter.NULL);

    ChukwaRecordKey key = buildKey(date, DATA_SOURCE, DATA_TYPE);
    Map<ChukwaRecordKey, ChukwaRecord> outputData = output.data;

    assertNotNull("No output data found.", outputData);
    assertEquals("Output data size not correct.", 1, outputData.size());

    ChukwaRecord record = outputData.get(key);
    assertNotNull("Output record not found.", record);
    assertEquals("Output record time not correct.", date.getTime(), record.getTime());
    assertEquals("Output record body not correct.", recordData,
            new String(record.getMapFields().get("body").get()));
  }

  private static ChukwaRecordKey buildKey(Date date, String dataSource, String dataType) {
    Calendar calendar = Calendar.getInstance();
    calendar.setTime(date);
    calendar.set(Calendar.MINUTE, 0);
    calendar.set(Calendar.SECOND, 0);
    calendar.set(Calendar.MILLISECOND, 0);

    ChukwaRecordKey key = new ChukwaRecordKey();
    key.setKey("" + calendar.getTimeInMillis() + "/" + dataSource + "/" + date.getTime());
    key.setReduceType(dataType);

    return key;
  }

  public void testParseIllegalRegex() {
    jobConf.set(TsProcessor.DEFAULT_TIME_REGEX, "(");

    ChunkBuilder cb = new ChunkBuilder();
    cb.addRecord("2012-10-25 00:18:44,818 some sample record data".getBytes());
    Chunk chunk = cb.getChunk();

    TsProcessor p = new TsProcessor();
    p.reset(chunk);

    ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord> output =
            new ChukwaTestOutputCollector<ChukwaRecordKey, ChukwaRecord>();

    p.process(null, chunk, output, Reporter.NULL);

    assertEquals("Output data size not correct.", 1, output.data.size());
    ChukwaRecordKey key = output.data.keySet().iterator().next();
    ChukwaRecord record = output.data.get(key);
    assertNull("Output should not be error.", record.getValue("cchunkData"));
  }
}