D-Net

[

{

  "type_source": "SVN", 

  "goal": "package -U -T 4C source:jar", 

  "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/icm-iis-core/trunk/", 

  "deploy_repository": "dnet4-snapshots", 

  "version": "4",

  "mail": "m.horst@icm.edu.pl",

  "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet4-snapshots", 

  "name": "icm-iis-core"

},

{

  "type_source": "SVN",

  "goal": "clean verify -U -e -X",

  "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/icm-iis-core/trunk/",

  "nightly" : "true",

  "cron" : "H H * * *",

  "version": "4",

  "mail": "m.horst@icm.edu.pl",

  "name": "icm-iis-core-embedded-integration-test"

}

]

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",

				"-SworkingDir=/users/joe/working_dir",

				"-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",

				"-SworkingDir=/users/joe/working_dir"};

		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",

				"-SworkingDir=/users/joe/working_dir"};

		Ports ports = createStandardPorts();

		CommandLine cmdLine = CmdLineParser.parse(args);

		parser.run(cmdLine, ports);

	}

	@Test(expected=CmdLineParserException.class)

	public void testMissingWorkingDir() 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"};

		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);

		Path expectedWorkingDir = 

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

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

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

		ProcessParameters cmdLineParams = new ProcessParameters(

				expectedBindings, expectedWorkingDir, 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.junit.Test;

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

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

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

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

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, HadoopContext context, 

			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, HadoopContext context,

			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",

				"-SworkingDir=hdfs://localhost:8020/users/joe/working_dir",

				"-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",

				"-SworkingDir=hdfs://localhost:8020/users/joe/working_dir",

				"-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.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import eu.dnetlib.iis.core.TestsIOUtils;

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

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, HadoopContext context,

			Map<String, String> parameters) throws Exception {

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

		FileSystem fs = FileSystem.get(context.getConfiguration());

		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;

	}

	/**

	 * Loads job properties defined along with the workflow in job.properties file.

	 * @param resourcesOozieAppLocalPath

	 * @return properties read from predefined job.properties file, null when file not found

	 */

	private Properties loadPredefinedJobProperties(String resourcesOozieAppLocalPath) {

		if (resourcesOozieAppLocalPath!=null && 

				resourcesOozieAppLocalPath.contains(""+WORKFLOW_LOCATION_SEPARATOR)) {

			try {

				String parentWorkflowDir = resourcesOozieAppLocalPath.substring(0, 

						resourcesOozieAppLocalPath.lastIndexOf(WORKFLOW_LOCATION_SEPARATOR));

				Properties resultProperties = new Properties();

				resultProperties.load(Thread.currentThread().getContextClassLoader().getResourceAsStream(

						parentWorkflowDir + WORKFLOW_LOCATION_SEPARATOR + JOB_PROPERTIES_FILE_NAME));

				return resultProperties;

			} catch (Exception e) {

				log.warn("unable to load job.properties for workflow location " + 

						resourcesOozieAppLocalPath);

			}

		}

//		fallback

		return null;

	}

	/**

	 * Run the workflow

	 * @deprecated Replaced by {@link #runWorkflow(String)}

	 */

	@Deprecated

	public void runWorkflow(Path oozieAppPath, Path workingDir) 

					throws OozieClientException, IOException{

		runWorkflow(oozieAppPath, workingDir, 

				WorkflowConfiguration.defaultJobProperties);

	}

	/**

	 * Run the workflow

	 * @deprecated Replaced by {@link #runWorkflow(String, WorkflowConfiguration)}

	 */

	@Deprecated

	public void runWorkflow(Path oozieAppPath, Path workingDir,

			Properties jobProperties) 

					throws OozieClientException, IOException{

		WorkflowConfiguration config = 

				new WorkflowConfiguration().setJobProps(jobProperties);

		Properties props = createWorkflowConfiguration(

				createJobConf(), getTestUser(),	oozieAppPath, 

				workingDir, null, config.getJobProps());

		runWorkflowBasedOnConfiguration(props, config.getTimeoutInSeconds(),

				config.getExpectedFinishStatus());

	}

	private static Properties createWorkflowConfiguration(

			JobConf jobConf, String testUser, 

			Path oozieAppPath, Path workingDir, 

			Properties defaultJobProps, Properties runtimeJobProps) 

					throws IOException{		

		OozieTestsIOUtils.saveConfiguration(jobConf);

		final OozieClient wc = LocalOozie.getClient();

		Properties conf = wc.createConfiguration();

		/** Standard settings*/

		conf.setProperty(OozieClient.APP_PATH, oozieAppPath.toString());

		conf.setProperty(OozieClient.USER_NAME, testUser);

        conf.setProperty("jobTracker", jobConf.get("mapred.job.tracker"));

        conf.setProperty("nameNode", jobConf.get("fs.default.name"));

		conf.setProperty("queueName", "default");

		/** Custom settings */

		conf.setProperty("workingDir", workingDir.toString());

        conf.setProperty("minioozieTestRun", "true");

		if (defaultJobProps!=null) {

			conf.putAll(defaultJobProps);

		}

		if (runtimeJobProps!=null) {

			conf.putAll(runtimeJobProps);

		}

		return conf;

	}

	/**

	 * Execute workflow and checks if it ran correctly.

	 * @param configuration

     * @param timeoutInSeconds

	 * @throws OozieClientException

	 */

	private void runWorkflowBasedOnConfiguration(Properties configuration, 

			int timeoutInSeconds, WorkflowJob.Status expectedFinishStatus)

			throws OozieClientException{

		final OozieClient wc = LocalOozie.getClient();

		final String jobId = wc.submit(configuration);

		assertNotNull(jobId);

		WorkflowJob wf = wc.getJobInfo(jobId);

		assertNotNull(wf);

		assertEquals(WorkflowJob.Status.PREP, wf.getStatus());

		wc.start(jobId);

		waitFor(timeoutInSeconds * 1000, new Predicate() {

			public boolean evaluate() throws Exception {

				WorkflowJob wf = wc.getJobInfo(jobId);

				return (wf.getEndTime() != null);

			}

		});

    	wf = wc.getJobInfo(jobId);

        printErrors(wc, wf);

		assertNotNull(wf);

		assertEquals(expectedFinishStatus, wf.getStatus());

	}

    private void printErrors(OozieClient wc, WorkflowJob wf) {

        /** 

         * We can safely ignore the exceptions thrown during 

         * printing error messages 

         */

        try {

            for (WorkflowAction wa : wf.getActions()) {

                String errorMessage = wa.getErrorMessage() == null ? "" : wa.getErrorMessage();

                log.info("Workflow action {0} {1}", wa.getId(), errorMessage);

                if ("sub-workflow".equals(wa.getType()) && wa.getExternalId() != null) {

                    WorkflowJob subwf = wc.getJobInfo(wa.getExternalId());

                    printErrors(wc, subwf);

                }

            }

        } catch (OozieClientException ex) {}

    }

}

