1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
import os
import shutil
import subprocess
import pytest
from bcbio import utils
from tests.conftest import install_cwl_test_files
@pytest.mark.docker
@pytest.mark.docker_multicore
def test_docker(install_test_files, data_dir, global_config):
"""Run an analysis with code and tools inside a docker container
Requires https://github.com/bcbio/bcbio-nextgen-vm
"""
fc_dir = os.path.join(data_dir, os.pardir, '100326_FC6107FAAXX')
run_config = os.path.join(data_dir, 'run_info-bam.yaml')
subprocess.check_call(['bcbio_vm.py', f'--datadir={data_dir}', 'run',
'--image=quay.io/bcbio/bcbio-vc', f'--systemconfig={global_config}',
f'--fcdir={fc_dir}', run_config])
@pytest.mark.docker
@pytest.mark.docker_ipython
def test_docker_ipython(install_test_files, data_dir, global_config):
"""Run an analysis with code and tools inside a docker container, driven via IPython
Requires https://github.com/bcbio/bcbio-nextgen-vm
"""
fc_dir = os.path.join(data_dir, os.pardir, '100326_FC6107FAAXX')
run_config = os.path.join(data_dir, 'run_info-bam.yaml')
subprocess.check_call(['bcbio_vm.py', f'--datadir={data_dir}', 'ipython',
f'--systemconfig={global_config}', f'--fcdir={fc_dir}', run_config,
'lsf', 'localrun'])
class TestCWL:
""" Run simple CWL workflows.
Requires https://github.com/bcbio/bcbio-nextgen-vm
"""
@pytest.mark.cwl
@pytest.mark.cwl_docker
@pytest.mark.cwl_docker_somatic
def test_cwl_docker_somatic_workflow(self):
with install_cwl_test_files() as workdir:
with utils.chdir(os.path.join(workdir, "somatic")):
subprocess.check_call(["bash", "./run_generate_cwl.sh"])
if os.path.exists("cromwell_work"):
shutil.rmtree("cromwell_work")
subprocess.check_call(["bcbio_vm.py", "cwlrun", "cromwell", "somatic-workflow"])
@pytest.mark.cwl
@pytest.mark.cwl_docker
@pytest.mark.cwl_docker_joint
def test_cwl_docker_joint_calling_workflow(self):
with install_cwl_test_files() as workdir:
with utils.chdir(os.path.join(workdir, "gvcf_joint")):
subprocess.check_call(["bash", "./run_generate_cwl.sh"])
if os.path.exists("cromwell_work"):
shutil.rmtree("cromwell_work")
subprocess.check_call(["bash", "./run_cromwell.sh"])
@pytest.mark.cwl
@pytest.mark.cwl_docker_rnaseq
def test_cwl_rnaseq(self, install_test_files):
with install_cwl_test_files() as work_dir:
with utils.chdir(os.path.join(work_dir, "rnaseq")):
if os.path.exists("cromwell_work"):
shutil.rmtree("cromwell_work")
subprocess.check_call(["bcbio_vm.py", "cwlrun", "cromwell", "rnaseq-workflow"])
@pytest.mark.cwl
@pytest.mark.cwl_local
@pytest.mark.install_required
def test_cwl_local_somatic_workflow(self, install_test_files):
with install_cwl_test_files() as workdir:
with utils.chdir(os.path.join(workdir, "somatic")):
subprocess.check_call(["bash", "./run_generate_cwl.sh"])
if os.path.exists("cwltool_work"):
shutil.rmtree("cwltool_work")
subprocess.check_call(["bash", "./run_cwltool.sh"])
@pytest.mark.cwl
@pytest.mark.cwl_arvados
def test_cwl_arvados_workflow(self, install_test_files):
"""Requires ARVADOS_API_HOST and ARVADOS_API_TOKEN set"""
if os.environ.get("ARVADOS_API_HOST") and os.environ.get("ARVADOS_API_TOKEN"):
with install_cwl_test_files() as workdir:
with utils.chdir(os.path.join(workdir, "arvados")):
subprocess.check_call(["bash", "./run_generate_cwl.sh"])
|
