/**
 * test that a dir off root has a listStatus() call that
 * works as expected. and that when a child is added. it changes
 *
 * @throws Exception on failures
 */
@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testDirectoriesOffRootHaveMatchingFileStatus() throws Exception {
  Path test = path("/test");
  fs.delete(test, true);
  mkdirs(test);
  assertExists("created test directory", test);
  FileStatus[] statuses = fs.listStatus(test);
  String statusString = statusToString(test.toString(), statuses);
  assertEquals("Wrong number of elements in file status " + statusString, 0,
               statuses.length);

  Path src = path("/test/file");

  //create a zero byte file
  SwiftTestUtils.touch(fs, src);
  //stat it
  statuses = fs.listStatus(test);
  statusString = statusToString(test.toString(), statuses);
  assertEquals("Wrong number of elements in file status " + statusString, 1,
               statuses.length);
  SwiftFileStatus stat = (SwiftFileStatus) statuses[0];
  assertTrue("isDir(): Not a directory: " + stat, stat.isDir());
  extraStatusAssertions(stat);
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testDeleteSmallPartitionedFile() throws Throwable {
  final Path path = new Path("/test/testDeleteSmallPartitionedFile");

  final int len1 = 1024;
  final byte[] src1 = SwiftTestUtils.dataset(len1, 'A', 'Z');
  SwiftTestUtils.writeDataset(fs, path, src1, len1, 1024, false);
  assertExists("Exists", path);

  Path part_0001 = new Path(path, SwiftUtils.partitionFilenameFromNumber(1));
  Path part_0002 = new Path(path, SwiftUtils.partitionFilenameFromNumber(2));
  String ls = SwiftTestUtils.ls(fs, path);
  assertExists("Partition 0001 Exists in " + ls, part_0001);
  assertPathDoesNotExist("partition 0002 found under " + ls, part_0002);
  assertExists("Partition 0002 Exists in " + ls, part_0001);
  fs.delete(path, false);
  assertPathDoesNotExist("deleted file still there", path);
  ls = SwiftTestUtils.ls(fs, path);
  assertPathDoesNotExist("partition 0001 file still under " + ls, part_0001);
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testDeleteSmallPartitionedFile() throws Throwable {
  final Path path = new Path("/test/testDeleteSmallPartitionedFile");

  final int len1 = 1024;
  final byte[] src1 = SwiftTestUtils.dataset(len1, 'A', 'Z');
  SwiftTestUtils.writeDataset(fs, path, src1, len1, 1024, false);
  assertExists("Exists", path);

  Path part_0001 = new Path(path, SwiftUtils.partitionFilenameFromNumber(1));
  Path part_0002 = new Path(path, SwiftUtils.partitionFilenameFromNumber(2));
  String ls = SwiftTestUtils.ls(fs, path);
  assertExists("Partition 0001 Exists in " + ls, part_0001);
  assertPathDoesNotExist("partition 0002 found under " + ls, part_0002);
  assertExists("Partition 0002 Exists in " + ls, part_0001);
  fs.delete(path, false);
  assertPathDoesNotExist("deleted file still there", path);
  ls = SwiftTestUtils.ls(fs, path);
  assertPathDoesNotExist("partition 0001 file still under " + ls, part_0001);
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testDeletePartitionedFile() throws Throwable {
  final Path path = new Path("/test/testDeletePartitionedFile");

  SwiftTestUtils.writeDataset(fs, path, data, data.length, 1024, false);
  assertExists("Exists", path);

  Path part_0001 = new Path(path, SwiftUtils.partitionFilenameFromNumber(1));
  Path part_0002 = new Path(path, SwiftUtils.partitionFilenameFromNumber(2));
  String ls = SwiftTestUtils.ls(fs, path);
  assertExists("Partition 0001 Exists in " + ls, part_0001);
  assertExists("Partition 0002 Exists in " + ls, part_0001);
  fs.delete(path, false);
  assertPathDoesNotExist("deleted file still there", path);
  ls = SwiftTestUtils.ls(fs, path);
  assertPathDoesNotExist("partition 0001 file still under " + ls, part_0001);
  assertPathDoesNotExist("partition 0002 file still under " + ls, part_0002);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testSeekBigFile() throws Throwable {
  Path testSeekFile = new Path(testPath, "bigseekfile.txt");
  byte[] block = SwiftTestUtils.dataset(65536, 0, 255);
  createFile(testSeekFile, block);
  instream = fs.open(testSeekFile);
  assertEquals(0, instream.getPos());
  //expect that seek to 0 works
  instream.seek(0);
  int result = instream.read();
  assertEquals(0, result);
  assertEquals(1, instream.read());
  assertEquals(2, instream.read());

  //do seek 32KB ahead
  instream.seek(32768);
  assertEquals("@32768", block[32768], (byte) instream.read());
  instream.seek(40000);
  assertEquals("@40000", block[40000], (byte) instream.read());
  instream.seek(8191);
  assertEquals("@8191", block[8191], (byte) instream.read());
  instream.seek(0);
  assertEquals("@0", 0, (byte) instream.read());
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testPositionedBulkReadDoesntChangePosition() throws Throwable {
  Path testSeekFile = new Path(testPath, "bigseekfile.txt");
  byte[] block = SwiftTestUtils.dataset(65536, 0, 255);
  createFile(testSeekFile, block);
  instream = fs.open(testSeekFile);
  instream.seek(39999);
  assertTrue(-1 != instream.read());
  assertEquals (40000, instream.getPos());

  byte[] readBuffer = new byte[256];
  instream.read(128, readBuffer, 0, readBuffer.length);
  //have gone back
  assertEquals(40000, instream.getPos());
  //content is the same too
  assertEquals("@40000", block[40000], (byte) instream.read());
  //now verify the picked up data
  for (int i = 0; i < 256; i++) {
    assertEquals("@" + i, block[i + 128], readBuffer[i]);
  }
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testDeletePartitionedFile() throws Throwable {
  final Path path = new Path("/test/testDeletePartitionedFile");

  SwiftTestUtils.writeDataset(fs, path, data, data.length, 1024, false);
  assertExists("Exists", path);

  Path part_0001 = new Path(path, SwiftUtils.partitionFilenameFromNumber(1));
  Path part_0002 = new Path(path, SwiftUtils.partitionFilenameFromNumber(2));
  String ls = SwiftTestUtils.ls(fs, path);
  assertExists("Partition 0001 Exists in " + ls, part_0001);
  assertExists("Partition 0002 Exists in " + ls, part_0001);
  fs.delete(path, false);
  assertPathDoesNotExist("deleted file still there", path);
  ls = SwiftTestUtils.ls(fs, path);
  assertPathDoesNotExist("partition 0001 file still under " + ls, part_0001);
  assertPathDoesNotExist("partition 0002 file still under " + ls, part_0002);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testSeekBigFile() throws Throwable {
  Path testSeekFile = new Path(testPath, "bigseekfile.txt");
  byte[] block = SwiftTestUtils.dataset(65536, 0, 255);
  createFile(testSeekFile, block);
  instream = fs.open(testSeekFile);
  assertEquals(0, instream.getPos());
  //expect that seek to 0 works
  instream.seek(0);
  int result = instream.read();
  assertEquals(0, result);
  assertEquals(1, instream.read());
  assertEquals(2, instream.read());

  //do seek 32KB ahead
  instream.seek(32768);
  assertEquals("@32768", block[32768], (byte) instream.read());
  instream.seek(40000);
  assertEquals("@40000", block[40000], (byte) instream.read());
  instream.seek(8191);
  assertEquals("@8191", block[8191], (byte) instream.read());
  instream.seek(0);
  assertEquals("@0", 0, (byte) instream.read());
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testRenamedConsistence() throws IOException {
  assumeRenameSupported();
  describe("verify that overwriting a file with new data doesn't impact" +
          " the existing content");

  final Path filePath = new Path("/test/home/user/documents/file.txt");
  final Path newFilePath = new Path("/test/home/user/files/file.txt");
  mkdirs(newFilePath.getParent());
  int len = 1024;
  byte[] dataset = dataset(len, 'A', 26);
  byte[] dataset2 = dataset(len, 'a', 26);
  writeDataset(fs, filePath, dataset, len, len, false);
  rename(filePath, newFilePath, true, false, true);
  SwiftTestUtils.writeAndRead(fs, filePath, dataset2, len, len, false, true);
  byte[] dest = readDataset(fs, newFilePath, len);
  compareByteArrays(dataset, dest, len);
  String reread = readBytesToString(fs, newFilePath, 20);
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testDeletePartitionedFile() throws Throwable {
  final Path path = new Path("/test/testDeletePartitionedFile");

  SwiftTestUtils.writeDataset(fs, path, data, data.length, 1024, false);
  assertExists("Exists", path);

  Path part_0001 = new Path(path, SwiftUtils.partitionFilenameFromNumber(1));
  Path part_0002 = new Path(path, SwiftUtils.partitionFilenameFromNumber(2));
  String ls = SwiftTestUtils.ls(fs, path);
  assertExists("Partition 0001 Exists in " + ls, part_0001);
  assertExists("Partition 0002 Exists in " + ls, part_0001);
  fs.delete(path, false);
  assertPathDoesNotExist("deleted file still there", path);
  ls = SwiftTestUtils.ls(fs, path);
  assertPathDoesNotExist("partition 0001 file still under " + ls, part_0001);
  assertPathDoesNotExist("partition 0002 file still under " + ls, part_0002);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testSeekBigFile() throws Throwable {
  Path testSeekFile = new Path(testPath, "bigseekfile.txt");
  byte[] block = SwiftTestUtils.dataset(65536, 0, 255);
  createFile(testSeekFile, block);
  instream = fs.open(testSeekFile);
  assertEquals(0, instream.getPos());
  //expect that seek to 0 works
  instream.seek(0);
  int result = instream.read();
  assertEquals(0, result);
  assertEquals(1, instream.read());
  assertEquals(2, instream.read());

  //do seek 32KB ahead
  instream.seek(32768);
  assertEquals("@32768", block[32768], (byte) instream.read());
  instream.seek(40000);
  assertEquals("@40000", block[40000], (byte) instream.read());
  instream.seek(8191);
  assertEquals("@8191", block[8191], (byte) instream.read());
  instream.seek(0);
  assertEquals("@0", 0, (byte) instream.read());
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testRenamedConsistence() throws IOException {
  assumeRenameSupported();
  describe("verify that overwriting a file with new data doesn't impact" +
          " the existing content");

  final Path filePath = new Path("/test/home/user/documents/file.txt");
  final Path newFilePath = new Path("/test/home/user/files/file.txt");
  mkdirs(newFilePath.getParent());
  int len = 1024;
  byte[] dataset = dataset(len, 'A', 26);
  byte[] dataset2 = dataset(len, 'a', 26);
  writeDataset(fs, filePath, dataset, len, len, false);
  rename(filePath, newFilePath, true, false, true);
  SwiftTestUtils.writeAndRead(fs, filePath, dataset2, len, len, false, true);
  byte[] dest = readDataset(fs, newFilePath, len);
  compareByteArrays(dataset, dest, len);
  String reread = readBytesToString(fs, newFilePath, 20);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testBlocksizeNonZeroForFile() throws Throwable {
  Path smallfile = new Path("/test/smallfile");
  SwiftTestUtils.writeTextFile(fs, smallfile, "blocksize", true);
  createFile(smallfile);
  FileStatus status = getFs().getFileStatus(smallfile);
  assertTrue("Zero blocksize in " + status,
             status.getBlockSize() != 0L);
  assertTrue("Zero replication in " + status,
             status.getReplication() != 0L);
}
 

/**
 * Asserts that a zero byte file has a status of file and not
 * directory or symlink
 *
 * @throws Exception on failures
 */
@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testZeroByteFilesAreDirectories() throws Exception {
  Path src = path("/test/testZeroByteFilesAreFiles");
  //create a zero byte file
  SwiftTestUtils.touch(fs, src);
  SwiftTestUtils.assertIsDirectory(fs, src);
}
 

/**
 * Asserts that a zero byte file has a status of file and not
 * directory or symlink
 *
 * @throws Exception on failures
 */
@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testMultiByteFilesAreFiles() throws Exception {
  Path src = path("/test/testMultiByteFilesAreFiles");
  SwiftTestUtils.writeTextFile(fs, src, "testMultiByteFilesAreFiles", false);
  assertIsFile(src);
  FileStatus status = fs.getFileStatus(src);
  assertFalse(status.isDir());
}
 

/**
 * Setup creates dirs under test/hadoop
 *
 * @throws Exception
 */
@Override
public void setUp() throws Exception {
  super.setUp();
  //delete the test directory
  testPath = path("/test");
  smallSeekFile = new Path(testPath, "seekfile.txt");
  zeroByteFile = new Path(testPath, "zero.txt");
  byte[] block = SwiftTestUtils.dataset(SMALL_SEEK_FILE_LEN, 0, 255);
  //this file now has a simple rule: offset => value
  createFile(smallSeekFile, block);
  createEmptyFile(zeroByteFile);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testDeleteEmptyFile() throws IOException {
  final Path file = new Path("/test/testDeleteEmptyFile");
  createEmptyFile(file);
  SwiftTestUtils.noteAction("about to delete");
  assertDeleted(file, true);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testDeleteEmptyFileTwice() throws IOException {
  final Path file = new Path("/test/testDeleteEmptyFileTwice");
  createEmptyFile(file);
  assertDeleted(file, true);
  SwiftTestUtils.noteAction("multiple creates, and deletes");
  assertFalse("Delete returned true", fs.delete(file, false));
  createEmptyFile(file);
  assertDeleted(file, true);
  assertFalse("Delete returned true", fs.delete(file, false));
}
 

@Override
public void testWriteReadAndDeleteEmptyFile() throws Exception {
  try {
    super.testWriteReadAndDeleteEmptyFile();
  } catch (AssertionFailedError e) {
    SwiftTestUtils.downgrade("empty files get mistaken for directories", e);
  }
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testLocateRootDirectory() throws Throwable {
  describe("verify that locating the root directory is an error");
  FileStatus status = fs.getFileStatus(path("/"));
  SwiftTestUtils.assertIsDirectory(status);
  BlockLocation[] locations;
  locations = getFs().getFileBlockLocations(status,
                                            0,
                                            1);
  assertEmptyBlockLocations(locations);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testFileStatus() throws Throwable {
  Path path = new Path("/test/FileStatus");
  try {
    String text = "Testing File Status "
            + System.currentTimeMillis();
    writeTextFile(fs, path, text, false);
    SwiftTestUtils.assertIsFile(fs, path);
  } finally {
    delete(fs, path);
  }
}
 

/**
 * Test sticks up a very large partitioned file and verifies that
 * it comes back unchanged.
 * @throws Throwable
 */
@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testManyPartitionedFile() throws Throwable {
  final Path path = new Path("/test/testManyPartitionedFile");

  int len = PART_SIZE_BYTES * 15;
  final byte[] src = SwiftTestUtils.dataset(len, 32, 144);
  FSDataOutputStream out = fs.create(path,
                                     false,
                                     getBufferSize(),
                                     (short) 1,
                                     BLOCK_SIZE);

  out.write(src, 0, src.length);
  int expected =
    getExpectedPartitionsWritten(len, PART_SIZE_BYTES, true);
  out.close();
  assertPartitionsWritten("write completed", out, expected);
  assertEquals("too few bytes written", len,
               SwiftNativeFileSystem.getBytesWritten(out));
  assertEquals("too few bytes uploaded", len,
               SwiftNativeFileSystem.getBytesUploaded(out));
  //now we verify that the data comes back. If it
  //doesn't, it means that the ordering of the partitions
  //isn't right
  byte[] dest = readDataset(fs, path, len);
  //compare data
  SwiftTestUtils.compareByteArrays(src, dest, len);
  //finally, check the data
  FileStatus[] stats = fs.listStatus(path);
  assertEquals("wrong entry count in "
               + SwiftTestUtils.dumpStats(path.toString(), stats),
               expected, stats.length);
}
 

/**
 * Test that when a partitioned file is overwritten by a smaller one,
 * all the old partitioned files go away
 * @throws Throwable
 */
@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testOverwritePartitionedFile() throws Throwable {
  final Path path = new Path("/test/testOverwritePartitionedFile");

  final int len1 = 8192;
  final byte[] src1 = SwiftTestUtils.dataset(len1, 'A', 'Z');
  FSDataOutputStream out = fs.create(path,
                                     false,
                                     getBufferSize(),
                                     (short) 1,
                                     1024);
  out.write(src1, 0, len1);
  out.close();
  long expected = getExpectedPartitionsWritten(len1,
                                               PART_SIZE_BYTES,
                                               false);
  assertPartitionsWritten("initial upload", out, expected);
  assertExists("Exists", path);
  FileStatus status = fs.getFileStatus(path);
  assertEquals("Length", len1, status.getLen());
  //now write a shorter file with a different dataset
  final int len2 = 4095;
  final byte[] src2 = SwiftTestUtils.dataset(len2, 'a', 'z');
  out = fs.create(path,
                  true,
                  getBufferSize(),
                  (short) 1,
                  1024);
  out.write(src2, 0, len2);
  out.close();
  status = fs.getFileStatus(path);
  assertEquals("Length", len2, status.getLen());
  byte[] dest = readDataset(fs, path, len2);
  //compare data
  SwiftTestUtils.compareByteArrays(src2, dest, len2);
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testRenamePartitionedFile() throws Throwable {
  Path src = new Path("/test/testRenamePartitionedFileSrc");

  int len = data.length;
  SwiftTestUtils.writeDataset(fs, src, data, len, 1024, false);
  assertExists("Exists", src);

  String partOneName = SwiftUtils.partitionFilenameFromNumber(1);
  Path srcPart = new Path(src, partOneName);
  Path dest = new Path("/test/testRenamePartitionedFileDest");
  Path destPart = new Path(src, partOneName);
  assertExists("Partition Exists", srcPart);
  fs.rename(src, dest);
  assertPathExists(fs, "dest file missing", dest);
  FileStatus status = fs.getFileStatus(dest);
  assertEquals("Length of renamed file is wrong", len, status.getLen());
  byte[] destData = readDataset(fs, dest, len);
  //compare data
  SwiftTestUtils.compareByteArrays(data, destData, len);
  String srcLs = SwiftTestUtils.ls(fs, src);
  String destLs = SwiftTestUtils.ls(fs, dest);

  assertPathDoesNotExist("deleted file still found in " + srcLs, src);

  assertPathDoesNotExist("partition file still found in " + srcLs, srcPart);
}
 

/**
 * Setup creates dirs under test/hadoop
 *
 * @throws Exception
 */
@Override
public void setUp() throws Exception {
  super.setUp();
  //delete the test directory
  testPath = path("/test");
  smallSeekFile = new Path(testPath, "seekfile.txt");
  zeroByteFile = new Path(testPath, "zero.txt");
  byte[] block = SwiftTestUtils.dataset(SMALL_SEEK_FILE_LEN, 0, 255);
  //this file now has a simple rule: offset => value
  createFile(smallSeekFile, block);
  createEmptyFile(zeroByteFile);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testListFilesRootDir() throws Throwable {
  Path dir = path("/");
  Path child = new Path(dir, "test");
  fs.delete(child, true);
  SwiftTestUtils.writeTextFile(fs, child, "text", false);
  assertListFilesFinds(fs, dir, child, false);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testListFilesSubDir() throws Throwable {
  createTestSubdirs();
  Path dir = path("/test/subdir");
  Path child = new Path(dir, "text.txt");
  SwiftTestUtils.writeTextFile(fs, child, "text", false);
  assertListFilesFinds(fs, dir, child, false);
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testListFilesRecursive() throws Throwable {
  createTestSubdirs();
  Path dir = path("/test/recursive");
  Path child = new Path(dir, "hadoop/a/a.txt");
  SwiftTestUtils.writeTextFile(fs, child, "text", false);
  assertListFilesFinds(fs, dir, child, true);
}
 

@Override
@Before
public void setUp() throws Exception {
  Configuration conf = new Configuration();
  //small blocksize for faster remote tests
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, 2);
  URI serviceURI = SwiftTestUtils.getServiceURI(conf);
  fSys = FileSystem.get(serviceURI, conf);
  super.setUp();
}
 

@Test(timeout = SWIFT_TEST_TIMEOUT)
public void testLocateDirectory() throws Throwable {
  describe("verify that locating a directory is an error");
  createFile(path("/test/filename"));
  FileStatus status = fs.getFileStatus(path("/test"));
  LOG.info("Filesystem is " + fs + "; target is " + status);
  SwiftTestUtils.assertIsDirectory(status);
  BlockLocation[] locations;
  locations = getFs().getFileBlockLocations(status,
                                            0,
                                            1);
  assertEmptyBlockLocations(locations);
}