Mercurial > repos > rhpvorderman > shm_csr
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: |
