Skip to content

Commit

Permalink
review comments - update tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@nitin-ebi
nitin-ebi committed Apr 5, 2024
1 parent c9030c4 commit f6431f0
Show file tree
Hide file tree
Showing 3 changed files with 141 additions and 189 deletions.
139 changes: 34 additions & 105 deletions ...uk/ac/ebi/eva/accession/pipeline/runner/EvaAccessionJobLauncherCommandLineRunnerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;

import org.springframework.test.web.client.ExpectedCount;
import org.springframework.test.web.client.MockRestServiceServer;
import org.springframework.web.client.RestTemplate;
Expand All @@ -47,38 +46,35 @@
import uk.ac.ebi.eva.accession.pipeline.parameters.InputParameters;
import uk.ac.ebi.eva.accession.pipeline.test.BatchTestConfiguration;
import uk.ac.ebi.eva.commons.batch.io.VcfReader;
import uk.ac.ebi.eva.commons.core.utils.FileUtils;
import uk.ac.ebi.eva.metrics.count.CountServiceParameters;

import javax.sql.DataSource;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.util.Arrays;
import java.util.Collections;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.method;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
import static org.springframework.test.web.client.response.MockRestResponseCreators.withStatus;
import static uk.ac.ebi.eva.accession.pipeline.configuration.BeanNames.CREATE_SUBSNP_ACCESSION_JOB;
import static uk.ac.ebi.eva.accession.pipeline.configuration.BeanNames.CREATE_SUBSNP_ACCESSION_STEP;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.deleteTemporaryContigAndVariantFiles;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.getOriginalVcfContent;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.injectErrorIntoTempVcf;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.remediateTempVcfError;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.useOriginalVcfFile;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.useTempVcfFile;
import static uk.ac.ebi.eva.accession.pipeline.runner.runnerUtil.writeToTempVCFFile;

@RunWith(SpringRunner.class)
@ContextConfiguration(classes={BatchTestConfiguration.class})
@ContextConfiguration(classes = {BatchTestConfiguration.class})
@TestPropertySource("classpath:accession-pipeline-test.properties")
@SpringBatchTest
public class EvaAccessionJobLauncherCommandLineRunnerTest {
Expand Down Expand Up @@ -146,13 +142,13 @@ public void setUp() throws Exception {
originalVcfInputFilePath = inputParameters.getVcf();
originalVcfOutputFilePath = inputParameters.getOutputVcf();
originalVcfContent = getOriginalVcfContent(originalVcfInputFilePath);
writeToTempVCFFile(originalVcfContent);
writeToTempVCFFile(originalVcfContent, tempVcfInputFileToTestFailingJobs);
originalInputParametersCaptured = true;
}
jobRepositoryTestUtils = new JobRepositoryTestUtils(jobRepository, datasource);
runner.setJobNames(CREATE_SUBSNP_ACCESSION_JOB);
deleteTemporaryContigAndVariantFiles();
useOriginalVcfFile();
deleteTemporaryContigAndVariantFiles(inputParameters, tempVcfOutputDir);
useOriginalVcfFile(inputParameters, originalVcfInputFilePath, vcfReader);

mockServer = MockRestServiceServer.createServer(restTemplate);
mockServer.expect(ExpectedCount.manyTimes(), requestTo(new URI(countServiceParameters.getUrl() + URL_PATH_SAVE_COUNT)))
Expand Down Expand Up @@ -190,24 +186,23 @@ public void restartCompletedJobThatIsAlreadyInTheRepository() throws Exception {
runner.run();
assertEquals(EvaAccessionJobLauncherCommandLineRunner.EXIT_WITHOUT_ERRORS, runner.getExitCode());

deleteTemporaryContigAndVariantFiles();
deleteTemporaryContigAndVariantFiles(inputParameters, tempVcfOutputDir);

inputParameters.setForceRestart(true);
runner.run();
assertEquals(EvaAccessionJobLauncherCommandLineRunner.EXIT_WITHOUT_ERRORS, runner.getExitCode());
}



private JobInstance runJobAandCheckResults() throws Exception {
runner.run();
assertEquals(EvaAccessionJobLauncherCommandLineRunner.EXIT_WITH_ERRORS, runner.getExitCode());
JobInstance currentJobInstance = CommandLineRunnerUtils.getLastJobExecution(CREATE_SUBSNP_ACCESSION_JOB,
jobExplorer,
inputParameters.toJobParameters())
.getJobInstance();
jobExplorer,
inputParameters.toJobParameters())
.getJobInstance();
StepExecution stepExecution = jobRepository.getLastStepExecution(currentJobInstance,
CREATE_SUBSNP_ACCESSION_STEP);
CREATE_SUBSNP_ACCESSION_STEP);
//Ensure that only the first batch was written (batch size is 5 and error was at line#9)
assertEquals(inputParameters.getChunkSize(), stepExecution.getWriteCount());

