@Test
public void testDataSourceTask() {

	int keyCnt = 100;
	int valCnt = 20;
	
	this.outList = new ArrayList<Record>();
	
	try {
		InputFilePreparator.prepareInputFile(new UniformRecordGenerator(keyCnt, valCnt, false), 
			this.tempTestPath, true);
	} catch (IOException e1) {
		Assert.fail("Unable to set-up test input file");
	}
	
	super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
	super.addOutput(this.outList);
	
	DataSourceTask<Record> testTask = new DataSourceTask<Record>();
	
	super.registerFileInputTask(testTask, MockInputFormat.class, new File(tempTestPath).toURI().toString(), "\n");
	
	try {
		testTask.invoke();
	} catch (Exception e) {
		System.err.println(e);
		Assert.fail("Invoke method caused exception.");
	}
	
	Assert.assertTrue("Invalid output size. Expected: "+(keyCnt*valCnt)+" Actual: "+this.outList.size(),
		this.outList.size() == keyCnt * valCnt);
	
	HashMap<Integer,HashSet<Integer>> keyValueCountMap = new HashMap<Integer, HashSet<Integer>>(keyCnt);
	
	for (Record kvp : this.outList) {
		
		int key = kvp.getField(0, IntValue.class).getValue();
		int val = kvp.getField(1, IntValue.class).getValue();
		
		if(!keyValueCountMap.containsKey(key)) {
			keyValueCountMap.put(key,new HashSet<Integer>());
		}
		keyValueCountMap.get(key).add(val);
		
	}
	
	Assert.assertTrue("Invalid key count in out file. Expected: "+keyCnt+" Actual: "+keyValueCountMap.keySet().size(),
		keyValueCountMap.keySet().size() == keyCnt);
	
	for(Integer mapKey : keyValueCountMap.keySet()) {
		Assert.assertTrue("Invalid value count for key: "+mapKey+". Expected: "+valCnt+" Actual: "+keyValueCountMap.get(mapKey).size(),
			keyValueCountMap.get(mapKey).size() == valCnt);
	}
	
}