package de.tuberlin.dima.schubotz.wiki.mappers;

import de.tuberlin.dima.schubotz.wiki.WikiProgram;
import de.tuberlin.dima.schubotz.wiki.types.WikiQueryTuple;
import de.tuberlin.dima.schubotz.wiki.types.WikiTuple;
import eu.stratosphere.api.common.Plan;
import eu.stratosphere.api.java.DataSet;
import eu.stratosphere.api.java.ExecutionEnvironment;
import eu.stratosphere.api.java.functions.FlatMapFunction;
import eu.stratosphere.api.java.io.CsvReader;
import eu.stratosphere.api.java.io.TextInputFormat;
import eu.stratosphere.api.java.operators.DataSource;
import eu.stratosphere.api.java.typeutils.BasicTypeInfo;
import eu.stratosphere.client.LocalExecutor;
import eu.stratosphere.core.fs.FileSystem;
import eu.stratosphere.core.fs.Path;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;

import java.io.File;

import static org.junit.Assert.assertTrue;


/**
 * Used as template for test classes to compare actual and expected output.
 */
@Ignore
public abstract class WikiAbstractSubprocessTest {
    protected static final String STR_SPLIT = WikiProgram.STR_SPLIT;
    private static final String QUERY_SEPARATOR = WikiProgram.QUERY_SEPARATOR;
    private static final String WIKI_SEPARATOR = WikiProgram.WIKI_SEPARATOR;
    private static final String CSV_LINE_SEPARATOR = WikiProgram.CSV_LINE_SEPARATOR;
    private static final String CSV_FIELD_SEPARATOR = WikiProgram.CSV_FIELD_SEPARATOR;
    private static final Log LOG = LogFactory.getLog(WikiAbstractSubprocessTest.class);

    private final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    /**
     * Executes plan, compares test files. This method should be called by implementing classes.
     * @param outputSet dataset to output and compare to expected
     * @param expectedOutputFile expected file
     * @throws Exception execute(plan) can throw any exception
     */
    protected void testDataMap(DataSet<?> outputSet,
                               String expectedOutputFile) throws Exception {

        final File outputFile = File.createTempFile(this.getClass().getSimpleName(), "csv");
        outputFile.deleteOnExit();
        //File outputFile = new File("/home/jjl4/", "csv");

        outputSet.writeAsCsv(outputFile.getCanonicalPath(), CSV_LINE_SEPARATOR,
                CSV_FIELD_SEPARATOR, FileSystem.WriteMode.OVERWRITE);

        final Plan plan = env.createProgramPlan();
        LocalExecutor.execute(plan);

        final File expectedFile = new File(WikiAbstractSubprocessTest.class
                .getClassLoader().getResource(expectedOutputFile).getPath());
        assertTrue("Files could not be read!", expectedFile.canRead());
        assertTrue("Output does not match expected output!", FileUtils.contentEquals(outputFile, expectedFile));
    }

    protected DataSet<?> getCleanedData(String filename) throws Exception {
        String dir = null;
        try {
            dir = WikiAbstractSubprocessTest.class.getClassLoader().getResource(filename).getPath();
        } catch (NullPointerException e) {
            //Try again with absolute path
            dir = new File(filename).getPath();
        }
        final TextInputFormat format = new TextInputFormat(new Path(dir));
        if (dir.contains("expected")) { //Process as csv with tuples
            CsvReader reader = env.readCsvFile(dir);
            reader = reader.lineDelimiter(CSV_LINE_SEPARATOR);
            if (dir.contains("Query")) {
                return reader.tupleType(WikiQueryTuple.class);
            } else {
                return reader.tupleType(WikiTuple.class);
            }
        } else { //Process as normal data
            final FlatMapFunction<String, String> cleaner;
            if (dir.contains("Query")) {
                format.setDelimiter(QUERY_SEPARATOR);
                cleaner = new WikiQueryCleaner();
            } else { //WikiData
                format.setDelimiter(WIKI_SEPARATOR);
                cleaner = new WikiCleaner();
            }
            return new DataSource<String>(env,format, BasicTypeInfo.STRING_TYPE_INFO)
                    .flatMap(cleaner);
        }
    }
}

@@ -1034,16 +1034,81 @@
 output.%0A
+ * This file is dependent on the cleaners working. TODO REFACTOR%0A
  */%0A@Ign
@@ -1379,32 +1379,121 @@
 WIKI_SEPARATOR;%0A
+    //TODO REFACTOR THIS METHOD OF PASSING SEPARATORS to a getChar() method or setFormat%0A
     private stat
@@ -3640,32 +3640,107 @@
 INE_SEPARATOR);%0A
+            reader = reader.fieldDelimiter(CSV_FIELD_SEPARATOR.charAt(0));%0A
             if (

add refactoring comments