package tachyon.client;
import java.io.IOException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import tachyon.TestUtils;
import tachyon.master.LocalTachyonCluster;
/**
* Unit tests for <code>tachyon.client.FileInStream</code>.
*/
public class FileInStreamTest {
private static final int BLOCK_SIZE = 30;
private static final int MIN_LEN = BLOCK_SIZE + 1;
private static final int MAX_LEN = 255;
private static final int MEAN = (MIN_LEN + MAX_LEN) / 2;
private static final int DELTA = 33;
private LocalTachyonCluster mLocalTachyonCluster = null;
private TachyonFS mTfs = null;
@Rule
public ExpectedException thrown = ExpectedException.none();
@After
public final void after() throws Exception {
mLocalTachyonCluster.stop();
System.clearProperty("tachyon.user.quota.unit.bytes");
System.clearProperty("tachyon.user.default.block.size.byte");
}
@Before
public final void before() throws IOException {
System.setProperty("tachyon.user.quota.unit.bytes", "1000");
System.setProperty("tachyon.user.default.block.size.byte", String.valueOf(BLOCK_SIZE));
mLocalTachyonCluster = new LocalTachyonCluster(10000);
mLocalTachyonCluster.start();
mTfs = mLocalTachyonCluster.getClient();
}
/**
* Test <code>void read()</code>.
*/
@Test
public void readTest1() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
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(TestUtils.equalIncreasingByteArray(k, ret));
is.close();
is = (k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
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(TestUtils.equalIncreasingByteArray(k, ret));
is.close();
}
}
}
/**
* Test <code>void read(byte[] b)</code>.
*/
@Test
public void readTest2() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
byte[] ret = new byte[k];
Assert.assertEquals(k, is.read(ret));
Assert.assertTrue(TestUtils.equalIncreasingByteArray(k, ret));
is.close();
is = (k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
ret = new byte[k];
Assert.assertEquals(k, is.read(ret));
Assert.assertTrue(TestUtils.equalIncreasingByteArray(k, ret));
is.close();
}
}
}
/**
* Test <code>void read(byte[] b, int off, int len)</code>.
*/
@Test
public void readTest3() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
byte[] ret = new byte[k / 2];
Assert.assertEquals(k / 2, is.read(ret, 0, k / 2));
Assert.assertTrue(TestUtils.equalIncreasingByteArray(k / 2, ret));
is.close();
is = (k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
ret = new byte[k];
Assert.assertEquals(k, is.read(ret, 0, k));
Assert.assertTrue(TestUtils.equalIncreasingByteArray(k, ret));
is.close();
}
}
}
/**
* Test <code>void read(byte[] b, int off, int len)</code> for end of file.
*/
@Test
public void readEndOfFileTest() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
try {
byte[] ret = new byte[k / 2];
int readBytes = is.read(ret, 0, k / 2);
while (readBytes != -1) {
readBytes = is.read(ret);
Assert.assertTrue(0 != readBytes);
}
Assert.assertEquals(-1, readBytes);
} finally {
is.close();
}
}
}
}
/**
* Test <code>void seek(long pos)</code>. Validate the expected exception for seeking a negative
* position.
*
* @throws IOException
*/
@Test
public void seekExceptionTest1() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
try {
is.seek(-1);
} catch (IOException e) {
// This is expected
continue;
}
is.close();
throw new IOException("Except seek IOException");
}
}
}
/**
* Test <code>void seek(long pos)</code>. Validate the expected exception for seeking a position
* that is past EOF.
*
* @throws IOException
*/
@Test
public void seekExceptionTest2() throws IOException {
thrown.expect(IOException.class);
thrown.expectMessage("Seek position is past EOF");
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
is.seek(k + 1);
is.close();
}
}
}
/**
* Test <code>void seek(long pos)</code>.
*
* @throws IOException
*/
@Test
public void seekTest() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
is.seek(k / 3);
Assert.assertEquals(k / 3, is.read());
is.seek(k / 2);
Assert.assertEquals(k / 2, is.read());
is.seek(k / 4);
Assert.assertEquals(k / 4, is.read());
is.close();
}
}
}
/**
* Test <code>long skip(long len)</code>.
*/
@Test
public void skipTest() throws IOException {
for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
for (WriteType op : WriteType.values()) {
int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k);
TachyonFile file = mTfs.getFile(fileId);
InStream is =
(k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
Assert.assertEquals(k / 2, is.skip(k / 2));
Assert.assertEquals(k / 2, is.read());
is.close();
is = (k < MEAN ? file.getInStream(ReadType.CACHE) : file.getInStream(ReadType.NO_CACHE));
Assert.assertTrue(is instanceof FileInStream);
Assert.assertEquals(k / 3, is.skip(k / 3));
Assert.assertEquals(k / 3, is.read());
is.close();
}
}
}
}