import eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing.DatasetReducer;
import eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing.SourceMapper;
import eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing.TargetMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;
import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


public class PreprocessingTest {

    MapDriver<LongWritable, Text, Text, Text> mapDriver;
    ReduceDriver<Text, Text, Text, Text> reduceDriver;
    MapReduceDriver<LongWritable, Text, Text, Text, Text, Text> mapReduceDriver;

    @Before
    public void setUp() {
        DatasetReducer datasetReducer = new DatasetReducer();
        reduceDriver = ReduceDriver.newReduceDriver(datasetReducer);
       }

    @Test
    public void testMapper() throws IOException {
        Configuration configuration = new Configuration();
        String sourceMappings = "{\"type\":\"result\", \"fields\":[\"<http://purl.org/dc/terms/identifier>\",\"<http://www.eurocris.org/ontologies/cerif/1.3#name>\",\"<http://lod.openaire.eu/vocab/year>\"]}";
        configuration.set("lod.sourceMappings", sourceMappings);
        mapDriver = MapDriver.newMapDriver(new SourceMapper());
        mapDriver.withConfiguration(configuration);
        mapDriver.withInput(new LongWritable(1),  new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://www.eurocris.org/ontologies/cerif/1.3#ResultEntity> .\n"))
                .withInput(new LongWritable(2),
                        new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76> <http://www.eurocris.org/ontologies/cerif/1.3#name> \"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\" .\n"));
        mapDriver.withOutput(new Text("OA,result,<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>,"), new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>" +
                "	<http://www.eurocris.org/ontologies/cerif/1.3#name>	\"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\""));
        mapDriver.runTest(true);


    }



    @Test
    public void testTargetMapper() throws IOException {
        Configuration configuration = new Configuration();
        String targetMappings = "{\"type\":\"publications\",\"fields\":[\"<http://purl.org/dc/terms/issued>\",\"<http://www.w3.org/2000/01/rdf-schema#label>\",\"<http://purl.org/dc/terms/identifier>\"]}";
        configuration.set("lod.targetMappings", targetMappings);

        mapDriver = MapDriver.newMapDriver(new TargetMapper());
        mapDriver.withConfiguration(configuration);
        mapDriver.withInput(new LongWritable(1),  new Text("<http://xmlns.com/foaf/0.1/Document> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://www.w3.org/2000/01/rdf-schema#Class> .\n"));
     //   mapDriver.withOutput(new Text("TARGET,publications,<http://dblp.l3s.de/d2r/resource/journals/publications>"),new Text("\"4OR\""));
        mapDriver.runTest(true);

    }




    @Test
public void testReducer ()throws Exception {
    Configuration configuration = new Configuration();
    String sourceMappings = "{\"type\":\"result\", \"fields\":[\"<http://purl.org/dc/terms/identifier>\",\"<http://www.eurocris.org/ontologies/cerif/1.3#name>\",\"<http://lod.openaire.eu/vocab/year>\"]}";
    configuration.set("lod.sourceMappings", sourceMappings);

    List inputList = new ArrayList();
    inputList.add(new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>" +
            "	<http://www.eurocris.org/ontologies/cerif/1.3#name>	\"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\""));

    reduceDriver.withConfiguration(configuration);
    reduceDriver.withInput(new Text("OA,result,<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>,"), inputList);

    reduceDriver.run();
}
@Test
    public void genericTest(){
    String fieldValue="lalalal^^Sssss";
    fieldValue=fieldValue.substring(0,fieldValue.indexOf("^"));
    System.out.println(fieldValue);
    }

}

        /*     mapDriver.withConfiguration(configuration)
                .withInput(new LongWritable(1),
                        new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://www.eurocris.org/ontologies/cerif/1.3#ResultEntity> .\n"))

*//*


     */
/*   mapDriver .withOutput(new Text("OA,result,<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>"),
                new Text("<http://www.eurocris.org/ontologies/cerif/1.3#name>,\"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\","));
*//*

        mapDriver.runTest();
    }

    @Test
    public void testReducer() throws IOException {
        List<Text> values = new ArrayList<>();
        values.add(new Text("<http://www.eurocris.org/ontologies/cerif/1.3#name>,\"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\""));
        limesReducerDriver
                .withInput(new Text("OA,result,<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>"), values)
                .withOutput(new Text("<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>"), new Text("<http://www.eurocris.org/ontologies/cerif/1.3#name>,\"Dietary fish oil  MaxEPA  enhances pancreatic carcinogenesis in azaserine treated rats.\""));
        limesReducerDriver.run();

    }

    private static List<Pair<LongWritable, Text>> getInput() throws FileNotFoundException {
        File file = new File("C:\\Users\\eri_k\\workspace\\lodinterlinking\\branches\\cacheOptimized\\src\\main\\resources\\eu\\dnetlib\\data\\mapreduce\\hbase\\lodExport\\source.nt");
        Scanner scanner = new Scanner(new FileInputStream(file));
        List<Pair<LongWritable, Text>> input = new ArrayList<Pair<LongWritable, Text>>();
        while (scanner.hasNext()) {
            String line = scanner.nextLine();
            input.add(new Pair<LongWritable, Text>(new LongWritable(0), new Text(line)));
        }
        return input;
    }

    @Test
    public void testMapReduce() {
        */
/*mapReduceDriver.withInput(new LongWritable(), new Text("655209;1;796764372490213;804422938115889;6"));
        mapReduceDriver.withOutput(new Text("6"), new IntWritable(2));
        mapReduceDriver.runTest();

    }
}
*/