/*
 * 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.dataloader;

import junit.framework.TestCase;
import java.util.Calendar;
import org.apache.hadoop.chukwa.database.Macro;
import org.apache.hadoop.chukwa.util.DatabaseWriter;
import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.chukwa.util.ExceptionUtil;
import org.apache.hadoop.chukwa.database.TableCreator;
import org.apache.hadoop.chukwa.dataloader.MetricDataLoader;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;

public class TestDatabaseMetricDataLoader extends TestCase {

  long[] timeWindow = {7, 30, 91, 365, 3650};
  String[] tables = {"system_metrics","disk","mr_job","mr_task"}; //,"dfs_namenode","dfs_datanode","dfs_fsnamesystem","dfs_throughput","hadoop_jvm","hadoop_mapred","hdfs_usage"};
  String cluster = "demo";
  long current = Calendar.getInstance().getTimeInMillis();

  public void setUp() {
    System.setProperty("CLUSTER","demo");
    DatabaseWriter db = new DatabaseWriter(cluster);
    String buffer = "";
    File aFile = new File(System.getenv("CHUKWA_CONF_DIR")
                 + File.separator + "database_create_tables.sql");
    buffer = readFile(aFile);
    String tables[] = buffer.split(";");
    for(String table : tables) {
      if(table.length()>5) {
        try {
          db.execute(table);
        } catch (Exception e) {
          fail("Fail to retrieve meta data from database table: "+table);
        }
      }
    }
    db.close();
    for(int i=0;i<timeWindow.length;i++) {
      TableCreator tc = new TableCreator();
      long start = current;
      long end = current + (timeWindow[i]*1440*60*1000);
      try {
        tc.createTables(start, end);
      } catch (Exception e) {
        fail("Fail to create database tables.");
      }
    }
  }

  public void tearDown() {
    DatabaseWriter db = null;
    try {
      db = new DatabaseWriter(cluster);
      ResultSet rs = db.query("show tables");
      ArrayList<String> list = new ArrayList<String>();
      while(rs.next()) {
        String table = rs.getString(1);
        list.add(table);
      }
      for(String table : list) {
        db.execute("drop table "+table);
      }
    } catch(Throwable ex) {
    } finally {
      if(db!=null) {
        db.close();
      }
    }
  }

  public String readFile(File aFile) {
    StringBuffer contents = new StringBuffer();
    try {
      BufferedReader input = new BufferedReader(new FileReader(aFile));
      try {
        String line = null; // not declared within while loop
        while ((line = input.readLine()) != null) {
          contents.append(line);
          contents.append(System.getProperty("line.separator"));
        }
      } finally {
        input.close();
      }
    } catch (IOException ex) {
      ex.printStackTrace();
    }
    return contents.toString();
  }

  public void testMetricDataLoader() {
    boolean skip=false;
    String srcDir = System.getenv("CHUKWA_DATA_DIR") + File.separator + "samples";
    try {
      ChukwaConfiguration conf = new ChukwaConfiguration();
      FileSystem fs = FileSystem.get(conf);
      FileStatus[] sources = fs.listStatus(new Path(srcDir));
      for (FileStatus sequenceFile : sources) {
        MetricDataLoader mdl = new MetricDataLoader(conf, fs, sequenceFile.getPath().toUri().toString());
        mdl.call();
      }
      if(sources.length==0) {
        skip=true;
      }
    } catch (Throwable ex) {
      fail("SQL Exception: "+ExceptionUtil.getStackTrace(ex));
    }
    if(!skip) {
      DatabaseWriter db = new DatabaseWriter(cluster);
      for(int i=0;i<tables.length;i++) {
        String query = "select [avg("+tables[i]+")] from ["+tables[i]+"]";
        Macro mp = new Macro(current,query);
        query = mp.toString();
        try {
          ResultSet rs = db.query(query);
          ResultSetMetaData rsmd = rs.getMetaData();
          int numberOfColumns = rsmd.getColumnCount();
          while(rs.next()) {
            for(int j=1;j<=numberOfColumns;j++) {
              assertTrue("Table: "+tables[i]+", Column: "+rsmd.getColumnName(j)+", contains no data.",rs.getString(j)!=null);
            }
          }
        } catch(Throwable ex) {
          fail("MetricDataLoader failed: "+ExceptionUtil.getStackTrace(ex));
        }
      }
      db.close();
      assertTrue("MetricDataLoader executed successfully.",true);
    }
  }

}