How to test Spring batch configuration

Viewed 60

I like to test my class:

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.OraclePagingQueryProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;


@Configuration
@Slf4j
@RequiredArgsConstructor
@EnableScheduling
@EnableBatchProcessing
public class JobLDSExecutorConfiguration {
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final StepBuilderFactory stepBuilderFactory;
    private final JobBuilderFactory jobBuilderFactory;
    private final DataSource dataSource;
    private final JobRepository jobRepository;
    private static final int CHUNK_SIZE = 1;
    @Value("${kafka.rrss-correspondances-services-consommateur.topic}")
    private String topic;
    @Value("${task.thread.limit}")
    private int limteThread;
    @Value("${task.size}")
    private int size;
    @Value("${task.pageSize}")
    private int pageSize;

    private ChunkExecutionListener chunkListener() {
        return new ChunkExecutionListener();
    }

    @Bean
    public ItemReader<LdsOds> ldstJdbcItemReader() {
        return new JdbcPagingItemReaderBuilder<LdsOds>()
                .name("pagingItemReaderLDS")
                .fetchSize(size)
                .dataSource(dataSource)
                .pageSize(pageSize)
                .queryProvider(queryProvider())
                .rowMapper(new LdsOdsRowMapper(TypeDonnee.LDS))
                .build();
    }

    private OraclePagingQueryProvider queryProvider() {
        OraclePagingQueryProvider provider =
                new OraclePagingQueryProvider();
        provider.setSelectClause(SQL_SELECT_LDS);
        provider.setFromClause(SQL_FROM_LDS);
        Map<String, Order> sortKeys = new HashMap<>(1);
        sortKeys.put("SERVICE_DELIVERY_LOCATION_ID", Order.ASCENDING);
        provider.setSortKeys(sortKeys);
        return provider;
    }

    @Bean
    public ItemWriter<LdsOds> kafkaLDSItemWriter() {
        return items -> {
            for (LdsOds item : items) {
                kafkaTemplate.send(topic, item.getId().toString(), new ObjectMapper().writeValueAsString(item));
            }
        };
    }

    @Bean
    public TaskExecutor taskLDSExecutor() {
        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
        taskExecutor.setConcurrencyLimit(limteThread);
        return taskExecutor;
    }

    @Bean
    public Step loadLds() {
        return stepBuilderFactory.get("loadLds")
                .<LdsOds, LdsOds>chunk(CHUNK_SIZE)
                .reader(ldstJdbcItemReader())
                .writer(kafkaLDSItemWriter())
                .taskExecutor(taskLDSExecutor())
                .listener(chunkListener())
                .throttleLimit(limteThread)
                .build();
    }

    @Bean
    public JobExecutionListener jobLDSListener() {
        return new JobCompletionNotificationListener();
    }

    @Bean(name = "jobLDSExtraction")
    public Job jobLDSExtraction() {
        return jobBuilderFactory.get("jobLDSExtraction")
                .incrementer(new RunIdIncrementer())
                .listener(jobLDSListener())
                .start(loadLds())
                .build();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(15);
        taskExecutor.setMaxPoolSize(20);
        taskExecutor.setQueueCapacity(30);
        return taskExecutor;
    }

    @Bean
    public JobLauncher jobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setTaskExecutor(taskExecutor()); // Or below line
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }

    @Scheduled(fixedRate = 10000)
    public void executerMajIdentifiantLDS() {
        Map<String, JobParameter> confMap = new HashMap<>();
        confMap.put("time", new JobParameter(System.currentTimeMillis()));
        JobParameters jobParameters = new JobParameters(confMap);
        try {
            jobLauncher().run(jobLDSExtraction(), jobParameters);
        } catch (Exception ex) {
            log.error("Erreur pendant l'execution du jobLDSExtraction {}", ex.getMessage());
        }
    }
}

On my test

@Test
void testStep1() {
    JobExecution jobExecution = jobLauncherTestUtils.launchStep("loadLds");
     Assert.assertEquals(new ExitStatus("COMPLETED").getExitCode() ,jobExecution.getExitStatus().getExitCode());
}

But I have an error:

Error creating bean with name 'jobLDSExecutorConfiguration': Unsatisfied dependency expressed through constructor parameter 3; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'javax.sql.DataSource' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}

I would like to know if it's possible to MOCK or Inject datasource and how to test it.

0 Answers
Related