/*
 * Licensed to the University of California, Berkeley 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 tachyon.client.next;

import java.io.IOException;

import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;

import tachyon.Constants;
import tachyon.TachyonURI;
import tachyon.client.WriteType;
import tachyon.client.next.file.FileInStream;
import tachyon.client.next.file.TachyonFS;
import tachyon.conf.TachyonConf;
import tachyon.master.next.LocalTachyonCluster;
import tachyon.util.io.BufferUtils;
import tachyon.util.io.PathUtils;

/**
 * Integration tests for <code>tachyon.client.BlockInStream</code>.
 */
public class BlockInStreamIntegrationTest {
  private static final int MIN_LEN = 0;
  private static final int MAX_LEN = 255;
  private static final int MEAN = (MIN_LEN + MAX_LEN) / 2;
  private static final int DELTA = 33;

  private static LocalTachyonCluster sLocalTachyonCluster = null;
  private static TachyonFS sTfs = null;

  @AfterClass
  public static final void afterClass() throws Exception {
    sLocalTachyonCluster.stop();
    System.clearProperty("tachyon.user.quota.unit.bytes");
  }

  @BeforeClass
  public static final void beforeClass() throws Exception {
    sLocalTachyonCluster = new LocalTachyonCluster(10000, 1000, Constants.GB);
    sLocalTachyonCluster.start();
    sTfs = sLocalTachyonCluster.getClient();
  }

  /**
   * Test <code>void read()</code>.
   */
  @Test
  public void newReadTest1() throws IOException {
    System.out.println("========Starting new Test=======");
    tachyon.client.next.file.TachyonFS tfs = tachyon.client.next.file.TachyonFS.get();
    String uniqPath = PathUtils.uniqPath();
    int i = 0;
    for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
      for (WriteType op : WriteType.values()) {
        System.out.println("Iteration: " + i ++);
        TachyonURI path = new TachyonURI(uniqPath + "/file_" + k + "_" + op);
        ClientOptions options = new ClientOptions.Builder(new TachyonConf()).build();
        tachyon.client.next.file.FileOutStream os = tfs.getOutStream(path, options);
        System.out.println("==== Got output stream ====");
        for (int j = 0; j < k; j ++) {
          os.write((byte) j);
        }
        System.out.println("==== Wrote data! ====");
        os.close();
        System.out.println("==== Closed Stream! ====");

        tachyon.client.next.file.TachyonFile f = tfs.open(path);

        FileInStream is = tfs.getInStream(f, options);
        byte[] ret = new byte[k];
        int value = is.read();
        int cnt = 0;
        while (value != -1) {
          Assert.assertTrue(value >= 0);
          Assert.assertTrue(value < 256);
          ret[cnt ++] = (byte) value;
          value = is.read();
        }
        Assert.assertEquals(cnt, k);
        Assert.assertTrue(BufferUtils.equalIncreasingByteArray(k, ret));
        is.close();

        is = tfs.getInStream(f, options);
        ret = new byte[k];
        value = is.read();
        cnt = 0;
        while (value != -1) {
          Assert.assertTrue(value >= 0);
          Assert.assertTrue(value < 256);
          ret[cnt ++] = (byte) value;
          value = is.read();
        }
        Assert.assertEquals(cnt, k);
        Assert.assertTrue(BufferUtils.equalIncreasingByteArray(k, ret));
        is.close();
      }
    }
    System.out.println("========== This test is over!");
  }
}

@@ -1117,25 +1117,117 @@
 xt.file.
-TachyonFS
+FileOutStream;%0Aimport tachyon.client.next.file.TachyonFS;%0Aimport tachyon.client.next.file.TachyonFile
 ;%0Aimport
@@ -2298,155 +2298,8 @@
 n %7B%0A
-    System.out.println(%22========Starting new Test=======%22);%0A    tachyon.client.next.file.TachyonFS tfs = tachyon.client.next.file.TachyonFS.get();%0A
     
@@ -2342,23 +2342,8 @@
 ();%0A
-    int i = 0;%0A
     
@@ -2444,58 +2444,8 @@
 ) %7B%0A
-        System.out.println(%22Iteration: %22 + i ++);%0A
     
@@ -2616,33 +2616,8 @@
     
-tachyon.client.next.file.
 File
@@ -2627,25 +2627,26 @@
 Stream os = 
-t
+sT
 fs.getOutStr
@@ -2669,67 +2669,8 @@
 s);%0A
-        System.out.println(%22==== Got output stream ====%22);%0A
     
@@ -2756,163 +2756,29 @@
     
-System.out.println(%22==== Wrote data! ====%22);%0A        os.close();%0A        System.out.println(%22==== Closed Stream! ====%22);%0A%0A        tachyon.client.next.file.
+os.close();%0A%0A        
 Tach
@@ -2789,17 +2789,18 @@
 ile f = 
-t
+sT
 fs.open(
@@ -2829,25 +2829,26 @@
 Stream is = 
-t
+sT
 fs.getInStre
@@ -3283,17 +3283,18 @@
    is = 
-t
+sT
 fs.getIn
@@ -3722,65 +3722,8 @@
   %7D%0A
-    System.out.println(%22========== This test is over!%22);%0A
   %7D%0A

Clean up sample test case.