Expand Down Expand Up @@ -235,18 +230,19 @@ public void resumeFailingJobFromCorrectChunk() throws Exception {
// C is a job with the same parameters as A run after VCF fault remediation (as part of the
// runTestWithFaultInjection method), therefore should resume A and succeed.

useTempVcfFile();
injectErrorIntoTempVcf();
useTempVcfFile(inputParameters, tempVcfInputFileToTestFailingJobs, vcfReader);
String modifiedVcfContent = originalVcfContent.replace("76852", "76852jibberish");
injectErrorIntoTempVcf(modifiedVcfContent, tempVcfInputFileToTestFailingJobs);
JobInstance failingJobInstance = runJobAandCheckResults();

runJobBAndCheckResults();

remediateTempVcfError();
remediateTempVcfError(originalVcfContent, tempVcfInputFileToTestFailingJobs);
runJobCAndCheckResumption(failingJobInstance);
}

private void runJobBAndCheckResults() throws Exception {
useOriginalVcfFile();
useOriginalVcfFile(inputParameters, originalVcfInputFilePath, vcfReader);
// Back up contig and variant files (left behind by previous unsuccessful job A) to temp folder
// so as to not interfere with this job's execution which uses the original VCF file
backUpContigAndVariantFilesToTempFolder();
Expand All @@ -255,18 +251,18 @@ private void runJobBAndCheckResults() throws Exception {
assertEquals(EvaAccessionJobLauncherCommandLineRunner.EXIT_WITHOUT_ERRORS, runner.getExitCode());

//Restore state so that Job C can continue running after fault remediation
useTempVcfFile();
useTempVcfFile(inputParameters, tempVcfInputFileToTestFailingJobs, vcfReader);
restoreContigAndVariantFilesFromTempFolder();
}

private void runJobCAndCheckResumption(JobInstance failingJobInstance) throws Exception {
runner.run();
JobInstance currentJobInstance = CommandLineRunnerUtils.getLastJobExecution(CREATE_SUBSNP_ACCESSION_JOB,
jobExplorer,
inputParameters.toJobParameters())
.getJobInstance();
jobExplorer,
inputParameters.toJobParameters())
.getJobInstance();
StepExecution stepExecution = jobRepository.getLastStepExecution(currentJobInstance,
CREATE_SUBSNP_ACCESSION_STEP);
CREATE_SUBSNP_ACCESSION_STEP);
// Did we resume the previous failed job instance?
assertEquals(failingJobInstance.getInstanceId(), currentJobInstance.getInstanceId());

Expand All @@ -275,105 +271,38 @@ private void runJobCAndCheckResumption(JobInstance failingJobInstance) throws Ex
// Test resumption point - did we pick up where we left off?
// Ensure all the batches other than the first batch were processed
assertEquals(numberOfLinesInVcf - inputParameters.getChunkSize() - numberOfNonVariants,
stepExecution.getWriteCount());
stepExecution.getWriteCount());
assertEquals(EvaAccessionJobLauncherCommandLineRunner.EXIT_WITHOUT_ERRORS, runner.getExitCode());
}

private void injectErrorIntoTempVcf() throws Exception {
String modifiedVcfContent = originalVcfContent.replace("76852", "76852jibberish");
// Inject error in the VCF file to cause processing to stop at variant#9
writeToTempVCFFile(modifiedVcfContent);
}

private void remediateTempVcfError() throws Exception {
writeToTempVCFFile(originalVcfContent);
}

private void useOriginalVcfFile() throws Exception {
inputParameters.setVcf(originalVcfInputFilePath);
vcfReader.setResource(FileUtils.getResource(new File(originalVcfInputFilePath)));
}

