import eu.dnetlib.data.mapreduce.hbase.lodExport.linkage.LimesReducer;
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.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;

/**
 * Created by eri_k on 8/24/2016.
 */
public class LinkageTest {

    MapDriver<Text, Text, Text, Text> mapDriver;
    ReduceDriver<Text, Text, Text, Text> reduceDriver;
    MapReduceDriver<Text, Text, Text, Text, Text, Text> mapReduceDriver;
    Configuration configuration;

    @Before
    public void setUp() {
        configuration = new Configuration();
        configuration.set("lod.sourceMappings", "{\"result\":[\"http://www.w3.org/1999/02/22-rdf-syntax-ns#type\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/dateOfTransformation\",\"http://lod.openaire.eu/vocab/dateOfCollection\",\"http://purl.org/dc/terms/identifier\",\"http://www.eurocris.org/ontologies/cerif/1.3#name\",\"http://purl.org/dc/terms/dateAccepted\",\"http://purl.org/dc/terms/publisher\",\"http://purl.org/dc/terms/identifier\",\"http://purl.org/dc/terms/language\",\"http://purl.org/dc/terms/date\",\"http://lod.openaire.eu/vocab/resultSubject\",\"http://lod.openaire.eu/vocab/externalReference\",\"http://purl.org/dc/terms/source\",\"http://purl.org/dc/terms/format\",\"http://lod.openaire.eu/vocab/context\",\"http://dbpedia.org/ontology/country\",\"http://purl.org/dc/terms/accessRights\",\"http://purl.org/dc/terms/description\",\"http://lsdis.cs.uga.edu/projects/semdis/opus#journal_name\",\"http://lod.openaire.eu/vocab/dataSourceType\",\"http://lod.openaire.eu/vocab/device\",\"http://lod.openaire.eu/vocab/size\",\"http://lod.openaire.eu/vocab/version\",\"http://lod.openaire.eu/vocab/lastMetadataUpdate\",\"http://lod.openaire.eu/vocab/metadataVersion\",\"http://lod.openaire.eu/vocab/year\",\"http://lod.openaire.eu/vocab/resultType\"],\"project\": [\"http://www.w3.org/1999/02/22-rdf-syntax-ns#type\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/dateOfTransformation\",\"http://lod.openaire.eu/vocab/dateOfCollection\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/projectCode\",\"http://schema.org/url\",\"http://www.eurocris.org/ontologies/cerif/1.3#acronym\",\"http://www.eurocris.org/ontologies/cerif/1.3#name\",\"http://www.eurocris.org/ontologies/cerif/1.3#startDate\",\"http://www.eurocris.org/ontologies/cerif/1.3#endDate\",\"http://purl.org/cerif/frapo/hasCallIdentifier\",\"http://www.eurocris.org/ontologies/cerif/1.3#keyword\",\"http://www.w3.org/2006/time#hasDurationDescription\",\"http://lod.openaire.eu/vocab/ec_SC39\",\"http://lod.openaire.eu/vocab/contractType\",\"http://lod.openaire.eu/vocab/oaMandatePublications\",\"http://lod.openaire.eu/vocab/projectSubjects\",\"http://od.openaire.eu/vocab/ec_article29-3\",\"http://lod.openaire.eu/vocab/funder\",\"http://lod.openaire.eu/vocab/fundingLevel0\",\"http://lod.openaire.eu/vocab/fundingLevel1\",\"http://lod.openaire.eu/vocab/fundingLevel2\",\"http://lod.openaire.eu/vocab/fundingLevel3\"],\"person\": [\"http://www.w3.org/1999/02/22-rdf-syntax-ns#type\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/dateOfTransformation\",\"http://lod.openaire.eu/vocab/dateOfCollection\",\"http://purl.org/dc/terms/identifier\", \"http://xmlns.com/foaf/0.1/firstName\",\"http://xmlns.com/foaf/0.1/lastName\", \"http://xmlns.com/foaf/0.1/name\",\"http://schema.org/faxNumber\",\"http://xmlns.com/foaf/0.1/mbox\",\"http://xmlns.com/foaf/0.1/phone\", \"http://schema.org/nationality\",\"http://purl.org/dc/terms/identifier\", \"http://lod.openaire.eu/vocab/trust\"],\"organization\": [\"http://www.w3.org/1999/02/22-rdf-syntax-ns#type\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/dateOfTransformation\",\"http://lod.openaire.eu/vocab/dateOfCollection\",\"http://purl.org/dc/terms/identifier\",\"http://www.w3.org/2004/02/skos/core#altLabel\",\"http://www.w3.org/2004/02/skos/core#prefLabel\",\"http://lod.openaire.eu/vocab/webSiteUrl\",\"http://xmlns.com/foaf/0.1/logo\",\"http://dbpedia.org/ontology/country\",\"http://lod.openaire.eu/vocab/entityType\" ],\"datasource\":[\"http://www.w3.org/1999/02/22-rdf-syntax-ns#type\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/dateOfTransformation\",\"http://lod.openaire.eu/vocab/dateOfCollection\",\"http://purl.org/dc/terms/identifier\",\"http://lod.openaire.eu/vocab/datasourceType\",\"http://lod.openaire.eu/vocab/openAIRECompatibility\",\"http://dbpedia.org/ontology/officialName\",\"http://lod.openaire.eu/vocab/englishName\",\"http://schema.org/url\",\"http://xmlns.com/foaf/0.1/logo\",\"http://xmlns.com/foaf/0.1/mbox\",\"http://purl.org/vocab/vann/preferredNamespacePrefix\",\"http://www.w3.org/2003/01/geo/wgs84_pos#lat\",\"http://www.w3.org/2003/01/geo/wgs84_pos#long\",\"http://lod.openaire.eu/vocab/dateOfValidity\",\"http://purl.org/dc/terms/description\",\"http://lod.openaire.eu/vocab/subjectList\",\"http://lod.openaire.eu/numberOfItems\",\"http://purl.org/dc/terms/date\",\"http://lod.openaire.eu/vocab/policies\",\"http://lod.openaire.eu/vocab/languages\",\"http://lod.openaire.eu/vocab/contentType\",\"http://lod.openaire.eu/vocab/accessInfoPackage\",\"http://lod.openaire.eu/vocab/releaseStartDate\",\"http://lod.openaire.eu/vocab/releaseEndDate\",\"http://lod.openaire.eu/vocab/missionStatementUrl\",\"http://www.europeana.eu/schemas/edm/dataProvider\",\"http://lod.openaire.eu/vocab/serviceProvider\",\"http://lod.openaire.eu/vocab/databaseAccessType\",\"http://lod.openaire.eu/vocab/dataUploadType\",\"http://lod.openaire.eu/vocab/dataUploadRestrictions\",\"http://lod.openaire.eu/vocab/versioning\",\"http://lod.openaire.eu/vocab/citationGuidelineUrl\",\"http://lod.openaire.eu/vocab/qualityManagementKind\",\"http://lod.openaire.eu/vocab/pidSystems\",\"http://lod.openaire.eu/vocab/certificates\",\"http://purl.org/dc/terms/accessRights\"]}");
        configuration.set("lod.redisHost", "83.212.110.197");
        configuration.set("lod.redisPort", "13437");
        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)>");

