comparison tests/test_shm_csr.py @ 6:ea9d5fc4c001 draft default tip

"planemo upload commit 9ada186a78831ca2618ec817a23a77de6adf1a5d"
author rhpvorderman
date Wed, 22 Dec 2021 11:29:16 +0000
parents 495a521cf9f2
children
comparison
equal deleted inserted replaced
5:495a521cf9f2 6:ea9d5fc4c001
22 import shutil 22 import shutil
23 import subprocess 23 import subprocess
24 import sys 24 import sys
25 import tempfile 25 import tempfile
26 from pathlib import Path 26 from pathlib import Path
27 from xml.etree import ElementTree
28 from xml.etree.ElementTree import Element
27 29
28 import pytest 30 import pytest
29 31
30 GIT_ROOT = str(Path(__file__).parent.parent.absolute()) 32 GIT_ROOT = Path(__file__).parent.parent.absolute()
31 TEST_DIR = Path(__file__).parent 33 TEST_DIR = Path(__file__).parent
32 TEST_DATA_DIR = TEST_DIR / "data" 34 TEST_DATA_DIR = TEST_DIR / "data"
33 VALIDATION_DATA_DIR = TEST_DIR / "validation_data" 35 VALIDATION_DATA_DIR = TEST_DIR / "validation_data"
34 CONTROL_NWK377_PB_IGHC_MID1_40nt_2 = TEST_DATA_DIR / "CONTROL_NWK377_PB_IGHC_MID1_40nt_2.txz" 36 CONTROL_NWK377_PB_IGHC_MID1_40nt_2 = TEST_DATA_DIR / "CONTROL_NWK377_PB_IGHC_MID1_40nt_2.txz"
35 37
36 38
39 def get_container():
40 tool = ElementTree.parse(GIT_ROOT / "shm_csr.xml").getroot()
41 requirements: Element = tool.find("requirements")
42 container = requirements.find("container")
43 return container.text
44
45
37 @pytest.fixture(scope="module") 46 @pytest.fixture(scope="module")
38 def shm_csr_result(): 47 def shm_csr_result():
39 docker_container = "quay.io/rhpvorderman/mulled-v2-f7d31c9d7424063a492fc0e5ecbf89bc757c0107:2b50bdd4d8c1fefc6ec24b0753fad0dcecec843b-0"
40 temp_dir = Path(tempfile.mkdtemp()) 48 temp_dir = Path(tempfile.mkdtemp())
41 tool_dir = temp_dir / "shm_csr" 49 tool_dir = temp_dir / "shm_csr"
42 shutil.copytree(GIT_ROOT, tool_dir) 50 shutil.copytree(GIT_ROOT, tool_dir)
43 working_dir = temp_dir / "working" 51 working_dir = temp_dir / "working"
44 working_dir.mkdir(parents=True) 52 working_dir.mkdir(parents=True)
84 class_filter, 92 class_filter,
85 empty_region_filter, 93 empty_region_filter,
86 fast 94 fast
87 ] 95 ]
88 docker_cmd = ["docker", "run", "-v", f"{temp_dir}:{temp_dir}", 96 docker_cmd = ["docker", "run", "-v", f"{temp_dir}:{temp_dir}",
97 "--rm", # Remove container after running
89 "-v", f"{input}:{input}", 98 "-v", f"{input}:{input}",
90 "-w", str(working_dir), 99 "-w", str(working_dir),
91 docker_container] + cmd 100 # Run as current user which allows deletion of files.
101 # It also mitigates some security considerations
102 "-u", f"{os.getuid()}:{os.getgid()}",
103 # Run with default seccomp profile to mitigate mitigation slowdown
104 # http://mamememo.blogspot.com/2020/05/cpu-intensive-rubypython-code-runs.html
105 # This knocks down test runtime from 8 to 6 minutes.
106 "--security-opt", "seccomp=unconfined",
107 get_container()] + cmd
92 with open(temp_dir / "stderr", "wt") as stderr_file: 108 with open(temp_dir / "stderr", "wt") as stderr_file:
93 with open(temp_dir / "stdout", "wt") as stdout_file: 109 with open(temp_dir / "stdout", "wt") as stdout_file:
94 subprocess.run(docker_cmd, cwd=working_dir, stdout=stdout_file, 110 subprocess.run(docker_cmd, cwd=working_dir, stdout=stdout_file,
95 stderr=stderr_file, check=True) 111 stderr=stderr_file, check=True)
96 yield Path(out_files_path) 112 yield Path(out_files_path)
97 #shutil.rmtree(temp_dir)
98 113
99 114
100 def test_check_output(shm_csr_result): 115 def test_check_output(shm_csr_result):
101 assert shm_csr_result.exists() 116 assert shm_csr_result.exists()
102 117
103 118
104 @pytest.mark.parametrize("filename", os.listdir(VALIDATION_DATA_DIR)) 119 @pytest.mark.parametrize("filename", os.listdir(VALIDATION_DATA_DIR))
105 def test_results_match_validation(shm_csr_result, filename): 120 def test_results_match_validation(shm_csr_result, filename):
106 if filename == "shm_overview.txt":
107 # TODO: Fix errors in shm_overview.
108 return
109 with open(Path(shm_csr_result, filename)) as result_h: 121 with open(Path(shm_csr_result, filename)) as result_h:
110 with open(Path(VALIDATION_DATA_DIR, filename)) as validate_h: 122 with open(Path(VALIDATION_DATA_DIR, filename)) as validate_h:
111 for line in result_h: 123 for line in result_h:
112 assert line == validate_h.readline() 124 # Skip two faulty lines in shm_overview.
125 # TODO: Fix the issue.
126 validation_line = validate_h.readline()
127 if filename == "shm_overview.txt":
128 if (line.startswith("RGYW (%)") or
129 line.startswith("WRCY (%)")):
130 continue
131 assert line == validation_line
113 132
114 133
115 def test_nt_overview(shm_csr_result): 134 def test_nt_overview(shm_csr_result):
116 with open(Path(shm_csr_result, "sequence_overview", "ntoverview.txt") 135 with open(Path(shm_csr_result, "sequence_overview", "ntoverview.txt")
117 ) as result_h: 136 ) as result_h: