D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-hadoop-parent</artifactId>

		<version>1.0.0</version>

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<artifactId>icm-iis-core</artifactId>

	<packaging>jar</packaging>

	<version>1.0.1-SNAPSHOT</version>

	<scm>

	  <developerConnection>

	    scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/icm-iis-core/trunk

	  </developerConnection>

	</scm>

	<properties>

		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

	</properties>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>icm-iis-3rdparty-avrojsoncoders</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.apache.commons</groupId>

			<artifactId>commons-lang3</artifactId>

			<version>3.1</version>

		</dependency>

		<dependency>

			<groupId>commons-cli</groupId>

			<artifactId>commons-cli</artifactId>

			<version>1.2</version>

		</dependency>

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-common</artifactId>

			<version>${iis.hadoop.common.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-common</artifactId>

			<version>${iis.hadoop.common.version}</version>

			<type>test-jar</type>

			<scope>test</scope>

		</dependency>

		<!-- This is needed by map-reduce-related classes -->

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-core</artifactId>

			<version>${iis.hadoop.core.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-hdfs</artifactId>

			<version>${iis.hadoop.hdfs.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-hdfs</artifactId>

			<version>${iis.hadoop.hdfs.version}</version>

			<type>test-jar</type>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.hadoop</groupId>

			<artifactId>hadoop-test</artifactId>

			<version>${iis.hadoop.test.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.oozie</groupId>

			<artifactId>oozie-core</artifactId>

			<version>${iis.oozie.version}</version>

			<scope>provided</scope>

		</dependency>

		<!-- Needed by Avro { -->

		<dependency>

			<groupId>org.apache.avro</groupId>

			<artifactId>avro</artifactId>

			<version>${iis.avro.version}</version>

		</dependency>

		<dependency>

			<groupId>org.apache.avro</groupId>

			<artifactId>avro-mapred</artifactId>

			<version>${iis.avro.version}</version>

			<classifier>hadoop2</classifier>

		</dependency>

		<!-- Needed by Avro } -->

		<!-- Jar containing testing classes that have been missing -->

		<dependency>

			<groupId>org.apache.oozie</groupId>

			<artifactId>oozie-core</artifactId>

			<version>${iis.oozie.version}</version>

			<type>test-jar</type>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>4.10</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>2.2.4</version>

		</dependency>

	</dependencies>

	<build>

		<plugins>

			<!-- Plugin that generates Java classes from Avro schemas -->

			<plugin>

				<groupId>org.apache.avro</groupId>

				<artifactId>avro-maven-plugin</artifactId>

				<version>${iis.avro.version}</version>

				<executions>

					<execution>

						<phase>generate-sources</phase>

						<goals>

							<goal>schema</goal>

							<goal>idl-protocol</goal>

						</goals>

						<configuration>

							<sourceDirectory>${project.basedir}/src/test/resources/</sourceDirectory>

							<outputDirectory>${project.basedir}/target/generated-sources/java/</outputDirectory>

						</configuration>

					</execution>

				</executions>

			</plugin>

			<!-- This plugin makes the Maven->Update Project Configuration not forget 

				about the "target/generated-sources/java" source path -->

			<plugin>

				<groupId>org.codehaus.mojo</groupId>

				<artifactId>build-helper-maven-plugin</artifactId>

				<executions>

					<execution>

						<id>add-source</id>

						<phase>generate-sources</phase>

						<goals>

							<goal>add-source</goal>

						</goals>

						<configuration>

							<sources>

								<source>${project.build.directory}/generated-sources/java/</source>

							</sources>

						</configuration>

					</execution>

				</executions>

			</plugin>

			<!-- Plugin that generates jar with test classes -->

			<plugin>

				<groupId>org.apache.maven.plugins</groupId>

				<artifactId>maven-jar-plugin</artifactId>

				<version>2.2</version>

				<executions>

					<execution>

						<goals>

							<goal>test-jar</goal>

						</goals>

					</execution>

				</executions>

			</plugin>

		</plugins>

		<pluginManagement>

			<plugins>

				<!--This plugin's configuration is used to store Eclipse m2e settings 

					only. It has no influence on the Maven build itself. -->

				<plugin>

					<groupId>org.eclipse.m2e</groupId>

					<artifactId>lifecycle-mapping</artifactId>

					<version>1.0.0</version>

					<configuration>

						<lifecycleMappingMetadata>

							<pluginExecutions>

								<pluginExecution>

									<pluginExecutionFilter>

										<groupId>

											org.apache.avro

										</groupId>

										<artifactId>

											avro-maven-plugin

										</artifactId>

										<versionRange>

											[1.7.4,)

										</versionRange>

										<goals>

											<goal>schema</goal>

											<goal>idl-protocol</goal>

										</goals>

									</pluginExecutionFilter>

									<action>

										<ignore />

									</action>

								</pluginExecution>

								<pluginExecution>

									<pluginExecutionFilter>

										<groupId>

											org.codehaus.mojo

										</groupId>

										<artifactId>

											build-helper-maven-plugin

										</artifactId>

										<versionRange>

											[1.7,)

										</versionRange>

										<goals>

											<goal>add-source</goal>

										</goals>

									</pluginExecutionFilter>

									<action>

										<ignore />

									</action>

								</pluginExecution>

							</pluginExecutions>

						</lifecycleMappingMetadata>

					</configuration>

				</plugin>

			</plugins>

		</pluginManagement>

	</build>

	<repositories>

		<repository>

			<id>cloudera</id>

			<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>

			<releases>

				<enabled>true</enabled>

			</releases>

			<snapshots>

				<enabled>false</enabled>

			</snapshots>

		</repository>

	</repositories>

</project>

package eu.dnetlib.iis.core;

import java.io.DataInputStream;

import java.io.DataOutputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.oozie.util.IOUtils;

/**

 * Provides Oozie-related IO methods

 * @author Mateusz Kobos

 *

 */

public class OozieTestsIOUtils {

	private FileSystem fs;

	private final static String confStorageDirProperty = "hadoop.log.dir";

	private final static String confStorageFileName = "shared_configuration.conf";

	public OozieTestsIOUtils(FileSystem fs) {

		this.fs = fs;

	}

	public void copyResourceToHDFS(String resourcePath, Path destination)

			throws IOException {

		InputStream in = IOUtils.getResourceAsStream(resourcePath, -1);

		OutputStream out = fs.create(destination);

		IOUtils.copyStream(in, out);

	}

	public void copyLocalToHDFS(File source, Path destination)

			throws IOException {

		Path srcPath = new Path(source.getAbsolutePath());

		fs.copyFromLocalFile(srcPath, destination);

	}

//	public void copyHDFSToLocal(Path source, File destination) 

//			throws IOException{

//		InputStream in = fs.open(source);

//		OutputStream out = new FileOutputStream(destination);

//		IOUtils.copyStream(in, out);

//	}

	/** Saves the configuration in a local working directory created by 

	 * Oozie test case for given test method. This method should be called

	 * from the inside of an Oozie test case method, otherwise its functioning

	 * is undefined. The saved information can be retrieved by calling 

	 * {@link loadConfiguration()}.

	 * 

	 * @param conf configuration that should be retrieved from the inside of

	 * the Oozie test case method by calling {@link createJobConf()}

	 * @throws IOException

	 */

	public static void saveConfiguration(Configuration conf) throws IOException{

		File f = new File(System.getProperty(confStorageDirProperty), 

				confStorageFileName);

		DataOutputStream out = new DataOutputStream(new FileOutputStream(f));

		conf.write(out);

		out.close();

	}

	/** A counterpart of the {@link saveConfiguration()} method. It should be

	 * called from the inside of a code run on the Hadoop created 

	 * by Oozie test case.

	 * 

	 * @return

	 * @throws IOException

	 */

	public static Configuration loadConfiguration() throws IOException{

		File f = new File(System.getProperty(confStorageDirProperty), 

				confStorageFileName);

		DataInputStream in = new DataInputStream(new FileInputStream(f));

		Configuration conf = new Configuration();

		conf.readFields(in);

		in.close();

		return conf;

	}

}

package eu.dnetlib.iis.core.java;

import static org.junit.Assert.assertEquals;

import java.net.URISyntaxException;

import java.util.HashMap;

import java.util.Map;

import org.apache.commons.cli.CommandLine;

import org.apache.hadoop.fs.Path;

import org.junit.Test;

import eu.dnetlib.iis.core.java.CmdLineParserException;

import eu.dnetlib.iis.core.java.CmdLineParserForProcessRunParameters;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.Ports;

import eu.dnetlib.iis.core.java.ProcessParameters;

import eu.dnetlib.iis.core.java.porttype.AnyPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * 

 * @author Mateusz Kobos

 *

 */

public class CmdLineParserForProcessRunParametersTest{

	@Test

	public void testBasic() throws URISyntaxException {

		CmdLineParserForProcessRunParameters parser = 

				new CmdLineParserForProcessRunParameters();

		String[] args = new String[]{

				"-Iperson=/users/joe/person_input",

				"-Idocument=/users/joe/doc_input",

				"-Omerged=/users/joe/merged_out",

				"-Page=33"};

		Ports ports = createStandardPorts();

		CommandLine cmdLine = CmdLineParser.parse(args);

		ProcessParameters actual = parser.run(cmdLine, ports);

		ProcessParameters expected = createStandardCmdLineParams();

		assertEquals(expected, actual);

	}

	@Test(expected=CmdLineParserException.class)

	public void testTooManyPorts() {

		CmdLineParserForProcessRunParameters parser = 

				new CmdLineParserForProcessRunParameters();

		String[] args = new String[]{

				"-Iperson=/users/joe/person_input",

				"-Idocument=/users/joe/doc_input",

				"-Omerged=/users/joe/merged_out",

				"-Oother=/users/joe/other_out"};

		Ports ports = createStandardPorts();

		CommandLine cmdLine = CmdLineParser.parse(args);

		parser.run(cmdLine, ports);

	}

	@Test(expected=CmdLineParserException.class)

	public void testTooFewPorts() {

		CmdLineParserForProcessRunParameters parser = 

				new CmdLineParserForProcessRunParameters();

		String[] args = new String[]{

				"-Iperson=/users/joe/person_input",

				"-Omerged=/users/joe/merged_out"};

		Ports ports = createStandardPorts();

		CommandLine cmdLine = CmdLineParser.parse(args);

		parser.run(cmdLine, ports);

	}

	private static Ports createStandardPorts(){

		HashMap<String, PortType> inputPorts = 

				new HashMap<String, PortType>();

		inputPorts.put("person", new AnyPortType());

		inputPorts.put("document", new AnyPortType());

		HashMap<String, PortType> outputPorts = 

				new HashMap<String, PortType>();

		outputPorts.put("merged", new AnyPortType());

		return new Ports(inputPorts, outputPorts);	

	}

	private static ProcessParameters createStandardCmdLineParams() 

			throws URISyntaxException{

		HashMap<String, Path> inputBinding = new HashMap<String, Path>();

		inputBinding.put("person", 

				new Path("/users/joe/person_input"));

		inputBinding.put("document", 

				new Path("/users/joe/doc_input"));

		HashMap<String, Path> outputBinding = new HashMap<String, Path>();

		outputBinding.put("merged", 

				new Path("/users/joe/merged_out"));

		PortBindings expectedBindings = 

				new PortBindings(inputBinding, outputBinding);

		Map<String, String> params = new HashMap<String, String>();

		params.put("age", "33");

		ProcessParameters cmdLineParams = new ProcessParameters(

				expectedBindings, params);

		return cmdLineParams;

	}

}

package eu.dnetlib.iis.core.java.io;

import java.io.ByteArrayOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.StringWriter;

import java.util.List;

import org.apache.commons.io.IOUtils;

import org.junit.Test;

import junit.framework.Assert;

import eu.dnetlib.iis.core.schemas.standardexamples.Document;

/**

 * @author Mateusz Kobos

 */

public class JsonStreamWriterTest {

	@Test

	public void basicTest() throws IOException{

		ByteArrayOutputStream out = new ByteArrayOutputStream();

		JsonStreamWriter<Document> writer = 

				new JsonStreamWriter<Document>(Document.SCHEMA$, out);

		List<Document> documents = DataStoreExamples.getDocument();

		for(Document d: documents){

			writer.write(d);

		}

		writer.close();

		String actual = out.toString();

		InputStream in = Thread.currentThread().getContextClassLoader()

			.getResourceAsStream("eu/dnetlib/iis/core/java/io/document.json");

		String expected = toString(in);

		Assert.assertEquals(expected, actual);

	}

	private static String toString(InputStream in) throws IOException{

		StringWriter writer = new StringWriter();

		IOUtils.copy(in, writer);

		return writer.toString();

	}

}

package eu.dnetlib.iis.core.java.io;

import java.io.File;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.List;

import java.util.NoSuchElementException;

import org.apache.avro.Schema;

import org.apache.avro.file.DataFileWriter;

import org.apache.avro.generic.GenericContainer;

import org.apache.commons.io.FileUtils;

import org.junit.After;

import org.junit.Assert;

import org.junit.Before;

import org.junit.Test;

import com.google.common.io.Files;

import eu.dnetlib.iis.core.TestsIOUtils;

import eu.dnetlib.iis.core.schemas.standardexamples.Document;

import eu.dnetlib.iis.core.schemas.standardexamples.DocumentWithoutTitle;

import eu.dnetlib.iis.core.java.io.AvroDataStoreReader;

import eu.dnetlib.iis.core.java.io.DataStore;

import eu.dnetlib.iis.core.java.io.FileSystemPath;

/**

 * @author Mateusz Kobos

 */

public class DataStoreTest {

	private File tempDir = null;

	@Before

	public void setUp() throws IOException{

		tempDir = Files.createTempDir();

	}

	@After

	public void tearDown() throws IOException{

		FileUtils.deleteDirectory(tempDir);

	}

	@Test

	public void testSingleFile() throws IOException {

		List<Document> documents = DataStoreExamples.getDocument();

		FileSystemPath path = new FileSystemPath(new File(tempDir, "doc"));

		DataStore.create(documents, path);

		TestsIOUtils.assertEqualSets(documents, 

				new AvroDataStoreReader<Document>(path));	

	}

	@Test

	public void testReaderSchema() throws IOException {

		List<Document> documents = DataStoreExamples.getDocument();

		FileSystemPath path = new FileSystemPath(new File(tempDir, "doc"));

		DataStore.create(documents, path);

		List<DocumentWithoutTitle> documentsWithoutTitle = 

				DataStoreExamples.getDocumentWithoutTitle();

		TestsIOUtils.assertEqualSets(documentsWithoutTitle, 

				new AvroDataStoreReader<DocumentWithoutTitle>(

						path, DocumentWithoutTitle.SCHEMA$));	

	}

	@Test

	public void testClose() throws IOException{

		List<Document> documents = DataStoreExamples.getDocument();

		FileSystemPath path = new FileSystemPath(new File(tempDir, "doc"));

		DataStore.create(documents, path);

		AvroDataStoreReader<Document> reader = 

				new AvroDataStoreReader<Document>(path);

		Assert.assertEquals(documents.get(0), reader.next());

		Assert.assertEquals(documents.get(1), reader.next());

		reader.close();

		try{

			reader.next();

			Assert.fail("Didn't throw an exception when it supposed to");

		} catch(NoSuchElementException ex){

		}

	}

	@Test

	public void testManyFiles() throws IOException{

		List<Document> documents = DataStoreExamples.getDocument();

		createSingleFile(documents.subList(0, 3), 

				new FileSystemPath(new File(tempDir, "doc0")), 

				Document.SCHEMA$);

		createSingleFile(documents.subList(3, 4), 

				new FileSystemPath(new File(tempDir, "doc1")),

				Document.SCHEMA$);

		TestsIOUtils.assertEqualSets(documents, 

				new AvroDataStoreReader<Document>(

						new FileSystemPath(tempDir)));	

	}

	@Test

	public void testEmptyFiles() throws IOException{

		List<Document> documents = DataStoreExamples.getDocument();

		createSingleFile(documents.subList(0, 3), 

				new FileSystemPath(new File(tempDir, "doc0")),

				Document.SCHEMA$);

		createSingleFile(Arrays.<Document>asList(), 

				new FileSystemPath(new File(tempDir, "doc881")),

				Document.SCHEMA$);

		createSingleFile(documents.subList(3, 4), 

				new FileSystemPath(new File(tempDir, "doc22")),

				Document.SCHEMA$);

		createSingleFile(Arrays.<Document>asList(), 

				new FileSystemPath(new File(tempDir, "doc4")),

				Document.SCHEMA$);

		TestsIOUtils.assertEqualSets(documents, 

				new AvroDataStoreReader<Document>(

						new FileSystemPath(tempDir)));	

	}

	private static <T extends GenericContainer> void createSingleFile(

			List<T> elements, FileSystemPath path, Schema schema) 

					throws IOException{

		DataFileWriter<T> writer = DataStore.createSingleFile(path, schema);

		for(T i: elements){

			writer.append(i);

		}

		writer.close();

	}

}

class DataStoreExamples{

	static String bookTitleExtra = "An extraordinary book";

	static String bookTitleBasics = "Basics of the basics";

	static String bookTitleMoreInteresting = "Even more of interesting stuff";

	static String bookTitleEscapeCodes = "2.2. Stellar Rotation Hertszprung\u2013Russell diagram precludes detecting\n6.2. Conclusions\n";

	/**

	 * Sample data on some documents

	 */

	public static List<Document> getDocument(){

		ArrayList<Document> list = new ArrayList<Document>();

		list.add(new Document(20, bookTitleExtra,

				new ArrayList<Integer>(Arrays.asList(1, 20))));

		list.add(new Document(1, bookTitleBasics,

				new ArrayList<Integer>(Arrays.asList(10, 6, 1))));

		list.add(new Document(2, null, 

				new ArrayList<Integer>()));

		list.add(new Document(6, bookTitleMoreInteresting,

				new ArrayList<Integer>(Arrays.asList(1, 6))));

		return list;

	}

	/**

	 * The same as the {@link getDocument} but does not include document title

	 */

	public static List<DocumentWithoutTitle> getDocumentWithoutTitle(){

		ArrayList<DocumentWithoutTitle> list = new ArrayList<DocumentWithoutTitle>();

		list.add(new DocumentWithoutTitle(20, 

				new ArrayList<Integer>(Arrays.asList(1, 20))));

		list.add(new DocumentWithoutTitle(1, 

				new ArrayList<Integer>(Arrays.asList(10, 6, 1))));

		list.add(new DocumentWithoutTitle(2,

				new ArrayList<Integer>()));

		list.add(new DocumentWithoutTitle(6, 

				new ArrayList<Integer>(Arrays.asList(1, 6))));

		return list;

	}

	public static List<Document> getDocumentWithUnicodeEscapeCodes(){

		ArrayList<Document> list = new ArrayList<Document>();

		list.add(new Document(20, bookTitleEscapeCodes,

				new ArrayList<Integer>(Arrays.asList(1, 20))));

		return list;

	}

}

package eu.dnetlib.iis.core.java.io;

import java.io.IOException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.List;

import org.junit.Test;

import eu.dnetlib.iis.core.TestsIOUtils;

import eu.dnetlib.iis.core.schemas.standardexamples.Document;

/**

 * @author Mateusz Kobos

 */

public class JsonStreamReaderTest {

	@Test

	public void basicTest() throws IOException{

		InputStream in = Thread.currentThread().getContextClassLoader()

			.getResourceAsStream("eu/dnetlib/iis/core/java/io/document.json");

		CloseableIterator<Document> reader = new JsonStreamReader<Document>(

				Document.SCHEMA$, in, Document.class);

		List<Document> expected = DataStoreExamples.getDocument();

		List<Document> actual = new ArrayList<Document>();

		while(reader.hasNext()){

			Object record = reader.next();

			Document read = (Document) record;

			actual.add(read);

		}

		TestsIOUtils.assertEqualSets(expected, actual);	

	}

}

package eu.dnetlib.iis.core.java.io;

import java.io.File;

import java.io.IOException;

import java.io.InputStream;

import java.util.List;

import junit.framework.Assert;

import org.apache.avro.Schema;

import org.apache.commons.io.FileUtils;

import org.apache.commons.io.IOUtils;

import org.junit.After;

import org.junit.Before;

import org.junit.Test;

import com.google.common.io.Files;

import eu.dnetlib.iis.core.TestsIOUtils;

import eu.dnetlib.iis.core.schemas.standardexamples.Document;

/**

 * 

 * @author Mateusz Kobos

 *

 */

public class JsonUtilsTest {

	private File tempDir = null;

	@Before

	public void setUp() throws IOException{

		tempDir = Files.createTempDir();

	}

	@After

	public void tearDown() throws IOException{

		FileUtils.deleteDirectory(tempDir);

	}

	@Test

	public void testConvertToDataStoreSimple() throws IOException{

		checkConvertToDataStore(DataStoreExamples.getDocument(), 

				"eu/dnetlib/iis/core/java/io/document.json", Document.SCHEMA$);

	}

	@Test

	public void testConvertToDataStoreWithEscapeCodes() throws IOException{

		checkConvertToDataStore(DataStoreExamples.getDocumentWithUnicodeEscapeCodes(), 

				"eu/dnetlib/iis/core/java/io/document_with_unicode_escape_codes.json", 

				Document.SCHEMA$);

	}

	private <T> void checkConvertToDataStore(List<T> expectedRecords,

			String actualResourcePath, Schema actualSchema) throws IOException{

		InputStream in = Thread.currentThread().getContextClassLoader()

				.getResourceAsStream(actualResourcePath);

		FileSystemPath outPath = new FileSystemPath(new File(tempDir, "record"));

		JsonUtils.convertToDataStore(actualSchema, in, outPath);

		TestsIOUtils.assertEqualSets(expectedRecords, 

				new AvroDataStoreReader<T>(outPath));

	}

	@Test

	public void testConvertToListWithUnicodeEscapeCodes() throws IOException{

		List<Document> actual = JsonUtils.convertToList(

				"eu/dnetlib/iis/core/java/io/document_with_unicode_escape_codes.json",

				Document.SCHEMA$, Document.class);

		List<Document> expected = DataStoreExamples.getDocumentWithUnicodeEscapeCodes();

		TestsIOUtils.assertEqualSets(expected, actual);			

	}

	@Test

	public void testPrettyPrint() throws IOException{

		String uglyJson = getStringFromResourceFile(

				"eu/dnetlib/iis/core/java/io/json_pretty_print/ugly_json.json");

		String actual = JsonUtils.toPrettyJSON(uglyJson);

		String expected = getStringFromResourceFile(

				"eu/dnetlib/iis/core/java/io/json_pretty_print/expected_pretty_json.json");

		Assert.assertEquals(expected, actual);

	}

	private String getStringFromResourceFile(String resourcePath) throws IOException{

		InputStream in = Thread.currentThread().getContextClassLoader()

				.getResourceAsStream(resourcePath);

		return IOUtils.toString(in, "UTF-8");

	}

}

package eu.dnetlib.iis.core.java;

import static org.junit.Assert.assertEquals;

import java.util.Map;

import org.apache.commons.cli.CommandLine;

import org.apache.hadoop.conf.Configuration;

import org.junit.Test;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * 

 * @author Mateusz Kobos

 *

 */

class DummyProcess implements Process {

	public DummyProcess(){	

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return null;

	}

	@Override

	public Map<String, PortType> getOutputPorts(){

		return null;

	}

	@Override

	public void run(PortBindings portBindings, Configuration configuration, 

			Map<String, String> parameters){

	}

}

class DummyProcessWithParametersConstructor implements Process{

	public String[] params;

	public DummyProcessWithParametersConstructor(String[] params){

		this.params = params;

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return null;

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return null;

	}

	@Override

	public void run(PortBindings portBindings, Configuration configuration,

			Map<String, String> parameters) throws Exception {

	}

}

public class CmdLineParserForProcessConstructionTest{

	@Test

	public void testTrivial() {

		String[] args = new String[]{

				"eu.dnetlib.iis.core.java.DummyProcess"};

		CmdLineParserForProcessConstruction cmdLineParser = 

				new CmdLineParserForProcessConstruction();

		CommandLine cmdLine = CmdLineParser.parse(args);

		Process actual = cmdLineParser.run(cmdLine);

		assertEquals(DummyProcess.class, actual.getClass());

	}

	@Test

	public void testWithSomeParameters() {

		String[] args = new String[]{

				"eu.dnetlib.iis.core.java.DummyProcess",

				"-Iperson=hdfs://localhost:8020/users/joe/person_input",

				"-Idocument=hdfs://localhost:8020/users/joe/doc_input",

				"-Omerged=hdfs://localhost:8020/users/joe/merged_out",

				"-SclassName=java.util.String"};

		CmdLineParserForProcessConstruction cmdLineParser = 

				new CmdLineParserForProcessConstruction();

		CommandLine cmdLine = CmdLineParser.parse(args);

		Process actual = cmdLineParser.run(cmdLine);

		assertEquals(DummyProcess.class, actual.getClass());

	}

	@Test

	public void testWithSomeParametersWithParametersConstructor() {

		String[] constructorParams = new String[]{

				"some string", "some other string"};

		String[] args = new String[]{

				"eu.dnetlib.iis.core.java.DummyProcessWithParametersConstructor",

				"-C" + constructorParams[0],

				"-C" + constructorParams[1],

				"-Iperson=hdfs://localhost:8020/users/joe/person_input",

				"-Idocument=hdfs://localhost:8020/users/joe/doc_input",

				"-Omerged=hdfs://localhost:8020/users/joe/merged_out",

				"-SclassName=java.util.String"};

		CmdLineParserForProcessConstruction cmdLineParser = 

				new CmdLineParserForProcessConstruction();

		CommandLine cmdLine = CmdLineParser.parse(args);

		DummyProcessWithParametersConstructor actual = 

				(DummyProcessWithParametersConstructor) cmdLineParser.run(cmdLine);

		assertEquals(DummyProcessWithParametersConstructor.class, 

				actual.getClass());

		for(int i = 0; i < constructorParams.length; i++){

			assertEquals(constructorParams[i], actual.params[i]);

		}

	}

}

package eu.dnetlib.iis.core.java;

import org.apache.hadoop.conf.Configuration;

import eu.dnetlib.iis.core.OozieTestsIOUtils;

import eu.dnetlib.iis.core.java.ProcessWrapper;

/**

 * A version of the {@link ProcessWrapper} which uses the Configuration

 * object created by the Oozie unit test framework. This is used in the

 * XML workflow definition in the tests instead of {@link ProcessWrapper}. 

 * It makes referencing the underlying file system by the java Process run

 * during the tests possible.

 * @author Mateusz Kobos

 * 

 */

public class ProcessWrapperForTests extends ProcessWrapper {

	public Configuration getConfiguration() throws Exception {

		return OozieTestsIOUtils.loadConfiguration();

	}

	public static void main(String[] args) throws Exception {

		ProcessWrapper wrapper = new ProcessWrapperForTests();

		wrapper.run(args);

	}

}

package eu.dnetlib.iis.core.java.jsonworkflownodes;

import java.util.Map;

import junit.framework.Assert;

import org.junit.Test;

import eu.dnetlib.iis.core.common.AvroUtils;

import eu.dnetlib.iis.core.java.porttype.AvroPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

public class ProducerAndConsumerTest{

	@Test

	public void testPortCreationInProducer(){

		checkPortsCreator(new PortsCreator() {

			@Override 

			public Map<String, PortType> getPorts(String[] specificationStrings){

				Producer producer = 

						new Producer(specificationStrings);

				Assert.assertEquals(0, producer.getInputPorts().size());

				return producer.getOutputPorts();

			}

		});

	}

	@Test

	public void testPortCreationInTestingConsumer(){

		checkPortsCreator(new PortsCreator() {

			@Override 

			public Map<String, PortType> getPorts(String[] specificationStrings){

				TestingConsumer consumer = 

						new TestingConsumer(specificationStrings);

				Assert.assertEquals(0, consumer.getOutputPorts().size());

				return consumer.getInputPorts();

			}

		});

	}

	private void checkPortsCreator(PortsCreator creator) {

		PortSpec[] specs = new PortSpec[] {

				new PortSpec(

						"document",

						"eu.dnetlib.iis.core.schemas.standardexamples.Document",

						"fake/path/document.json"),

				new PortSpec("person",

						"eu.dnetlib.iis.core.schemas.standardexamples.Person",

						"fake/path/person.json") };

		String[] specsStr = new String[specs.length];

		for (int i = 0; i < specs.length; i++) {

			specsStr[i] = String.format("{%s, %s, %s}", specs[i].name,

					specs[i].schemaPath, specs[i].jsonPath);

		}

		Map<String, PortType> outs = creator.getPorts(specsStr);

		Assert.assertEquals(specs.length, outs.size());

		for (int i = 0; i < specs.length; i++) {

			AvroPortType actual = (AvroPortType) outs.get(specs[i].name);

			AvroPortType expected = new AvroPortType(

					AvroUtils.toSchema(specs[i].schemaPath));

			Assert.assertEquals(expected, actual);

		}

	}

}

interface PortsCreator{

	Map<String, PortType> getPorts(String[] specificationStrings);

}

class PortSpec{

	public String name;

	public String schemaPath;

	public String jsonPath;

	public PortSpec(String name, String schemaPath, String jsonPath){

		this.name = name;

		this.schemaPath = schemaPath;

		this.jsonPath = jsonPath;

	}

}

/** The package contains producer and consumer workflow nodes that are 

 * configured using JSON format to respectively produce and consume 

 * data stores. 

 */

package eu.dnetlib.iis.core.java.jsonworkflownodes;

package eu.dnetlib.iis.core.java.jsonworkflownodes;

import java.io.IOException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import org.apache.avro.specific.SpecificRecord;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import eu.dnetlib.iis.core.TestsIOUtils;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.Process;

import eu.dnetlib.iis.core.java.io.DataStore;

import eu.dnetlib.iis.core.java.io.FileSystemPath;

import eu.dnetlib.iis.core.java.io.JsonUtils;

import eu.dnetlib.iis.core.java.jsonworkflownodes.PortSpecifications.SpecificationValues;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * Reads the data stores from specified input ports and compares them with

 * expected JSON-encoded data stores. If There is a mismatch,

 * an exception is thrown.

 *

 * @author Mateusz Kobos

 */

public class TestingConsumer implements Process {

	private final PortSpecifications inputSpecs;

	/**

	 * @param inputSpecifications specifications of input. Each element of

	 * the array corresponds to a single specification. Single specification

	 * conforms to the following template:

	 * "{input port name, schema reference, path to JSON file in resources 

	 * corresponding to the expected input data store}",

	 * e.g. "{person, eu.dnetlib.iis.core.examples.schemas.documentandauthor.Person, 

	 * eu/dnetlib/iis/core/examples/person.json}"

	 */

	public TestingConsumer(String[] inputSpecifications){

		inputSpecs = new PortSpecifications(inputSpecifications);

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return inputSpecs.getPortTypes();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return new HashMap<String, PortType>();

	}

	@Override

	public void run(PortBindings portBindings, Configuration configuration,

			Map<String, String> parameters) throws Exception {

		Map<String, Path> input = portBindings.getInput();

		FileSystem fs = FileSystem.get(configuration);

		for(Map.Entry<String, Path> e: input.entrySet()){

			SpecificationValues specs = inputSpecs.get(e.getKey());

			check(new FileSystemPath(fs, e.getValue()), specs);

		}

	}

	private static void check(FileSystemPath actualPath, SpecificationValues specs) throws IOException{

		List<SpecificRecord> expected = JsonUtils.convertToList(

				specs.jsonFilePath, specs.schema, SpecificRecord.class);

		List<SpecificRecord> actual = DataStore.read(actualPath, specs.schema);

		TestsIOUtils.assertEqualSets(expected, actual, true);

	}

}

package eu.dnetlib.iis.core;

import java.util.Properties;

import org.apache.oozie.client.WorkflowJob;

/**

 * Configuration of the testing workflow

 * @author Mateusz Kobos

 */

public class WorkflowConfiguration {

	/** I had to change the duration of the wait from the original 

	 * 15 seconds because the original time wasn't enough for my 

	 * computer to execute the Oozie workflow. To be more precise, 

	 * the test ended with the following failure message: 

	 * junit.framework.AssertionFailedError: expected:<SUCCEEDED> but was:<RUNNING>

	 */

	public static final int defaultTimeoutInSeconds = 360;

	public static final WorkflowJob.Status defaultExpectedFinishStatus = 

			WorkflowJob.Status.SUCCEEDED;

	public static final Properties defaultJobProperties = null;

	private Properties jobProps = defaultJobProperties; 

	private int timeoutInSeconds = defaultTimeoutInSeconds;

	private WorkflowJob.Status expectedFinishStatus = defaultExpectedFinishStatus;

	public WorkflowConfiguration(){

	}

	/**

	 * See {@link #setJobProps} for description

	 * @return

	 */

	public Properties getJobProps() {

		return jobProps;

	}

	/**

	 * @param jobProps job properties

	 */

	public WorkflowConfiguration setJobProps(Properties jobProps) {

		this.jobProps = jobProps;

		return this;

	}

	/**

	 * See {@link #setTimeoutInSeconds} for description

	 */

	public int getTimeoutInSeconds() {

		return timeoutInSeconds;

	}

	/**

	 * @param timeoutInSeconds timeout in seconds. Workflow will be killed

     * if timeout is exceeded

	 */

	public WorkflowConfiguration setTimeoutInSeconds(int timeoutInSeconds) {

		this.timeoutInSeconds = timeoutInSeconds;

		return this;

	}

	/**

	 * See {@link #setExpectedFinishStatus} for description

	 * @return

	 */

	public WorkflowJob.Status getExpectedFinishStatus() {

		return expectedFinishStatus;

	}

	/**

	 * @param expectedFinishStatus expected status of the workflow after its

	 * finish

	 */

	public WorkflowConfiguration setExpectedFinishStatus(WorkflowJob.Status expectedFinishStatus) {

		this.expectedFinishStatus = expectedFinishStatus;

		return this;

	}

}

package eu.dnetlib.iis.core;

import java.io.BufferedReader;

import java.io.File;

import java.io.FileReader;

import java.io.FileWriter;

import java.io.IOException;

import java.io.PrintWriter;

import java.util.ArrayList;

import java.util.List;

/**

 * Replace strings conforming to given regexes "on the flight", i.e. while copying

 * file from one place to another.

 *

 * This is used in order to change the Oozie's "workflow.xml" so that it 

 * conforms to the requirements of the MiniOozie testing engine.

 * 

 * @author Mateusz Kobos

 *

 */

public class StringReplacer {

	private static final List<StringReplacement> replacements = getReplacements();

	private static List<StringReplacement> getReplacements(){

		List<StringReplacement> list = new ArrayList<StringReplacement>();

		/** We have to use a special wrapper for Java workflow nodes in order

		 * to make them use the special mock HDFS used during the tests.

		 */

		list.add(new StringReplacement(

				"ProcessWrapper", 

				"ProcessWrapperForTests"));

		/** The paths to Avro schemas stored in HDFS are understood differently 

		 * when running on real cluster and during the tests. When running

		 * during the tests, a file system prefix "hdfs://" is absent, which

		 * confuses the "avro-json" library that allows for reading and writing

		 * Avro files in Hadoop Streaming workflow node. See the code of

		 * {@code com.cloudera.science.avro.common.SchemaLoader.loadFromUrl()}

		 * method for details where this confusion comes from. 

		 * 

		 * Here, we're adding a prefix which is the address of the name

		 * node which is something that "avro-json" library expects.

		 */

		list.add(new StringReplacement(

				"<value>(\\s*\\$\\{wf:appPath\\(\\)\\}.*\\.avsc\\s*)</value>",

				"<value>\\${nameNode}$1</value>"));

		return list;

	}

	public void replace(File in, File out) throws IOException {

		BufferedReader reader = null;

		PrintWriter writer = null;

		try {

			reader = new BufferedReader(new FileReader(in));

			writer = new PrintWriter(new FileWriter(out));

			String line = null;

			while ((line = reader.readLine()) != null){

				String newLine = line;

				for(StringReplacement replacement: replacements){

					newLine = replacement.replaceAll(newLine);

				}

				writer.println(newLine);

			}

		} finally {

			if(reader != null){

				reader.close();

			}

			if (writer != null){

				writer.close();

			}

		}

	}

}

class StringReplacement{

	private String regex;

	private String replacement;

	public StringReplacement(String regex, String replacement) {

		this.regex = regex;

		this.replacement = replacement;

	}

	public String replaceAll(String str){

		return str.replaceAll(regex, replacement);

	}

}

package eu.dnetlib.iis.core.javamapreduce.hack;

import junit.framework.Assert;

import org.apache.avro.Schema;

import org.apache.avro.mapred.Pair;

import org.apache.hadoop.conf.Configuration;

import org.junit.Test;

import eu.dnetlib.iis.core.schemas.standardexamples.Document;

import eu.dnetlib.iis.core.schemas.standardexamples.Person;

import eu.dnetlib.iis.core.schemas.standardexamples.personwithdocuments.PersonWithDocuments;

import eu.dnetlib.iis.core.javamapreduce.hack.oldapi.SchemaSetter;

public class SchemaSetterTest {

	@Test

	public void testBasic() {

		Configuration conf = new Configuration();

		conf.set(SchemaSetter.inputClassName, 

			"eu.dnetlib.iis.core.schemas.standardexamples.Document");

		conf.set(SchemaSetter.mapOutputKeyClassName,

			SchemaSetter.primitiveTypePrefix+"DOUBLE");

		conf.set(SchemaSetter.mapOutputValueClassName,

			"eu.dnetlib.iis.core.schemas.standardexamples.Person");

		conf.set(SchemaSetter.outputClassName,

			"eu.dnetlib.iis.core.schemas.standardexamples.personwithdocuments.PersonWithDocuments");

		SchemaSetter.set(conf);

		Assert.assertEquals(Document.SCHEMA$.toString(), 

				conf.get(SchemaSetter.avroInput));

		Assert.assertEquals(Pair.getPairSchema(

					Schema.create(Schema.Type.DOUBLE),

					Person.SCHEMA$).toString(), 

				conf.get(SchemaSetter.avroMapOutput));

		Assert.assertEquals(PersonWithDocuments.SCHEMA$.toString(),

				conf.get(SchemaSetter.avroOutput));

	}

}

package eu.dnetlib.iis.core.common;

import static org.junit.Assert.*;

import org.junit.Test;

public class AvroToolsTests {

	@Test

	public void testToSchemaPrimitiveType() {

		checkSchema("\"string\"", "org.apache.avro.Schema.Type.STRING");

	}

	@Test

	public void testToSchemaClassName(){

		checkSchema("{\"type\":\"record\",\"name\":\"Person\",\"namespace\":\"eu.dnetlib.iis.core.schemas.standardexamples\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"}]}", 

				"eu.dnetlib.iis.core.schemas.standardexamples.Person");

	}

	private static void checkSchema(String expected, String className){

		String actual = AvroUtils.toSchema(className).toString();

		assertEquals(expected, actual);

	}

}

package eu.dnetlib.iis.core;

import java.io.IOException;

import java.util.Properties;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.mapred.JobConf;

import org.apache.oozie.client.*;

import org.apache.oozie.local.LocalOozie;

import org.apache.oozie.service.XLogService;

import org.apache.oozie.test.MiniOozieTestCase;

import org.apache.oozie.util.XLog;

import org.junit.After;

import org.junit.Before;

/**

 * A basis to be used as a base class of all test case classes that run 

 * Oozie workflows.

 * <p/>

 * ATTENTION: In order for your test case class that has this class as a parent 

 * to run properly, the Maven project which contains this class has to have

 * a directory called "core" in its root directory with appropriate contents. 

 * See the "core" directory in the "icm-iis-core-examples" to see what's 

 * the required content of this directory. 

 * You can copy this "core" directory to your project.

 * <p/>

 * <p/>

 * If something goes wrong with your test and you want to see the Hadoop logs 

 * to check what's the matter, you can look into directory 

 * <p/>

 * target/test-data/oozietests/NAME_OF_YOUR_TEST_CLASS/NAME_OF_YOUR_TEST_METHOD/userlogs

 * <p/>

 * and its subdirectories. 

 * <p/>

 * Unfortunately, I don't know where the files stored in virtual HDFS really 

 * are (such an information would also be helpful during debugging).

 *  

 * @author Mateusz Kobos

 *

 */

public abstract class AbstractWorkflowTestCase extends MiniOozieTestCase {

	private static final char WORKFLOW_LOCATION_SEPARATOR = '/';

	private static final String JOB_PROPERTIES_FILE_NAME = "job.properties";

	@Override

	@Before

	protected void setUp() throws Exception {

		System.setProperty(XLogService.LOG4J_FILE, "oozie-log4j.properties");

		log = new XLog(LogFactory.getLog(getClass()));

		super.setUp();

	}

	@Override

	@After

	protected void tearDown() throws Exception {

		super.tearDown();

	}

	/**

	 * See {@link #runWorkflow(String resourcesOozieAppLocalPath, WorkflowConfiguration config)}

	 * method for description of parameters

	 */

	public RemoteOozieAppManager runWorkflow(String resourcesOozieAppLocalPath) 

			throws IOException, OozieClientException{

		return runWorkflow(resourcesOozieAppLocalPath, new WorkflowConfiguration());

	}

	/**

	 * Run the workflow

	 * @param resourcesOozieAppLocalPath path to the directory containing the Oozie

     * application. This directory contains the main {@code workflow.xml} file

     * @param config optional configuration of the workflow

	 */

	public RemoteOozieAppManager runWorkflow(

			String resourcesOozieAppLocalPath, WorkflowConfiguration config)

					throws IOException, OozieClientException{

		return runWorkflow(resourcesOozieAppLocalPath, config, false);

	}

	/**

	 * Run the workflow

	 * @param resourcesOozieAppLocalPath path to the directory containing the Oozie

     * application. This directory contains the main {@code workflow.xml} file

     * @param config optional configuration of the workflow

     * @param skipPredefinedJobProperties skips reading predefined job.properties if any set

	 */

	public RemoteOozieAppManager runWorkflow(

			String resourcesOozieAppLocalPath, WorkflowConfiguration config,

			boolean skipPredefinedJobProperties)

					throws IOException, OozieClientException{

//		File oozieAppLocalPath = 

//		new File(Thread.currentThread().getContextClassLoader()

//				.getResource(resourcesOozieAppLocalPath).getPath());

		RemoteOozieAppManager appManager = RemoteOozieAppManager.fromPrimedClassDir(

			getFileSystem(), getFsTestCaseDir(), resourcesOozieAppLocalPath);

		Properties props = createWorkflowConfiguration(

				createJobConf(), getTestUser(),	appManager.getOozieAppPath(), 

				appManager.getWorkingDir(), 

				skipPredefinedJobProperties?null:

						loadPredefinedJobProperties(resourcesOozieAppLocalPath),

				config.getJobProps());

		runWorkflowBasedOnConfiguration(props, config.getTimeoutInSeconds(),

				config.getExpectedFinishStatus());

		return appManager;

	}