        LinkMapper mapper = new LinkMapper();
        LimesReducer reducer = new LimesReducer();
        mapDriver = MapDriver.newMapDriver(mapper);
        reduceDriver = ReduceDriver.newReduceDriver(reducer);
        mapReduceDriver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
    }

    @Test
    public void testMapper() {


        mapDriver.withConfiguration(configuration)
                .withInput(new Text("1"),
                        new Text("source_<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>,target_<http://dblp.l3s.de/d2r/resource/publications/journals/advai/Luis-GarciaP16>,\n"))
                .withInput(new Text("com"), new Text("target_<http://dblp.l3s.de/d2r/resource/publications/journals/advai/Luis-GarciaP16>,"));

        mapDriver
                .withOutput(new Text("1"), new Text("source_<http://lod.openaire.eu/data/result/od_______908::bbaecb13949279cda128a66545446b76>," +
                        "target_<http://dblp.l3s.de/d2r/resource/publications/journals/advai/Luis-GarciaP16>,\n"))
                .withOutput(new Text("com"), new Text("target_<http://dblp.l3s.de/d2r/resource/publications/journals/advai/Luis-GarciaP16>,"));
        mapDriver.runTest();
    }

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


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

    }

    /* "target_<http://dblp.l3s.de/d2r/resource/publications/journals/bmcbi/NewmanC07>"
    "id\t<http://dblp.l3s.de/d2r/resource/publications/journals/bmcbi/NewmanC07>,<http://www.w3.org/2000/01/rdf-schema#label>\t\"The CP properties of the lightest Higgs boson with sbottom effects\",<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>\t<http://swrc.ontoware.org/ontology#Article>,<http://purl.org/dc/terms/identifier>\t\"DOI 10.1186%2F1471-2105-8-382\",<http://purl.org/dc/terms/issued>\t\"2002\","

    source_<http://lod.openaire.eu/data/result/od________18::dc4f16056a8df260a2e7475d5faec554>
            "id\t<http://lod.openaire.eu/data/result/od________18::dc4f16056a8df260a2e7475d5faec554>,<http://lod.openaire.eu/vocab/year>\t\"2002\",<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>\t<http://www.eurocris.org/ontologies/cerif/1.3#ResultEntity>,<http://purl.org/dc/terms/identifier>\t\"10.1088/0954-3899/28/9/304\",<http://www.eurocris.org/ontologies/cerif/1.3#name>\t\"The CP properties of the lightest Higgs boson with sbottom effects\","
*/


    @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();*/

    }
}