D-Net

package eu.dnetlib.data.mapreduce.hbase.broker;

import java.io.IOException;

import java.io.InputStream;

import java.util.List;

import java.util.Map;

import java.util.Set;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.transform.Column;

import eu.dnetlib.data.transform.Row;

import eu.dnetlib.data.transform.XsltRowTransformer;

import eu.dnetlib.data.transform.XsltRowTransformerFactory;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.mapreduce.Counter;

import org.apache.hadoop.mapreduce.Reducer.Context;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

import org.junit.Before;

import org.junit.Ignore;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.mockito.Mock;

import org.mockito.junit.MockitoJUnitRunner;

import static org.junit.Assert.*;

import static org.mockito.ArgumentMatchers.anyString;

import static org.mockito.Mockito.when;

/**

 * Created by claudio on 22/07/16.

 */

@RunWith(MockitoJUnitRunner.Silent.class)

public class EventWrapperTest {

	private static final Log log = LogFactory.getLog(EventWrapperTest.class);

	private Oaf oaf1_1;

	private Oaf oaf1_2;

	private Oaf oaf2_1;

	private Oaf oaf2_2;

	private Oaf oaf3_1;

	private Oaf oaf3_2;

	private String xslt;

	@Mock

	private Context context;

	@Mock

	private Counter counter;

	private Map<String, String> baseUrlMap = Maps.newHashMap();

	private static final String basePathProfiles = "/eu/dnetlib/test/profiles/TransformationRuleDSResources/TransformationRuleDSResourceType/2hbase/";

	@Before

	public void setUp() throws Exception {

		xslt = loadFromTransformationProfile("oaf2hbase.xml");

		oaf1_1 = asOaf("recordOaf_1_1.xml");

		oaf1_2 = asOaf("recordOaf_1_2.xml");

		oaf2_1 = asOaf("recordOaf_2_1.xml");

		oaf2_2 = asOaf("recordOaf_2_2.xml");

		oaf3_1 = asOaf("recordOaf_3_1.xml");

		oaf3_2 = asOaf("recordOaf_3_2.xml");

		when(context.getCounter(anyString(), anyString())).thenReturn(counter);

		baseUrlMap.put("publication", "https://explore.openaire.eu/search/publication?articleId=%s");

		baseUrlMap.put("dataset", "https://explore.openaire.eu/search/dataset?datasetId=%s");

		baseUrlMap.put("software", "https://explore.openaire.eu/search/software?softwareId=%s");

		baseUrlMap.put("other", "https://explore.openaire.eu/search/other?orpId=%s");

	}

	@Test

