@Test public void testProcess() throws Exception { List<ResourceEntry> mockedResourceEntries = createFakeResourceEntries(); List<Stage> mockedStages = createFakeStages(); Map<String, String> mockedProperties = createFakeBatchProperties(); NewBatchJob mockedJob = new NewBatchJob(BATCHJOBID, "192.168.59.11", "finished", 1, mockedProperties, mockedStages, mockedResourceEntries); mockedJob.setTopLevelSourceId(TEMP_DIR); Iterable<Error> fakeErrorIterable = createFakeErrorIterable(); RangedWorkNote workNote = RangedWorkNote.createSimpleWorkNote(BATCHJOBID); List<Stage> mockDeltaStage = new LinkedList<Stage>(); List<Metrics> mockDeltaMetrics = new LinkedList<Metrics>(); List<Stage> mockPersistenceStages = new LinkedList<Stage>(); List<Metrics> mockPersistenceMetrics = new LinkedList<Metrics>(); Metrics duplicateCountMetric = new Metrics(RESOURCEID, RECORDS_CONSIDERED, RECORDS_FAILED); HashMap<String, Long> dupMap = new HashMap<String, Long>(); dupMap.put(DUP_ENTITY, DUP_COUNT); duplicateCountMetric.setDuplicateCounts(dupMap); mockDeltaMetrics.add(duplicateCountMetric); mockDeltaStage.add(new Stage(BatchJobStageType.DELTA_PROCESSOR.getName(), "Drop records that are duplicates", "finished", new Date(), new Date(), mockDeltaMetrics)); mockPersistenceMetrics.add(new Metrics(RESOURCEID, RECORDS_CONSIDERED, RECORDS_FAILED,RECORDS_DELETED)); mockPersistenceStages.add(new Stage(BatchJobStageType.PERSISTENCE_PROCESSOR.getName(), "Persists records to the sli database", "finished", new Date(), new Date(), mockPersistenceMetrics)); Mockito.when(mockedBatchJobDAO.findBatchJobById(Matchers.eq(BATCHJOBID))).thenReturn(mockedJob); Mockito.when(mockedBatchJobDAO.getBatchJobStages(Matchers.eq(BATCHJOBID), Matchers.eq(BatchJobStageType.DELTA_PROCESSOR))).thenReturn(mockDeltaStage); Mockito.when(mockedBatchJobDAO.getBatchJobStages(Matchers.eq(BATCHJOBID), Matchers.eq(BatchJobStageType.PERSISTENCE_PROCESSOR))).thenReturn(mockPersistenceStages); Mockito.when( mockedBatchJobDAO.getBatchJobErrors(Matchers.eq(BATCHJOBID), Matchers.eq(RESOURCEID), Matchers.eq(FaultType.TYPE_ERROR), Matchers.anyInt())).thenReturn(fakeErrorIterable); Mockito.when( mockedBatchJobDAO.getBatchJobErrors(Matchers.eq(BATCHJOBID), (String) Matchers.isNull(), Matchers.eq(FaultType.TYPE_ERROR), Matchers.anyInt())).thenReturn(fakeErrorIterable); Mockito.when( mockedBatchJobDAO.getBatchJobErrors(Matchers.eq(BATCHJOBID), Matchers.eq(RESOURCEID), Matchers.eq(FaultType.TYPE_WARNING), Matchers.anyInt())).thenReturn(fakeErrorIterable); Exchange exchange = new DefaultExchange(new DefaultCamelContext()); exchange.getIn().setBody(workNote, RangedWorkNote.class); jobReportingProcessor.setBatchJobDAO(mockedBatchJobDAO); jobReportingProcessor.setTenantDA(mockedTenantDA); jobReportingProcessor.process(exchange); FileReader fr = new FileReader(TEMP_DIR + OUTFILE); BufferedReader br = new BufferedReader(fr); assertTrue(br.readLine().contains("jobId: " + BATCHJOBID)); assertTrue(br.readLine().contains( "[file] " + RESOURCEID + " (" + FileFormat.EDFI_XML.getCode() + "/" + FileType.XML_STUDENT_PARENT_ASSOCIATION.getName() + ")")); assertTrue(br.readLine().contains("[file] " + RESOURCEID + " records considered for processing: " + RECORDS_CONSIDERED)); assertTrue(br.readLine().contains("[file] " + RESOURCEID + " records ingested successfully: " + RECORDS_PASSED)); assertTrue(br.readLine().contains("[file] " + RESOURCEID + " records deleted successfully: " + RECORDS_DELETED)); assertTrue(br.readLine().contains("[file] " + RESOURCEID + " records failed processing: " + RECORDS_FAILED)); assertTrue(br.readLine().contains("[file] " + RESOURCEID + " records not considered for processing: " + 0)); assertTrue(br.readLine().contains("[configProperty] purge: false")); assertTrue(br.readLine().contains(RESOURCEID + " " + DUP_ENTITY + " " + DUP_COUNT + " deltas!")); assertTrue(br.readLine().contains("Not all records were processed completely due to errors.")); assertTrue(br.readLine().contains("Processed " + RECORDS_CONSIDERED + " records.")); String errorFileName = getErrorFileName(); fr = new FileReader(TEMP_DIR + errorFileName); br = new BufferedReader(fr); assertTrue(br.readLine().contains("ERROR " + ERRORDETAIL)); assertTrue(br.readLine().contains("ERROR " + NULLERRORDETAIL)); fr.close(); }
@Override public void process(Exchange exchange) { WorkNote workNote = exchange.getIn().getBody(WorkNote.class); if (workNote == null || workNote.getBatchJobId() == null) { missingBatchJobIdError(exchange); } else { processJobReporting(exchange, workNote); } }
JobReportingProcessor implements Processor { @Override public void process(Exchange exchange) { WorkNote workNote = exchange.getIn().getBody(WorkNote.class); if (workNote == null || workNote.getBatchJobId() == null) { missingBatchJobIdError(exchange); } else { processJobReporting(exchange, workNote); } } }
JobReportingProcessor implements Processor { @Override public void process(Exchange exchange) { WorkNote workNote = exchange.getIn().getBody(WorkNote.class); if (workNote == null || workNote.getBatchJobId() == null) { missingBatchJobIdError(exchange); } else { processJobReporting(exchange, workNote); } }  }
JobReportingProcessor implements Processor { @Override public void process(Exchange exchange) { WorkNote workNote = exchange.getIn().getBody(WorkNote.class); if (workNote == null || workNote.getBatchJobId() == null) { missingBatchJobIdError(exchange); } else { processJobReporting(exchange, workNote); } }  @Override void process(Exchange exchange); File getLogFile(NewBatchJob job); File createFile(NewBatchJob job, String fileName); void setCommandTopicUri(String commandTopicUri); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setTenantDA(TenantDA tenantDA); }
JobReportingProcessor implements Processor { @Override public void process(Exchange exchange) { WorkNote workNote = exchange.getIn().getBody(WorkNote.class); if (workNote == null || workNote.getBatchJobId() == null) { missingBatchJobIdError(exchange); } else { processJobReporting(exchange, workNote); } }  @Override void process(Exchange exchange); File getLogFile(NewBatchJob job); File createFile(NewBatchJob job, String fileName); void setCommandTopicUri(String commandTopicUri); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setTenantDA(TenantDA tenantDA); static final BatchJobStageType BATCH_JOB_STAGE; static final String JOB_STAGE_RESOURCE_ID; static final String ORCHESTRATION_STAGES_NAME; static final String ORCHESTRATION_STAGES_DESC; static final String ERROR_FILE_TYPE; static final String WARNING_FILE_TYPE; }