adding testing classes

dump/src/main/java/eu/dnetlib/dhp/oa/graph/dump/eosc/SparkDumpOrganizationProject.java

@@ -40,7 +40,7 @@ public class SparkDumpOrganizationProject implements Serializable {
 			.toString(
 				SparkDumpOrganizationProject.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/oa/graph/dump/input_relationdump_parameters.json"));
+						"/eu/dnetlib/dhp/oa/graph/dump/eosc_organizationprojectrelations_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 		parser.parseArgument(args);

dump/src/main/java/eu/dnetlib/dhp/oa/graph/dump/eosc/SparkUpdateProjectInfo.java

@@ -13,11 +13,9 @@ import java.util.stream.Collectors;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.ForeachFunction;
 import org.apache.spark.api.java.function.MapFunction;
-import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.Encoders;
-import org.apache.spark.sql.SaveMode;
-import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.*;
 import org.dom4j.Document;
 import org.dom4j.DocumentException;
 import org.dom4j.Node;
@@ -96,7 +94,6 @@ public class SparkUpdateProjectInfo implements Serializable {
 		String resultType) {
 
 		Dataset<Result> result = Utils.readPath(spark, workingPath + resultType + "extendedaffiliation", Result.class);
-
 		Dataset<ResultProject> resultProject = Utils.readPath(spark, preparedInfoPath, ResultProject.class);
 		result
 			.joinWith(

dump/src/test/java/eu/dnetlib/dhp/oa/graph/dump/eosc/ExtendProjectTest.java

@@ -0,0 +1,218 @@
+
+package eu.dnetlib.dhp.oa.graph.dump.eosc;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import eu.dnetlib.dhp.eosc.model.*;
+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.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 java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * @author miriam.baglioni
+ * @Date 25/10/23
+ */
+public class ExtendProjectTest {
+
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static SparkSession spark;
+
+	private static Path workingDir;
+
+	private static final Logger log = LoggerFactory
+		.getLogger(ExtendProjectTest.class);
+
+	private static HashMap<String, String> map = new HashMap<>();
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(ExtendProjectTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(ExtendProjectTest.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(ExtendProjectTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void ExtendEoscResultWithProjectTest() throws Exception {
+		final String sourcePath = getClass()
+			.getResource("/eu/dnetlib/dhp/oa/graph/dump/eosc/input")
+			.getPath();
+
+		final String workingPath = getClass()
+			.getResource("/eu/dnetlib/dhp/oa/graph/dump/eosc/working/")
+			.getPath();
+
+		final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+
+		spark
+			.read()
+			.textFile(workingPath + "publication")
+			.write()
+			.text(workingDir.toString() + "/working/publicationextendedaffiliation");
+
+
+
+		SparkUpdateProjectInfo.main(new String[] {
+			"-isSparkSessionManaged", Boolean.FALSE.toString(),
+			"-outputPath", workingDir.toString() + "/",
+			"-sourcePath", sourcePath,
+			"-resultType", "publication",
+			"-workingPath", workingDir.toString() + "/working/",
+				"-preparedInfoPath", workingPath + "preparedInfo"
+		});
+
+		JavaRDD<Result> tmp = sc
+			.textFile(workingDir.toString() + "/working/publicationextendedproject")
+			.map(item -> OBJECT_MAPPER.readValue(item, Result.class));
+
+		Assertions.assertEquals(3, tmp.count());
+
+		Assertions
+			.assertEquals(
+				2,
+				tmp
+					.filter(r -> Optional.ofNullable(r.getProjects()).isPresent() && r.getProjects().size() > 0)
+					.count());
+
+		Assertions
+			.assertEquals(
+				2,
+				tmp
+					.filter(r -> r.getId().equalsIgnoreCase("50|06cdd3ff4700::93859bd27121c3ee7c6ee4bfb1790cba"))
+					.first()
+					.getProjects()
+					.size());
+
+		Assertions
+				.assertEquals(
+						3,
+						tmp
+								.filter(r -> r.getId().equalsIgnoreCase("50|06cdd3ff4700::cd7711c65d518859f1d87056e2c45d98"))
+								.first()
+								.getProjects()
+								.size());
+
+		List<ProjectSummary> projectSummaries = tmp
+			.filter(r -> r.getId().equalsIgnoreCase("50|06cdd3ff4700::93859bd27121c3ee7c6ee4bfb1790cba"))
+			.first()
+			.getProjects();
+
+		Assertions
+			.assertTrue(
+				projectSummaries.stream().anyMatch(p -> p.getFunder().getShortName().equals("NSF")));
+
+		Assertions
+				.assertTrue(
+						projectSummaries.stream().anyMatch(p -> p.getFunder().getShortName().equals("UKRI")));
+
+
+
+		JavaRDD<Project> projects = sc
+				.textFile(workingDir.toString() + "/project")
+				.map(item -> OBJECT_MAPPER.readValue(item, Project.class));
+
+		JavaRDD<Relation> rels = sc
+				.textFile(workingDir.toString() + "/resultProject")
+				.map(item -> OBJECT_MAPPER.readValue(item, Relation.class));
+
+		System.out.println(projects.count());
+
+		Assertions.assertEquals(5, projects.count());
+
+		Assertions.assertEquals(5, rels.count());
+
+		rels.foreach(r -> Assertions.assertTrue(r.getSource().startsWith("50|")));
+		rels.foreach(r -> Assertions.assertTrue(r.getTarget().startsWith("40|")));
+
+	}
+
+	@Test
+	public void selectEoscResults() throws Exception {
+
+		final String sourcePath = getClass()
+			.getResource("/eu/dnetlib/dhp/oa/graph/dump/eosc/input")
+			.getPath();
+
+		final String workingPath = getClass()
+			.getResource("/eu/dnetlib/dhp/oa/graph/dump/eosc/working/")
+			.getPath();
+
+		final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+
+		spark
+				.read()
+				.textFile(workingPath + "publication")
+				.write()
+				.text(workingDir.toString() + "/working/publicationextendedaffiliation");
+
+		spark
+				.read()
+				.textFile(workingPath + "preparedInfo")
+				.write()
+				.text(workingDir.toString() + "/working/preparedInfo");
+
+		ExtendEoscResultWithOrganizationStep2.main(new String[] {
+			"-isSparkSessionManaged", Boolean.FALSE.toString(),
+			"-outputPath", workingDir.toString() + "/",
+			"-sourcePath", sourcePath,
+			"-resultType", "publication",
+			"-workingPath", workingDir.toString() + "/working/"
+		});
+
+
+		JavaRDD<Organization> tmp = sc
+			.textFile(workingDir.toString() + "/organization")
+			.map(item -> OBJECT_MAPPER.readValue(item, Organization.class));
+
+		JavaRDD<Relation> rels = sc
+			.textFile(workingDir.toString() + "/resultProject")
+			.map(item -> OBJECT_MAPPER.readValue(item, Relation.class));
+
+		System.out.println(tmp.count());
+
+		Assertions.assertEquals(2, tmp.count());
+
+		Assertions.assertEquals(2, rels.count());
+
+		rels.foreach(r -> Assertions.assertTrue(r.getSource().startsWith("50|")));
+		rels.foreach(r -> Assertions.assertTrue(r.getTarget().startsWith("20|")));
+
+	}
+
+}

dump/src/test/resources/eu/dnetlib/dhp/oa/graph/dump/eosc/working/preparedInfo/preparedInfo

@@ -0,0 +1,2 @@
+{"projectsList":[{"code":"0430175","funder":{"fundingStream":"Directorate for Computer & Information Science & Engineering","jurisdiction":"US","name":"National Science Foundation","shortName":"NSF"},"id":"40|nsf_________::d1c070f4252c32e23ccc3f4211c9c621","provenance":{"provenance":"Harvested","trust":"0.900"},"title":"Collaborative Research: Temporal Aspects"},{"code":"EP/F01161X/1","funder":{"fundingStream":"EPSRC","jurisdiction":"GB","name":"UK Research and Innovation","shortName":"UKRI"},"id":"40|ukri________::081b09db1211a7b89eb3610d3160e9ba","provenance":{"provenance":"Harvested","trust":"0.900"},"title":"The complexity of valued constraints"}],"resultId":"50|06cdd3ff4700::93859bd27121c3ee7c6ee4bfb1790cba"}
+{"projectsList":[{"acronym":"METIS","code":"317669","funder":{"fundingStream":"FP7","jurisdiction":"EU","name":"European Commission","shortName":"EC"},"id":"40|corda_______::175629cbea2038ed02c85e7132fc4be2","provenance":{"provenance":"Harvested","trust":"0.900"},"title":"Mobile and wireless communications  Enablers for Twenty-twenty (2020) Information Society"},{"code":"unidentified","funder":{"jurisdiction":"CA","name":"Natural Sciences and Engineering Research Council of Canada","shortName":"NSERC"},"id":"40|nserc_______::1e5e62235d094afd01cd56e65112fc63","provenance":{"provenance":"Harvested","trust":"0.900"},"title":"unidentified"},{"acronym":"MiLC","code":"753431","funder":{"fundingStream":"H2020","jurisdiction":"EU","name":"European Commission","shortName":"EC"},"id":"40|corda__h2020::5e49c0ee515f36e416a00cc292dfb310","provenance":{"provenance":"Harvested","trust":"0.900"},"title":"Monotonicity in Logic and Complexity"}],"resultId":"50|06cdd3ff4700::cd7711c65d518859f1d87056e2c45d98"}