import eu.dnetlib.data.mapreduce.hbase.lodExport.build.BlockReducer;
import eu.dnetlib.data.mapreduce.hbase.lodExport.build.SourceBuildMapper;
import eu.dnetlib.data.mapreduce.hbase.lodExport.build.TargetBuildMapper;
import eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing.DatasetReducer;
import eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing.SourceMapper;
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.*;
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 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() {
        SourceMapper sourceMapper = new SourceMapper();
        DatasetReducer datasetReducer = new DatasetReducer();
        mapDriver = MapDriver.newMapDriver(sourceMapper);
        reduceDriver = ReduceDriver.newReduceDriver(datasetReducer);
        mapReduceDriver = MapReduceDriver.newMapReduceDriver(sourceMapper, datasetReducer);
    }

    @Test
    public void testMapper() {
        Configuration 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\"]}");


        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"))
                .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://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.\""));
        reduceDriver
                .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.\""));
        reduceDriver.run();

    }

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

    }
}