package eu.stratosphere.test.spargel;

import eu.stratosphere.api.common.Plan;
import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.spargel.java.examples.connectedcomponents.SpargelConnectedComponents;
import eu.stratosphere.test.iterative.nephele.ConnectedComponentsNepheleITCase;
import eu.stratosphere.test.util.TestBase2;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.io.BufferedReader;
import java.util.Collection;

@RunWith(Parameterized.class)
public class SpargelConnectedComponentsITCase extends TestBase2 {

	private static final long SEED = 0xBADC0FFEEBEEFL;

	private static final int NUM_VERTICES = 1000;

	private static final int NUM_EDGES = 10000;

	protected String verticesPath;

	protected String edgesPath;

	protected String resultPath;

	public SpargelConnectedComponentsITCase(Configuration config) {
		super(config);
	}

	@Override
	protected void preSubmit() throws Exception {
		verticesPath = createTempFile("vertices.txt", ConnectedComponentsNepheleITCase.getEnumeratingVertices(NUM_VERTICES));
		edgesPath = createTempFile("edges.txt", ConnectedComponentsNepheleITCase.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
		resultPath = getTempFilePath("results");
	}

	@Override
	protected Plan getTestJob() {
		int dop = config.getInteger("ConnectedComponents#NumSubtasks", 1);
		int maxIterations = config.getInteger("ConnectedComponents#NumIterations", 1);
		String[] params = { String.valueOf(dop) , verticesPath, edgesPath, resultPath, String.valueOf(maxIterations) };

		SpargelConnectedComponents cc = new SpargelConnectedComponents();
		return cc.getPlan(params);
	}

	@Override
	protected void postSubmit() throws Exception {
		for (BufferedReader reader : getResultReader(resultPath)) {
			ConnectedComponentsNepheleITCase.checkOddEvenResult(reader);
		}
	}

	@Parameterized.Parameters
	public static Collection<Object[]> getConfigurations() {
		Configuration config1 = new Configuration();
		config1.setInteger("ConnectedComponents#NumSubtasks", 4);
		config1.setInteger("ConnectedComponents#NumIterations", 100);
		return toParameterList(config1);
	}
}
@@ -259,16 +259,8 @@
 ive.
-nephele.
 Conn
@@ -278,23 +278,16 @@
 ents
-Nephele
 ITCase;%0A
 impo
@@ -286,1855 +286,579 @@
 se;%0A
-import eu.stratosphere.test.util.TestBase2;%0Aimport org.junit.runner.RunWith;%0Aimport org.junit.runners.Parameterized;%0A%0Aimport java.io.BufferedReader;%0Aimport java.util.Collection;%0A%0A@RunWith(Parameterized.class)%0Apublic class SpargelConnectedComponentsITCase extends TestBase2 %7B%0A%0A%09private static final long SEED = 0xBADC0FFEEBEEFL;%0A%0A%09private static final int NUM_VERTICES = 1000;%0A%0A%09private static final int NUM_EDGES = 10000;%0A%0A%09protected String verticesPath;%0A%0A%09protected String edgesPath;%0A%0A%09protected String resultPath;%0A%0A%09public SpargelConnectedComponentsITCase(Configuration config) %7B%0A%09%09super(config);%0A%09%7D%0A%0A%09@Override%0A%09protected void preSubmit() throws Exception %7B%0A%09%09verticesPath = createTempFile(%22vertices.txt%22, ConnectedComponentsNepheleITCase.getEnumeratingVertices(NUM_VERTICES));%0A%09%09edgesPath = createTempFile(%22edges.txt%22, ConnectedComponentsNepheleITCase.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));%0A%09%09resultPath = getTempFilePath(%22results%22);%0A%09%7D%0A%0A%09@Override%0A%09protected Plan getTestJob() %7B%0A%09%09int dop = config.getInteger(%22ConnectedComponents#NumSubtasks%22, 1);%0A%09%09int maxIterations = config.getInteger(%22ConnectedComponents#NumIterations%22, 1);%0A%09%09String%5B%5D params = %7B String.valueOf(dop) , verticesPath, edgesPath, resultPath, String.valueOf(maxIterations) %7D;%0A%0A%09%09SpargelConnectedComponents cc = new SpargelConnectedComponents();%0A%09%09return cc.getPlan(params);%0A%09%7D%0A%0A%09@Override%0A%09protected void postSubmit() throws Exception %7B%0A%09%09for (BufferedReader reader : getResultReader(resultPath)) %7B%0A%09%09%09ConnectedComponentsNepheleITCase.checkOddEvenResult(reader);%0A%09%09%7D%0A%09%7D%0A%0A%09@Parameterized.Parameters%0A%09public static Collection%3CObject%5B%5D%3E getConfigurations() %7B%0A%09%09Configuration config1 = new Configuration();%0A%09%09config1.setInteger(%22ConnectedComponents#NumSubtasks%22, 4);%0A%09%09config1.setInteger(%22ConnectedComponents#NumIterations%22, 100);%0A%09%09return toParameterList(config1
+%0Apublic class SpargelConnectedComponentsITCase extends ConnectedComponentsITCase %7B%0A%0A%09public SpargelConnectedComponentsITCase(Configuration config) %7B%0A%09%09super(config);%0A%09%7D%0A%0A%09@Override%0A%09protected Plan getTestJob() %7B%0A%09%09int dop = config.getInteger(%22ConnectedComponents#NumSubtasks%22, 1);%0A%09%09int maxIterations = config.getInteger(%22ConnectedComponents#NumIterations%22, 1);%0A%09%09String%5B%5D params = %7B String.valueOf(dop) , verticesPath, edgesPath, resultPath, String.valueOf(maxIterations) %7D;%0A%0A%09%09SpargelConnectedComponents cc = new SpargelConnectedComponents();%0A%09%09return cc.getPlan(params
 );%0A%09

Extend ConnectedComponentsITCase instead of code duplication for SpargelConnectedComponentsITCase