### Question:
DeepSparkContext extends JavaSparkContext implements Serializable { public RDD<Cells> createHDFSRDD(ExtractorConfig<Cells> config) { Serializable host = config.getValues().get(ExtractorConstants.HOST); Serializable port = config.getValues().get(ExtractorConstants.PORT); Serializable path = config.getValues().get(ExtractorConstants.FS_FILE_PATH); final TextFileDataTable textFileDataTable = UtilFS.createTextFileMetaDataFromConfig(config, this); String filePath = path.toString(); if (config.getExtractorImplClassName().equals(ExtractorConstants.HDFS)) { filePath = ExtractorConstants.HDFS_PREFIX + host.toString() + ":" + port + path.toString(); } return createRDDFromFilePath(filePath, textFileDataTable); } DeepSparkContext(SparkContext sc);  DeepSparkContext(String master, String appName);  DeepSparkContext(String master, String appName, String sparkHome, String jarFile);  DeepSparkContext(String master, String appName, String sparkHome, String[] jars);  DeepSparkContext(String master, String appName, String sparkHome, String[] jars,
                            Map<String, String> environment);  DeepSparkContext(DeepSparkConfig deepSparkConfig); @Override SparkContext sc(); RDD<T> createRDD(ExtractorConfig<T> config); RDD<T> createRDD(DeepJobConfig<T, ?> deepJobConfig); JavaRDD<T> createJavaRDD(
            ExtractorConfig<T> config); JavaRDD<T> createJavaRDD(DeepJobConfig<T, ?> config); static JavaRDD<Row> createJavaRowRDD(JavaRDD<Cells> cellsRDD); DataFrame createJavaSchemaRDD(ExtractorConfig<Cells> config); DataFrame sql(String query); static void saveRDD(RDD<T> rdd, S config, UpdateQueryBuilder queryBuilder); static void saveRDD(RDD<T> rdd, S config); SQLContext getSQLContext(); RDD textFile(ExtractorConfig<Cells> config); RDD<Cells> createHDFSRDD(ExtractorConfig<Cells> config); RDD<Cells> createS3RDD(ExtractorConfig<Cells> config);  }### Answer:
@Test public void createHDFSRDDTest() throws Exception { deepSparkContext = createDeepSparkContext(); DeepSparkContext deepSparkContextSpy = PowerMockito.spy(deepSparkContext); SQLContext sqlContext = mock(SQLContext.class); Whitebox.setInternalState(deepSparkContextSpy, "sc", sparkContext); Whitebox.setInternalState(deepSparkContextSpy, "sqlContext", sqlContext); RDD<String> rdd = mock(RDD.class); JavaRDD<String> javaRdd = mock(JavaRDD.class); when(deepSparkContextSpy.sc().textFile(anyString(), anyInt())).thenReturn(rdd); doReturn(javaRdd).when(deepSparkContextSpy).textFile(anyString()); when(rdd.toJavaRDD()).thenReturn(javaRdd); when(rdd.toJavaRDD().map(any(Function.class))).thenReturn(singleRdd); ExtractorConfig<Cells> config = createHDFSDeepJobConfig(); RDD rddReturn = deepSparkContextSpy.createHDFSRDD(config); verify(deepSparkContextSpy.sc(), times(1)).textFile(anyString(), anyInt()); verify(javaRdd, times(1)).map(any(Function.class)); }