import eu.dnetlib.data.mapreduce.hbase.lodExport.linkage.LinkMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;
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 LinkageTest {

    MapDriver<Text, Text, Text, Text> mapDriver;
    ReduceDriver<Text, Text, Text, Text> limesReducerDriver;
    ReduceDriver<Text, Text, Text, Text> linkCustomReducerDriver;
    Configuration configuration;

    @Before
    public void setUp() {
        configuration = new Configuration();
        configuration.set("lod.sourceMappings", "{\"type\":\"result\",\"fields\":[\"<http://www.eurocris.org/ontologies/cerif/1.3#name>\",\"<http://lod.openaire.eu/vocab/year>\",\"<http://purl.org/dc/terms/identifier>\"]}");
        configuration.set("lod.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.redisHost", "83.212.96.39");
        configuration.set("lod.redisPort", "6379");
        configuration.set("mapred.reduce.tasks", "1");
        configuration.set("lod.configXML",
                "<?xml version=\"1.0\" encoding=\"UTF-8\"?> <!DOCTYPE LIMES SYSTEM \"limes.dtd\"> <LIMES> <PREFIX> <NAMESPACE>http://www.w3.org/1999/02/22-rdf-syntax-ns#</NAMESPACE> <LABEL>rdf</LABEL> </PREFIX> <PREFIX> <NAMESPACE>http://www.w3.org/2000/01/rdf-schema#</NAMESPACE> <LABEL>rdfs</LABEL> </PREFIX> <SOURCE> <ID>source1</ID> <ENDPOINT>/user/kanakakis/openaireNT</ENDPOINT> <VAR>?x</VAR> <PAGESIZE>100</PAGESIZE> <RESTRICTION>?x rdf:type http://www.eurocris.org/ontologies/cerif/1.3#ResultEntity</RESTRICTION> <PROPERTY>http://lod.openaire.eu/vocab/year RENAME Year</PROPERTY> <PROPERTY>http://www.eurocris.org/ontologies/cerif/1.3#name AS lowercase->regexreplace(\"[^A-Za-z0-9]\",\" \") RENAME publicationName</PROPERTY> </SOURCE> <TARGET> <ID>source2</ID> <ENDPOINT>/user/kanakakis/dblpNT</ENDPOINT> <VAR>?y</VAR> <PAGESIZE>100</PAGESIZE> <RESTRICTION>?y rdf:type http://swrc.ontoware.org/ontology#Article</RESTRICTION> <PROPERTY>http://purl.org/dc/terms/issued RENAME Year</PROPERTY> <PROPERTY>http://www.w3.org/2000/01/rdf-schema#label AS lowercase->regexreplace(\"[^A-Za-z0-9]\",\" \") RENAME articleName</PROPERTY> </TARGET> <METRIC>AND(jaro(x.publicationName,y.articleName)|0.95,jaro(x.Year,y.Year)|1.0)</METRIC> <ACCEPTANCE> <THRESHOLD>0.95</THRESHOLD> <FILE>/user/kanakakis/accepted_links_0.95</FILE> <RELATION>owl:sameAs</RELATION> </ACCEPTANCE> <REVIEW> <THRESHOLD>0.95</THRESHOLD> <FILE>/user/kanakakis/verified_links_0.95</FILE> <RELATION>owl:sameAs</RELATION> </REVIEW> <EXECUTION>Default</EXECUTION> <OUTPUT>TTL</OUTPUT> </LIMES>");
        configuration.set("lod.limesDTD", "<?xml version=\"1.0\" encoding=\"utf-8\"?> <!ELEMENT LIMES (PREFIX*, SOURCE, TARGET, METRIC, ACCEPTANCE, REVIEW, EXECUTION*, GRANULARITY*, OUTPUT*)> <!ELEMENT PREFIX (NAMESPACE, LABEL)> <!ELEMENT NAMESPACE (#PCDATA)> <!ELEMENT LABEL (#PCDATA)> <!ELEMENT SOURCE (ID, ENDPOINT, GRAPH*, VAR, PAGESIZE, RESTRICTION+, PROPERTY+, TYPE*)> <!ELEMENT TARGET (ID, ENDPOINT, GRAPH*, VAR, PAGESIZE, RESTRICTION+, PROPERTY+, TYPE*)> <!ELEMENT ID (#PCDATA)> <!ELEMENT RESTRICTION (#PCDATA)> <!ELEMENT METRIC (#PCDATA)> <!ELEMENT ACCEPTANCE (THRESHOLD, FILE, RELATION)> <!ELEMENT REVIEW (THRESHOLD, FILE, RELATION)> <!ELEMENT RELATION (#PCDATA)> <!ELEMENT ENDPOINT (#PCDATA)> <!ELEMENT GRAPH (#PCDATA)> <!ELEMENT VAR (#PCDATA)> <!ELEMENT CLASS (#PCDATA)> <!ELEMENT PROPERTY (#PCDATA)> <!ELEMENT TYPE (#PCDATA)> <!ELEMENT THRESHOLD (#PCDATA)> <!ELEMENT FILE (#PCDATA)> <!ELEMENT PAGESIZE (#PCDATA)> <!ELEMENT EXECUTION (#PCDATA)> <!ELEMENT GRANULARITY (#PCDATA)> <!ELEMENT OUTPUT (#PCDATA)>");

        mapDriver = MapDriver.newMapDriver(new LinkMapper());

    }

    @Test
    public void testMapper() {
        mapDriver.withConfiguration(configuration)
                .withInput(new Text("0 equals"),
                        new Text("source_<http://lod.openaire.eu/data/result/od________18::7d2f9a16dec5143412d4bddf796d3c35>,target_<http://dblp.l3s.de/d2r/resource/publications/journals/corr/WangYP15>,"));
        mapDriver
                .withOutput(new Text("(0 equals"), new Text("source_<http://lod.openaire.eu/data/result/od________18::7d2f9a16dec5143412d4bddf796d3c35>,target_<http://dblp.l3s.de/d2r/resource/publications/journals/corr/WangYP15>,"));

        mapDriver.runTest();
    }

    @Test
    public void testReducer() throws IOException {
        List<Text> values = new ArrayList<>();
        values.add(new Text("source_<http://lod.openaire.eu/data/result/webcrawl____::478b3b9a8279a5ec96356c2c68bfab60>"));
        values.add(new Text("target_<http://dblp.l3s.de/d2r/resource/publications/journals/corr/WangYP15>"));

        limesReducerDriver.withConfiguration(configuration)
                .withInput(new Text("OA"), values)
                .run();

    }

    @Test
    public void Custom() throws IOException {
        List<Text> values = new ArrayList<>();
        values.add(new Text("source_<http://lod.openaire.eu/data/result/webcrawl____::478b3b9a8279a5ec96356c2c68bfab60>,target_<http://dblp.l3s.de/d2r/resource/publications/journals/corr/WangYP15>"));
        //    values.add(new Text("target_<http://dblp.l3s.de/d2r/resource/publications/journals/corr/WangYP15>"));

        linkCustomReducerDriver.withConfiguration(configuration)
                .withInput(new Text("OA"), values)
                .run();

    }


}