ClassPathResourceToHdfsCopier implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } }  @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts();  }
@Test(expected = NullPointerException.class) public void givenParametersWithoutInputClassPathResource_whenRunIsCalled_thenCheckFails() throws Exception { classPathResourceToHdfsCopier.run(mock(PortBindings.class), conf, new HashMap<>()); }
@Test(expected = NullPointerException.class) public void givenParametersWithoutOutputHdfsFileLocation_whenRunIsCalled_thenCheckFails() throws Exception { HashMap<String, String> parameters = new HashMap<>(); parameters.put("inputClasspathResource", "/path/to/resource"); classPathResourceToHdfsCopier.run(mock(PortBindings.class), conf, parameters); }
@Test public void givenParametersWithValues_whenRunIsCalled_thenResourceIsCopiedToLocation() throws Exception { Path tempDirectory = Files.createTempDirectory(this.getClass().getSimpleName()); Path file = tempDirectory.resolve("file.tmp"); HashMap<String, String> parameters = new HashMap<>(); parameters.put("inputClasspathResource", "/path/to/resource"); parameters.put("outputHdfsFileLocation", file.toString()); when(classPathResourceProvider.apply("/path/to/resource")).thenReturn( new ByteArrayInputStream("resource content".getBytes(StandardCharsets.UTF_8))); classPathResourceToHdfsCopier.run(mock(PortBindings.class), conf, parameters); assertTrue(Files.exists(file)); assertEquals("resource content", FileUtils.readFileToString(file.toFile(), StandardCharsets.UTF_8)); }