added test class for the selection of valid relations and description

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/complete/SparkSelectValidRelationsJob.java

@@ -18,6 +18,14 @@ import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.oa.graph.dump.Utils;
 import eu.dnetlib.dhp.schema.oaf.*;
 
+/**
+ * It selects the valid relations among those present in the graph. One relation is valid if it is not deletedbyinference
+ * and if both the source and the target node are present in the graph and are not deleted by inference nor invisible.
+ * To check this I made a view of the ids of all the entities in the graph, and select the relations for which a join exists
+ * with this view for both the source and the target
+ */
+
+
 public class SparkSelectValidRelationsJob implements Serializable {
 
     private static final Logger log = LoggerFactory.getLogger(SparkSelectValidRelationsJob.class);

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/dump/complete/SelectRelationTest.java

@@ -0,0 +1,99 @@
+package eu.dnetlib.dhp.oa.graph.dump.complete;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.ForeachFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.schema.oaf.Relation;
+import net.sf.saxon.expr.instruct.ForEach;
+
+public class SelectRelationTest {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private static SparkSession spark;
+
+    private static Path workingDir;
+
+    private static final Logger log = LoggerFactory
+            .getLogger(SelectRelationTest.class);
+
+    private static HashMap<String, String> map = new HashMap<>();
+
+    @BeforeAll
+    public static void beforeAll() throws IOException {
+        workingDir = Files
+                .createTempDirectory(SelectRelationTest.class.getSimpleName());
+        log.info("using work dir {}", workingDir);
+
+        SparkConf conf = new SparkConf();
+        conf.setAppName(SelectRelationTest.class.getSimpleName());
+
+        conf.setMaster("local[*]");
+        conf.set("spark.driver.host", "localhost");
+        conf.set("hive.metastore.local", "true");
+        conf.set("spark.ui.enabled", "false");
+        conf.set("spark.sql.warehouse.dir", workingDir.toString());
+        conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+        spark = SparkSession
+                .builder()
+                .appName(SelectRelationTest.class.getSimpleName())
+                .config(conf)
+                .getOrCreate();
+    }
+
+    @AfterAll
+    public static void afterAll() throws IOException {
+        FileUtils.deleteDirectory(workingDir.toFile());
+        spark.stop();
+    }
+
+    @Test
+    public void test1() throws Exception {
+
+        final String sourcePath = getClass()
+                .getResource("/eu/dnetlib/dhp/oa/graph/dump/selectrelations")
+                .getPath();
+
+        SparkSelectValidRelationsJob.main(new String[] {
+                "-isSparkSessionManaged", Boolean.FALSE.toString(),
+                "-outputPath", workingDir.toString() + "/relation",
+                "-sourcePath", sourcePath
+        });
+
+//		dumpCommunityProducts.exec(MOCK_IS_LOOK_UP_URL,Boolean.FALSE, workingDir.toString()+"/dataset",sourcePath,"eu.dnetlib.dhp.schema.oaf.Dataset","eu.dnetlib.dhp.schema.dump.oaf.Dataset");
+
+        final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+
+        JavaRDD<eu.dnetlib.dhp.schema.oaf.Relation> tmp = sc
+                .textFile(workingDir.toString() + "/relation")
+                .map(item -> OBJECT_MAPPER.readValue(item, eu.dnetlib.dhp.schema.oaf.Relation.class));
+
+        org.apache.spark.sql.Dataset<Relation> verificationDataset = spark
+                .createDataset(tmp.rdd(), Encoders.bean(eu.dnetlib.dhp.schema.oaf.Relation.class));
+
+        Assertions.assertTrue(verificationDataset.count() == 7);
+
+    }
+
+}