/*
 * 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.nifi.processors.hadoop;

import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import com.google.common.collect.Maps;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.hadoop.KerberosProperties;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class TestGetHDFSFileInfo {
    private static final Pattern SINGLE_JSON_PATTERN = Pattern.compile("^\\{[^\\}]*\\}$");

    private TestRunner runner;
    private GetHDFSFileInfoWithMockedFileSystem proc;
    private NiFiProperties mockNiFiProperties;
    private KerberosProperties kerberosProperties;

    @Before
    public void setup() throws InitializationException {
        mockNiFiProperties = mock(NiFiProperties.class);
        when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null);
        kerberosProperties = new KerberosProperties(null);

        proc = new GetHDFSFileInfoWithMockedFileSystem(kerberosProperties);
        runner = TestRunners.newTestRunner(proc);

        runner.setProperty(GetHDFSFileInfo.HADOOP_CONFIGURATION_RESOURCES, "src/test/resources/core-site.xml");
    }

    @Test
    public void testInvalidBatchSizeWhenDestinationAndGroupingDoesntAllowBatchSize() throws Exception {
        Arrays.asList("1", "2", "100").forEach(
            validBatchSize -> {
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_ALL, validBatchSize, false);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_PARENT_DIR, validBatchSize, false);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_NONE, validBatchSize, false);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_ALL, validBatchSize, false);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_PARENT_DIR, validBatchSize, false);
            }
        );
    }

    @Test
    public void testInvalidBatchSizeWhenValueIsInvalid() throws Exception {
        Arrays.asList("-1", "0", "someString").forEach(
            inValidBatchSize -> {
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, inValidBatchSize, false);
            }
        );
    }

    @Test
    public void testValidBatchSize() throws Exception {
        Arrays.asList("1", "2", "100").forEach(
            validBatchSize -> {
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, validBatchSize, true);
            }
        );

        Arrays.asList((String)null).forEach(
            nullBatchSize -> {
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_ALL, nullBatchSize, true);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_PARENT_DIR, nullBatchSize, true);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_NONE, nullBatchSize, true);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_ALL, nullBatchSize, true);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_PARENT_DIR, nullBatchSize, true);
                testValidateBatchSize(GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, nullBatchSize, true);
            }
        );
    }

    private void testValidateBatchSize(AllowableValue destination, AllowableValue grouping, String batchSize, boolean expectedValid) {
        runner.clearProperties();

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, destination);
        runner.setProperty(GetHDFSFileInfo.GROUPING, grouping);
        if (batchSize != null) {
            runner.setProperty(GetHDFSFileInfo.BATCH_SIZE, "" + batchSize);
        }

        if (expectedValid) {
            runner.assertValid();
        } else {
            runner.assertNotValid();
        }
    }

    @Test
    public void testNoRunOnIncomingConnectionExists() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(true);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "${literal('/some/home/mydir'):substring(0,15)}");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);
    }

    @Test
    public void testRunOnScheduleNoConnections() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "false");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 1);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);
    }

    @Test
    public void testValidELFunction() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(true);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "${literal('/some/home/mydir'):substring(0,16)}");
        runner.setProperty(GetHDFSFileInfo.DIR_FILTER, "${literal('_^(dir.*)$_'):substring(1,10)}");
        runner.setProperty(GetHDFSFileInfo.FILE_FILTER, "${literal('_^(.*)$_'):substring(1,7)}");
        runner.setProperty(GetHDFSFileInfo.FILE_EXCLUDE_FILTER, "${literal('_^(none.*)$_'):substring(1,11)}");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "false");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);
        runner.enqueue("foo", new HashMap<String,String>());

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 1);

        final MockFlowFile mff = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_ORIGINAL).get(0);
        ProcessContext context = runner.getProcessContext();

        assertEquals(context.getProperty(GetHDFSFileInfo.FULL_PATH).evaluateAttributeExpressions(mff).getValue(), "/some/home/mydir");
        assertEquals(context.getProperty(GetHDFSFileInfo.DIR_FILTER).evaluateAttributeExpressions(mff).getValue(), "^(dir.*)$");
        assertEquals(context.getProperty(GetHDFSFileInfo.FILE_FILTER).evaluateAttributeExpressions(mff).getValue(), "^(.*)$");
        assertEquals(context.getProperty(GetHDFSFileInfo.FILE_EXCLUDE_FILTER).evaluateAttributeExpressions(mff).getValue(), "^(none.*)$");
    }

    @Test
    public void testRunWithConnections() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        Map<String, String> attributes = Maps.newHashMap();
        attributes.put("input.dir", "/some/home/mydir");

        runner.setIncomingConnection(true);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "${input.dir}");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "false");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.enqueue("foo", attributes);
        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 1);

        final MockFlowFile mff = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_ORIGINAL).get(0);
        ProcessContext context = runner.getProcessContext();

        assertEquals(context.getProperty(GetHDFSFileInfo.FULL_PATH).evaluateAttributeExpressions(mff).getValue(), "/some/home/mydir");
    }

    @Test
    public void testRunWithIOException() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);
        proc.fileSystem.addFileStatus(proc.fileSystem.newDir("/some/home/mydir"), proc.fileSystem.newFile("/some/home/mydir/exception_java.io.InterruptedIOException"));

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir/exception_java.io.InterruptedIOException");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "false");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 1);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);

        final MockFlowFile mff = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_FAILURE).get(0);
        mff.assertAttributeEquals("hdfs.status", "Failed due to: java.io.InterruptedIOException");
    }

    @Test
    public void testRunWithPermissionsExceptionAttributes() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);
        proc.fileSystem.addFileStatus(proc.fileSystem.newDir("/some/home/mydir/dir1"), proc.fileSystem.newDir("/some/home/mydir/dir1/list_exception_java.io.InterruptedIOException"));

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_ATTRIBUTES);
        runner.setProperty(GetHDFSFileInfo.GROUPING, GetHDFSFileInfo.GROUP_PARENT_DIR);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 6);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);
    }

    @Test
    public void testRunWithPermissionsExceptionContent() throws Exception {

        setFileSystemBasicTree(proc.fileSystem);
        proc.fileSystem.addFileStatus(proc.fileSystem.newDir("/some/home/mydir/dir1"), proc.fileSystem.newDir("/some/home/mydir/dir1/list_exception_java.io.InterruptedIOException"));

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);
        runner.setProperty(GetHDFSFileInfo.GROUPING, GetHDFSFileInfo.GROUP_ALL);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 1);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);

        final MockFlowFile mff = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_SUCCESS).get(0);
        mff.assertContentEquals(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRunWithPermissionsExceptionContent.json"));
    }

    @Test
    public void testObjectNotFound() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir/ObjectThatDoesNotExist");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 1);
    }

    @Test
    public void testRecursive() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_CONTENT);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 1);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);
    }

    @Test
    public void testRecursiveGroupAllToAttributes() throws Exception {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_ATTRIBUTES);
        runner.setProperty(GetHDFSFileInfo.GROUPING, GetHDFSFileInfo.GROUP_ALL);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 1);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);

        final MockFlowFile mff = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_SUCCESS).get(0);
        mff.assertAttributeEquals("hdfs.objectName", "mydir");
        mff.assertAttributeEquals("hdfs.path", "/some/home");
        mff.assertAttributeEquals("hdfs.type", "directory");
        mff.assertAttributeEquals("hdfs.owner", "owner");
        mff.assertAttributeEquals("hdfs.group", "group");
        mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
        mff.assertAttributeEquals("hdfs.length", ""+900);
        mff.assertAttributeEquals("hdfs.count.files", ""+9);
        mff.assertAttributeEquals("hdfs.count.dirs", ""+10);
        mff.assertAttributeEquals("hdfs.replication", ""+3);
        mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
        mff.assertAttributeNotExists("hdfs.status");

        final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupAllToAttributes.json")));
        mff.assertAttributeEquals("hdfs.full.tree", expected);
    }

    @Test
    public void testRecursiveGroupNoneToAttributes() throws InterruptedException {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_ATTRIBUTES);
        runner.setProperty(GetHDFSFileInfo.GROUPING, GetHDFSFileInfo.GROUP_NONE);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 9);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);

        int matchCount = 0;
        for (final MockFlowFile mff : runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_SUCCESS)) {
            if (mff.getAttribute("hdfs.objectName").equals("mydir")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+900);
                mff.assertAttributeEquals("hdfs.count.files", ""+9);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+10);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("dir1")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+200);
                mff.assertAttributeEquals("hdfs.count.files", ""+2);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+3);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("dir2")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+200);
                mff.assertAttributeEquals("hdfs.count.files", ""+2);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+3);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("regDir")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir1");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+0);
                mff.assertAttributeEquals("hdfs.count.files", ""+0);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+1);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("regDir2")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir2");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+0);
                mff.assertAttributeEquals("hdfs.count.files", ""+0);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+1);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("regFile")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir1");
                mff.assertAttributeEquals("hdfs.type", "file");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+100);
                mff.assertAttributeNotExists("hdfs.count.files");
                mff.assertAttributeNotExists("hdfs.count.dirs");
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rw-r--r--");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("regFile2")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir2");
                mff.assertAttributeEquals("hdfs.type", "file");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+100);
                mff.assertAttributeNotExists("hdfs.count.files");
                mff.assertAttributeNotExists("hdfs.count.dirs");
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rw-r--r--");
                mff.assertAttributeNotExists("hdfs.status");
            }else if (mff.getAttribute("hdfs.objectName").equals("regFile4")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "file");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+100);
                mff.assertAttributeNotExists("hdfs.count.files");
                mff.assertAttributeNotExists("hdfs.count.dirs");
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rw-r--r--");
                mff.assertAttributeNotExists("hdfs.status");
              }else if (mff.getAttribute("hdfs.objectName").equals("regFile5")) {
              matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "file");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+100);
                mff.assertAttributeNotExists("hdfs.count.files");
                mff.assertAttributeNotExists("hdfs.count.dirs");
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rw-r--r--");
                mff.assertAttributeNotExists("hdfs.status");
            }else {
               runner.assertNotValid();
            }
        }
        Assert.assertEquals(matchCount, 9);
    }

    @Test
    public void testRecursiveGroupDirToAttributes() throws Exception {

        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, GetHDFSFileInfo.DESTINATION_ATTRIBUTES);
        runner.setProperty(GetHDFSFileInfo.GROUPING, GetHDFSFileInfo.GROUP_PARENT_DIR);

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, 5);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);

        int matchCount = 0;
        for (final MockFlowFile mff : runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_SUCCESS)) {
            if (mff.getAttribute("hdfs.objectName").equals("mydir")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+900);
                mff.assertAttributeEquals("hdfs.count.files", ""+9);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+10);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
                final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupDirToAttributes-mydir.json")));
                mff.assertAttributeEquals("hdfs.full.tree", expected);
            }else if (mff.getAttribute("hdfs.objectName").equals("dir1")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+200);
                mff.assertAttributeEquals("hdfs.count.files", ""+2);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+3);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
                final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupDirToAttributes-dir1.json")));
                mff.assertAttributeEquals("hdfs.full.tree", expected);
            }else if (mff.getAttribute("hdfs.objectName").equals("dir2")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+200);
                mff.assertAttributeEquals("hdfs.count.files", ""+2);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+3);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
                final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupDirToAttributes-dir2.json")));
                mff.assertAttributeEquals("hdfs.full.tree", expected);
            }else if (mff.getAttribute("hdfs.objectName").equals("regDir")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir1");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+0);
                mff.assertAttributeEquals("hdfs.count.files", ""+0);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+1);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
                final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupDirToAttributes-regDir.json")));
                mff.assertAttributeEquals("hdfs.full.tree", expected);
            }else if (mff.getAttribute("hdfs.objectName").equals("regDir2")) {
                matchCount++;
                mff.assertAttributeEquals("hdfs.path", "/some/home/mydir/dir2");
                mff.assertAttributeEquals("hdfs.type", "directory");
                mff.assertAttributeEquals("hdfs.owner", "owner");
                mff.assertAttributeEquals("hdfs.group", "group");
                mff.assertAttributeEquals("hdfs.lastModified", ""+1523456000000L);
                mff.assertAttributeEquals("hdfs.length", ""+0);
                mff.assertAttributeEquals("hdfs.count.files", ""+0);
                mff.assertAttributeEquals("hdfs.count.dirs", ""+1);
                mff.assertAttributeEquals("hdfs.replication", ""+3);
                mff.assertAttributeEquals("hdfs.permissions", "rwxr-xr-x");
                mff.assertAttributeNotExists("hdfs.status");
                final String expected = new String(Files.readAllBytes(Paths.get("src/test/resources/TestGetHDFSFileInfo/testRecursiveGroupDirToAttributes-regDir2.json")));
                mff.assertAttributeEquals("hdfs.full.tree", expected);
            }else {
               runner.assertNotValid();
            }
        }
        Assert.assertEquals(matchCount, 5);
    }

    @Test
    public void testBatchSizeWithDestAttributesGroupAllBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_ALL, 1);
    }

    @Test
    public void testBatchSizeWithDestAttributesGroupDirBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_PARENT_DIR, 5);
    }

    @Test
    public void testBatchSizeWithDestAttributesGroupNoneBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_ATTRIBUTES, GetHDFSFileInfo.GROUP_NONE, 9);
    }

    @Test
    public void testBatchSizeWithDestContentGroupAllBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_ALL, 1);
    }

    @Test
    public void testBatchSizeWithDestContentGroupDirBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_PARENT_DIR, 5);
    }

    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSizeNull() throws Exception {
        testBatchSize(null, GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 9);

        checkContentSizes(Arrays.asList(1, 1, 1, 1, 1, 1, 1, 1, 1));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize1() throws Exception {
        testBatchSize("1", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 9);
        checkContentSizes(Arrays.asList(1, 1, 1, 1, 1, 1, 1, 1, 1));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize3() throws Exception {
        testBatchSize("3", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 3);
        checkContentSizes(Arrays.asList(3, 3, 3));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize4() throws Exception {
        testBatchSize("4", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 3);
        checkContentSizes(Arrays.asList(4, 4, 1));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize5() throws Exception {
        testBatchSize("5", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 2);
        checkContentSizes(Arrays.asList(5, 4));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize9() throws Exception {
        testBatchSize("9", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 1);
        checkContentSizes(Arrays.asList(9));
    }
    @Test
    public void testBatchSizeWithDestContentGroupNoneBatchSize100() throws Exception {
        testBatchSize("100", GetHDFSFileInfo.DESTINATION_CONTENT, GetHDFSFileInfo.GROUP_NONE, 1);
        checkContentSizes(Arrays.asList(9));
    }

    private void testBatchSize(String batchSize, AllowableValue destination, AllowableValue grouping, int expectedNrTransferredToSuccess) {
        setFileSystemBasicTree(proc.fileSystem);

        runner.setIncomingConnection(false);
        runner.setProperty(GetHDFSFileInfo.FULL_PATH, "/some/home/mydir");
        runner.setProperty(GetHDFSFileInfo.RECURSE_SUBDIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_DIRS, "true");
        runner.setProperty(GetHDFSFileInfo.IGNORE_DOTTED_FILES, "true");
        runner.setProperty(GetHDFSFileInfo.DESTINATION, destination);
        runner.setProperty(GetHDFSFileInfo.GROUPING, grouping);
        if (batchSize != null) {
            runner.setProperty(GetHDFSFileInfo.BATCH_SIZE, batchSize);
        }

        runner.run();

        runner.assertTransferCount(GetHDFSFileInfo.REL_ORIGINAL, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_SUCCESS, expectedNrTransferredToSuccess);
        runner.assertTransferCount(GetHDFSFileInfo.REL_FAILURE, 0);
        runner.assertTransferCount(GetHDFSFileInfo.REL_NOT_FOUND, 0);
    }

    private void checkContentSizes(List<Integer> expectedNumberOfRecords) {
        List<Integer> actualNumberOfRecords = runner.getFlowFilesForRelationship(GetHDFSFileInfo.REL_SUCCESS).stream()
            .map(MockFlowFile::toByteArray)
            .map(String::new)
            .map(
                content -> Arrays.stream(content.split("\n"))
                    .filter(line -> SINGLE_JSON_PATTERN.matcher(line).matches())
                    .count()
            )
            .map(Long::intValue)
            .collect(Collectors.toList());

        assertEquals(expectedNumberOfRecords, actualNumberOfRecords);
    }

    /*
     * For all basic tests, this provides a structure of files in dirs:
     * Total number of dirs:  9 (1 root, 4 dotted)
     * Total number of files: 8 (4 dotted)
     */
    protected void setFileSystemBasicTree(final MockFileSystem fs) {
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newFile("/some/home/mydir/regFile4"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newFile("/some/home/mydir/regFile5"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newFile("/some/home/mydir/.dotFile4"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newFile("/some/home/mydir/.dotFile5"));

        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newDir("/some/home/mydir/dir1"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newDir("/some/home/mydir/dir2"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newDir("/some/home/mydir/.dir3"));
        fs.addFileStatus(fs.newDir("/some/home/mydir"), fs.newDir("/some/home/mydir/.dir4"));

        fs.addFileStatus(fs.newDir("/some/home/mydir/dir1"), fs.newFile("/some/home/mydir/dir1/.dotFile"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir1"), fs.newDir("/some/home/mydir/dir1/.dotDir"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir1"), fs.newFile("/some/home/mydir/dir1/regFile"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir1"), fs.newDir("/some/home/mydir/dir1/regDir"));

        fs.addFileStatus(fs.newDir("/some/home/mydir/dir2"), fs.newFile("/some/home/mydir/dir2/.dotFile2"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir2"), fs.newDir("/some/home/mydir/dir2/.dotDir2"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir2"), fs.newFile("/some/home/mydir/dir2/regFile2"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/dir2"), fs.newDir("/some/home/mydir/dir2/regDir2"));

        fs.addFileStatus(fs.newDir("/some/home/mydir/.dir3"), fs.newFile("/some/home/mydir/.dir3/regFile3"));
        fs.addFileStatus(fs.newDir("/some/home/mydir/.dir3"), fs.newDir("/some/home/mydir/.dir3/regDir3"));
    }

    static FsPermission perms(short p) {
        return new FsPermission(p);
    }


    private class GetHDFSFileInfoWithMockedFileSystem extends GetHDFSFileInfo {
        private final MockFileSystem fileSystem = new MockFileSystem();
        private final KerberosProperties testKerberosProps;

        public GetHDFSFileInfoWithMockedFileSystem(KerberosProperties kerberosProperties) {
            this.testKerberosProps = kerberosProperties;
        }

        @Override
        protected KerberosProperties getKerberosProperties(File kerberosConfigFile) {
            return testKerberosProps;
        }

        @Override
        protected FileSystem getFileSystem() {
            return fileSystem;
        }

        @Override
        protected FileSystem getFileSystem(final Configuration config) throws IOException {
            return fileSystem;
        }
    }

    private class MockFileSystem extends FileSystem {
        private final Map<Path, Set<FileStatus>> fileStatuses = new HashMap<>();
        private final Map<Path, FileStatus> pathToStatus = new HashMap<>();

        public void addFileStatus(final FileStatus parent, final FileStatus child) {
            Set<FileStatus> children = fileStatuses.get(parent.getPath());
            if (children == null) {
                children = new HashSet<>();
                fileStatuses.put(parent.getPath(), children);
            }
            if (child != null) {
                children.add(child);
                if (child.isDirectory() && !fileStatuses.containsKey(child.getPath())) {
                    fileStatuses.put(child.getPath(), new HashSet<FileStatus>());
                }
            }

            pathToStatus.put(parent.getPath(), parent);
            pathToStatus.put(child.getPath(), child);
        }

        @Override
        @SuppressWarnings("deprecation")
        public long getDefaultBlockSize() {
            return 1024L;
        }

        @Override
        @SuppressWarnings("deprecation")
        public short getDefaultReplication() {
            return 1;
        }

        @Override
        public URI getUri() {
            return null;
        }

        @Override
        public FSDataInputStream open(final Path f, final int bufferSize) throws IOException {
            return null;
        }

        @Override
        public FSDataOutputStream create(final Path f, final FsPermission permission, final boolean overwrite, final int bufferSize, final short replication,
                                         final long blockSize, final Progressable progress) throws IOException {
            return null;
        }

        @Override
        public FSDataOutputStream append(final Path f, final int bufferSize, final Progressable progress) throws IOException {
            return null;
        }

        @Override
        public boolean rename(final Path src, final Path dst) throws IOException {
            return false;
        }

        @Override
        public boolean delete(final Path f, final boolean recursive) throws IOException {
            return false;
        }

        @Override
        public FileStatus[] listStatus(final Path f) throws FileNotFoundException, IOException {
            if (!fileStatuses.containsKey(f)) {
                throw new FileNotFoundException();
            }
            if (f.getName().startsWith("list_exception_")) {
                String clzName = f.getName().substring("list_exception_".length(), f.getName().length());
                IOException exception = null;
                try {
                     exception = (IOException)Class.forName(clzName).newInstance();
                } catch (Throwable t) {
                    throw new RuntimeException(t);
                }
                throw exception;
            }
            final Set<FileStatus> statuses = fileStatuses.get(f);
            if (statuses == null) {
                return new FileStatus[0];
            }

            for (FileStatus s : statuses) {
                getFileStatus(s.getPath()); //support exception handling only.
            }

            return statuses.toArray(new FileStatus[statuses.size()]);
        }

        @Override
        public void setWorkingDirectory(final Path new_dir) {

        }

        @Override
        public Path getWorkingDirectory() {
            return new Path(new File(".").getAbsolutePath());
        }

        @Override
        public boolean mkdirs(final Path f, final FsPermission permission) throws IOException {
            return false;
        }

        @Override
        public FileStatus getFileStatus(final Path f) throws IOException {
            if (f!=null && f.getName().startsWith("exception_")) {
                String clzName = f.getName().substring("exception_".length(), f.getName().length());
                IOException exception = null;
                try {
                     exception = (IOException)Class.forName(clzName).newInstance();
                } catch (Throwable t) {
                    throw new RuntimeException(t);
                }
                throw exception;
            }
            final FileStatus fileStatus = pathToStatus.get(f);
            if (fileStatus == null) throw new FileNotFoundException();
            return fileStatus;
        }

        public FileStatus newFile(String p) {
            return new FileStatus(100L, false, 3, 128*1024*1024, 1523456000000L, 1523457000000L, perms((short)0644), "owner", "group", new Path(p));
        }
        public FileStatus newDir(String p) {
            return new FileStatus(1L, true, 3, 128*1024*1024, 1523456000000L, 1523457000000L, perms((short)0755), "owner", "group", new Path(p));
        }
    }
}