private void useTempVcfFile() throws Exception {
// The following does not actually change the wiring of the vcfReader since the wiring happens before the tests
// This setVcf is only to facilitate identifying jobs in the job repo by parameter
// (those that use original vs temp VCF)
inputParameters.setVcf(tempVcfInputFileToTestFailingJobs.getAbsolutePath());
/*
* Change the auto-wired VCF for VCFReader at runtime
* Rationale:
* 1) Why not use two test configurations, one for a VCF that fails validation and another for a VCF
* that won't and test resumption?
* Beginning Spring Boot 2, job resumption can only happen when input parameters to the restarted job
* is the same as the failed job.
* Therefore, a test to check resumption cannot have two different config files with different
* parameters.vcf.
* This test therefore creates a dynamic VCF and injects errors at runtime to the VCF thus preserving
* the VCF parameter but changing the VCF content.
* 2) Why not artificially inject a VcfReader exception?
* This will preclude us from verifying job resumption from a precise line in the VCF.
*/
vcfReader.setResource(FileUtils.getResource(tempVcfInputFileToTestFailingJobs));
}

private void backUpContigAndVariantFilesToTempFolder() {
moveFile(Paths.get(originalVcfOutputFilePath + ".contigs"),
Paths.get(tempVcfOutputDir + "/accession-output.vcf.contigs"));
Paths.get(tempVcfOutputDir + "/accession-output.vcf.contigs"));
moveFile(Paths.get(originalVcfOutputFilePath + ".variants"),
Paths.get(tempVcfOutputDir + "/accession-output.vcf.variants"));
Paths.get(tempVcfOutputDir + "/accession-output.vcf.variants"));
}

private void restoreContigAndVariantFilesFromTempFolder() {
moveFile(Paths.get(tempVcfOutputDir + "/accession-output.vcf.contigs"),
Paths.get(Paths.get(originalVcfOutputFilePath).getParent() + "/accession-output.vcf.contigs"));
Paths.get(Paths.get(originalVcfOutputFilePath).getParent() + "/accession-output.vcf.contigs"));
moveFile(Paths.get(tempVcfOutputDir + "/accession-output.vcf.variants"),
Paths.get(Paths.get(originalVcfOutputFilePath).getParent() + "/accession-output.vcf.variants"));
Paths.get(Paths.get(originalVcfOutputFilePath).getParent() + "/accession-output.vcf.variants"));
}

private void moveFile(Path source, Path destination) {
try {
Files.move(source, destination, StandardCopyOption.REPLACE_EXISTING);
}
catch (Exception ex) {
} catch (Exception ex) {
if (!(ex instanceof NoSuchFileException)) {
throw new RuntimeException(ex);
}
}
}

private void writeToTempVCFFile(String modifiedVCFContent) throws IOException {
FileOutputStream outputStream = new FileOutputStream(tempVcfInputFileToTestFailingJobs.getAbsolutePath());
GZIPOutputStream gzipOutputStream = new GZIPOutputStream(outputStream);
gzipOutputStream.write(modifiedVCFContent.getBytes(StandardCharsets.UTF_8));
gzipOutputStream.close();
}

private String getOriginalVcfContent(String inputVcfPath) throws Exception {
StringBuilder originalVCFContent = new StringBuilder();

GZIPInputStream gzipInputStream = new GZIPInputStream(new FileInputStream(inputVcfPath));
BufferedReader reader = new BufferedReader(new InputStreamReader(gzipInputStream));

String read;
while ((read = reader.readLine()) != null) {
originalVCFContent.append(read).append(System.lineSeparator());
}
return originalVCFContent.toString();
}

private void deleteTemporaryContigAndVariantFiles() throws Exception {
Files.deleteIfExists(Paths.get(inputParameters.getOutputVcf()));
Files.deleteIfExists(Paths.get(inputParameters.getOutputVcf() + ".variants"));
Files.deleteIfExists(Paths.get(inputParameters.getOutputVcf() + ".contigs"));
Files.deleteIfExists(
Paths.get(tempVcfOutputDir + "/accession-output.vcf.variants"));
Files.deleteIfExists(
Paths.get(tempVcfOutputDir + "/accession-output.vcf.contigs"));
}

private int getNumberOfLinesInVcfString(String vcfString) {
return (int) Arrays.stream(vcfString.split(System.lineSeparator()))
.filter(line -> !line.startsWith("#"))
.count();
.filter(line -> !line.startsWith("#"))
.count();
}
}
Loading

0 comments on commit f6431f0

Please sign in to comment.