@Test public void testProcess() throws Exception { List<Stage> mockedStages = createFakeStages(); Map<String, String> mockedProperties = createFakeBatchProperties(); File testZip = IngestionTest.getFile("ctl_tmp/test.zip"); NewBatchJob job = new NewBatchJob(BATCHJOBID, testZip.getAbsolutePath(), "finished", 29, mockedProperties, mockedStages, null); job.setBatchProperties(mockedProperties); job.setTenantId(tenantId); ResourceEntry entry = new ResourceEntry(); entry.setResourceName(testZip.getAbsolutePath()); entry.setResourceFormat(FileFormat.ZIP_FILE.getCode()); job.addResourceEntry(entry); WorkNote workNote = Mockito.mock(WorkNote.class); Mockito.when(workNote.getBatchJobId()).thenReturn(BATCHJOBID); Exchange exchange = new DefaultExchange(new DefaultCamelContext()); exchange.getIn().setBody(workNote); Mockito.when(mockedBatchJobDAO.findBatchJobById(Matchers.eq(BATCHJOBID))).thenReturn(job); controlFilePreProcessor.setBatchJobDAO(mockedBatchJobDAO); controlFilePreProcessor.setTenantDA(mockedTenantDA); controlFilePreProcessor.process(exchange); Assert.assertEquals(2, job.getResourceEntries().size()); Assert.assertEquals(29, job.getTotalFiles()); }
@Override public void process(Exchange exchange) throws Exception { processUsingNewBatchJob(exchange); }
ControlFilePreProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { processUsingNewBatchJob(exchange); } }
ControlFilePreProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { processUsingNewBatchJob(exchange); }  }
ControlFilePreProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { processUsingNewBatchJob(exchange); }  @Override void process(Exchange exchange); Set<String> getShardCollections(); void setShardCollections(Set<String> shardCollections); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setTenantDA(TenantDA tenantDA); }
ControlFilePreProcessor implements Processor { @Override public void process(Exchange exchange) throws Exception { processUsingNewBatchJob(exchange); }  @Override void process(Exchange exchange); Set<String> getShardCollections(); void setShardCollections(Set<String> shardCollections); void setBatchJobDAO(BatchJobDAO batchJobDAO); void setTenantDA(TenantDA tenantDA); static final BatchJobStageType BATCH_JOB_STAGE; static final String INDEX_SCRIPT; }