7
$\begingroup$

I've been playing with snakemake, which is really powerful, however, I'd like to create a 'unit test' for the whole pipeline. e.g. a way to call the pipeline that tests that all the rules work together, all the steps complete, etc.

I assumed I could find some guides about how to do this, but I don't find any clear information about how to tests pipelines.

Any advice or suggestions would be welcome.

$\endgroup$
1

2 Answers 2

5
$\begingroup$

You can run CI on a snakemake workflow when provoding a minimal example dataset. Maybe an example from the snakemake-workflows repo will be helpful: https://github.com/snakemake-workflows/dna-seq-varlociraptor

$\endgroup$
2
$\begingroup$

This is what I do when I want to check whether a program runs from start to finish and produces the expected output on test data. Here I used it to test a C program but the idea can be applied also to a snakemake pipeline as in your case.

The idea is pretty simple really: use python's subprocess module to run the pipeline then check the results using the unittest module (or your favourite unit test package).

For example, save this in test.py:

#!/usr/bin/env python3

import unittest
import os
import subprocess as sp

class Tester(unittest.TestCase):

    def testPipelineMakesStuff(self):
        # Run the pipeline
        p= sp.Popen('snakemake [some parameters]', shell=True, stdout= sp.PIPE, stderr= sp.PIPE)
        
        # Optionally, get stdout and stderr
        stdout, stderr= p.communicate()

        # Check exits code and other expected output            
        self.assertEqual(0, p.returncode)
        self.assertTrue(os.path.isfile('some.pipeline.output.txt'))

        ...Maybe read some.pipeline.output.txt and check it looks ok

    def testPipelineReturnsWithError(self):
        p= sp.Popen('snakemake [some parameters]', shell=True, stdout= sp.PIPE, stderr= sp.PIPE)
        stdout, stderr= p.communicate()
        self.assertTrue(p.returncode != 0)
        self.assertTrue('Something went wrong' in stderr.decode())
    
    def ...more test cases...

if __name__ == '__main__':
    unittest.main()

Run this script as:

 python test.py 

and check all tests pass ok. Of course, you can embed python test.py in travis or other continuous integration system as I've done here.

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.