MorphlineUtils { public static List<Record> executePipeline(Pipeline pipeline, Record inputRecord) { return executePipeline(pipeline, inputRecord, true); }  static Pipeline getPipeline(String morphlineFile, String morphlineId); static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord, boolean errorOnEmpty); @SuppressWarnings("serial") static FlatMapFunction<Row, Row> morphlineMapper(final String morphlineFile, final String morphlineId,
                                                          final StructType outputSchema, final boolean errorOnEmpty); static Row convertToRow(StructType schema, Record record);  }
@Test public void executePipeline( final @Mocked MorphlineUtils.Pipeline pipeline, final @Mocked Command morphline ) throws Exception { final Record inputRecord = new Record(); final Record outputRecord = new Record(); outputRecord.put("field1", "value1"); new Expectations() {{ morphline.process(inputRecord); result = true; pipeline.getCollector().getRecords(); result = Lists.newArrayList(outputRecord); }}; List<Record> outputList = MorphlineUtils.executePipeline(pipeline, inputRecord); assertEquals(Lists.newArrayList(outputRecord), outputList); }
@Test (expected = MorphlineRuntimeException.class) public void executePipelineProcessError( final @Mocked MorphlineUtils.Pipeline pipeline, final @Mocked Command morphline ) throws Exception { final Record inputRecord = new Record(); new Expectations() {{ morphline.process(inputRecord); result = false; }}; MorphlineUtils.executePipeline(pipeline, inputRecord); }
@Test (expected = MorphlineRuntimeException.class) public void executePipelineNoRecords( final @Mocked MorphlineUtils.Pipeline pipeline, final @Mocked Command morphline ) throws Exception { final Record inputRecord = new Record(); new Expectations() {{ morphline.process(inputRecord); result = true; pipeline.getCollector().getRecords(); result = Lists.newArrayList(); }}; MorphlineUtils.executePipeline(pipeline, inputRecord); }
@Test public void executePipelineNoRecordsNoError( final @Mocked MorphlineUtils.Pipeline pipeline, final @Mocked Command morphline ) throws Exception { final Record inputRecord = new Record(); new Expectations() {{ morphline.process(inputRecord); result = true; pipeline.getCollector().getRecords(); result = Lists.newArrayList(); }}; assertEquals("Invalid number of Rows returned", 0, MorphlineUtils.executePipeline(pipeline, inputRecord, false).size()); }