	public void testAbstractEvent() throws Exception {

		final String id1 = AbstractEventFactory.process(oaf1_1, oaf1_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id2 = AbstractEventFactory.process(oaf2_1, oaf2_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id3 = AbstractEventFactory.process(oaf3_1, oaf3_2, 1.0f).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

	}

	@Test

	public void testOAVersionEvent() throws Exception {

		final Set<String> untrustedOaDs = Sets.newHashSet();

		final String id1 = OAVersionEventFactory.process(oaf1_1, oaf1_2, 1.0f, untrustedOaDs).get(0).asBrokerEvent().getEventId();

		final String id2 = OAVersionEventFactory.process(oaf2_1, oaf2_2, 1.0f, untrustedOaDs).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

	}

	@Test

	public void testPIDEvent() throws Exception {

		final String id1 = PIDEventFactory.process(oaf1_1, oaf1_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id2 = PIDEventFactory.process(oaf2_1, oaf2_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id3 = PIDEventFactory.process(oaf3_1, oaf3_2, 1.0f).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

	}

	@Test

	//@Ignore

	public void testProjectEvent() throws Exception {

		Oaf oaf1_1_c = enrichWithCachedOafRel(oaf1_1, "oafRel_isProducedBy_1.json", "project1.json");

		Oaf oaf1_2_c = enrichWithCachedOafRel(oaf1_2, "oafRel_isProducedBy_2.json", "project2.json");

		final String id1 = ProjectEventFactory.process(context, oaf1_1_c, oaf1_2_c, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		/*final String id2 = ProjectEventFactory.process(context, oaf2_1, oaf2_2, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		final String id3 = ProjectEventFactory.process(context, oaf3_1, oaf3_2, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

		*/

	}

	private Oaf enrichWithCachedOafRel(final Oaf oaf, final String oafRel, final String oafProject) throws IOException {

		final Oaf.Builder cachedRel = Oaf.newBuilder();

		JsonFormat.merge(IOUtils.toString(getClass().getResourceAsStream(oafRel)), cachedRel);

		final Oaf.Builder cachedTarget = Oaf.newBuilder();

		JsonFormat.merge(IOUtils.toString(getClass().getResourceAsStream(oafProject)), cachedTarget);

		cachedRel.getRelBuilder().setCachedOafTarget(cachedTarget);

		Oaf.Builder builder = Oaf.newBuilder(oaf);

		builder.getEntityBuilder().addCachedOafRel(cachedRel);

		return builder.build();

	}

	@Test

	public void testPublicationDateEvent() throws Exception {

		final String id1 = PublicationDateEventFactory.process(oaf1_1, oaf1_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id2 = PublicationDateEventFactory.process(oaf2_1, oaf2_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id3 = PublicationDateEventFactory.process(oaf3_1, oaf3_2, 1.0f).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

	}

	@Test

	@Ignore

	public void testSoftwareEventFactory() throws Exception {

		final String id1 = SoftwareEventFactory.process(context, oaf1_1, oaf1_2, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		final String id2 = SoftwareEventFactory.process(context, oaf2_1, oaf2_2, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		final String id3 = SoftwareEventFactory.process(context, oaf3_1, oaf3_2, 1.0f, baseUrlMap).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

	}

	@Test

	public void testSubjectEventFactory() throws Exception {

		final String id1 = SubjectEventFactory.process(context, oaf1_1, oaf1_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id2 = SubjectEventFactory.process(context, oaf2_1, oaf2_2, 1.0f).get(0).asBrokerEvent().getEventId();

		final String id3 = SubjectEventFactory.process(context, oaf3_1, oaf3_2, 1.0f).get(0).asBrokerEvent().getEventId();

		verifyEquals(id1, id2);

		verifyDifferent(id1, id3);

	}

	private void verifyEquals(final String x, final String y) {

		assertNotNull(x);

		assertNotNull(y);

		assertTrue(x.length() > 10);

		assertTrue(y.length() > 10);

		assertEquals(x, y);

		log.info(String.format("%s = %s", x, y));

	}

	private void verifyDifferent(final String x, final String y) {

		log.info(String.format("%s != %s", x, y));

		assertNotNull(x);

		assertNotNull(y);

		assertTrue(x.length() > 10);

		assertTrue(y.length() > 10);

		assertFalse(x.equals(y));

	}

	private Oaf asOaf(final String xmlFile) throws Exception {

		for (final Row row : asRows(getClass().getResourceAsStream(xmlFile))) {

			for (final Column<String, byte[]> c : row.getColumns()) {

				final String cf = row.getColumnFamily();

				if (cf.equals("result") && c.getName().equals("body")) { return Oaf.parseFrom(c.getValue()); }

			}

		}

		throw new RuntimeException("Oaf not found");

	}

	private String loadFromTransformationProfile(final String profilePath) {

		log.info("Loading xslt from: " + basePathProfiles + profilePath);

		final InputStream profile = getClass().getResourceAsStream(basePathProfiles + profilePath);

		final SAXReader saxReader = new SAXReader();

		try {

			final Document doc = saxReader.read(profile);

			return doc.selectSingleNode("//SCRIPT/CODE/*[local-name()='stylesheet']").asXML();

		} catch (final DocumentException e) {

			e.printStackTrace();

			throw new RuntimeException(e);

		}

	}

	private List<Row> asRows(final InputStream recordStream)

			throws Exception {

		final XsltRowTransformer transformer = new XsltRowTransformerFactory().getTransformer(xslt);

		assertNotNull(transformer);

		final String record = IOUtils.toString(recordStream);

		final List<Row> rows = transformer.apply(record);

		assertNotNull(rows);

		assertFalse(rows.isEmpty());

		return rows;

	}

}