	@Test
	public void testFullFormatProfilerJob() throws Exception {

		// prepare for test
		//createTextInputFile();

		log.info("Checking input file is present...");
		// Check that the input file is present:
		Path[] inputFiles = FileUtil.stat2Paths(getFileSystem().listStatus(
				input, new OutputLogFilter()));
		Assert.assertEquals(1, inputFiles.length);
		
		// Set up arguments for the job:
		// FIXME The input file could be written by this test.
		String[] args = {"src/test/resources/test-inputs.txt", this.output.getName()};
		
		// Set up the config and tool
		@SuppressWarnings("unused")
		Config config = ConfigFactory.load();
		FormatProfiler wir = new FormatProfiler();

		// run job
		log.info("Setting up job config...");
		JobConf conf = this.mrCluster.createJobConf();
		wir.createJobConf(conf, args);
		// Disable speculative execution for tests:
		conf.set( "mapred.reduce.tasks.speculative.execution", "false" );
		log.info("Running job...");
		JobClient.runJob(conf);
		log.info("Job finished, checking the results...");

		// check the output
		Path[] outputFiles = FileUtil.stat2Paths( getFileSystem().listStatus( output ) );
		//Assert.assertEquals( config.getInt( "warc.hadoop.num_reducers" ) + 1, outputFiles.length );
		
		// Check contents of the output:
		for( Path output : outputFiles ) {
			log.info(" --- output : "+output);
			if( getFileSystem().isFile(output) ) {
				InputStream is = getFileSystem().open(output);
				BufferedReader reader = new BufferedReader(new InputStreamReader(is));
				String line = null;
				while( ( line = reader.readLine()) != null ) {
					log.info(line);
				}
				reader.close();
			} else {
				log.info(" --- ...skipping directory...");
			}
		}
	}