package eu.dnetlib.iis.core;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.IOException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Iterator;

import java.util.List;

import junit.framework.Assert;

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

/**

 * Utility class for tests dealing with IO

 * 

 * @author Mateusz Kobos

 * 

 */

public class TestsIOUtils {

	private final static int filesCompareBufferSize = 1024;

	private final static int dumpRecordsCountlimit = 100;

	/**

	 * {@code expected} and {@code actual} are equal if they have the same

	 * elements, possibly in different order.

	 */

	public static <T> void assertEqualSets(List<T> expected, Iterable<T> actual) {

		assertEqualSets(expected, actual.iterator());

	}

	/**

	 * {@code expected} and {@code actual} are equal if they have the same

	 * elements, possibly in different order.

	 */

	public static <T> void assertEqualSets(List<T> expected, Iterator<T> actual) {

		List<T> actualList = new ArrayList<T>();

		while(actual.hasNext()) {

			actualList.add(actual.next());

		}

		assertEqualSets(expected, actualList);

	}

	/**

	 * {@code expected} and {@code actual} are equal if they have the same

	 * elements, possibly in different order.

	 */

	public static <T> void assertEqualSets(

			List<T> expected, List<T> actual) {

		assertEqualSets(expected, actual, false);

	}

	/**

	 * {@code expected} and {@code actual} are equal if they have the same

	 * elements, possibly in different order.

	 * @param toPrettyJson if the sets are not equal, an error message with

	 * elements of both lists is printed. If this parameter is set to 

	 * {@code true}, the string representation of each element is treated as 

	 * JSON and printed in a pretty way.

	 */

	public static <T> void assertEqualSets(

			List<T> expected, List<T> actual, boolean toPrettyJson) {

		Assert.assertEquals(String.format("The number of the "+

				"expected elements (%s) is not equal to the number "+

				"of the actual elements (%s).\n\n%s",