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
|
import pytest
import augur.io.vcf
from treetime.vcf_utils import read_vcf
@pytest.fixture
def mock_run_shell_command(mocker):
mocker.patch("augur.io.vcf.run_shell_command")
class TestVCF:
# The `read_vcf` functionality used to be in an augur module when these
# tests were originally written but we now use TreeTime's function of the
# same name. The tests remain here to protect against any unforeseen changes.
def test_read_vcf_compressed(self):
seq_keep = list(read_vcf("tests/data/tb_lee_2015.vcf.gz")['sequences'].keys())
assert len(seq_keep) == 150
assert seq_keep[149] == "G22733"
def test_read_vcf_uncompressed(self):
seq_keep = list(read_vcf("tests/data/tb_lee_2015.vcf")['sequences'].keys())
assert len(seq_keep) == 150
assert seq_keep[149] == "G22733"
def test_write_vcf_compressed_input(self, mock_run_shell_command):
augur.io.vcf.write_vcf(
"tests/data/tb_lee_2015.vcf.gz", "output_file.vcf.gz", []
)
augur.io.vcf.run_shell_command.assert_called_once_with(
"vcftools --gzvcf tests/data/tb_lee_2015.vcf.gz --recode --stdout | gzip -c > output_file.vcf.gz",
raise_errors=True,
)
def test_write_vcf_uncompressed_input(self, mock_run_shell_command):
augur.io.vcf.write_vcf(
"tests/data/tb_lee_2015.vcf", "output_file.vcf.gz", []
)
augur.io.vcf.run_shell_command.assert_called_once_with(
"vcftools --vcf tests/data/tb_lee_2015.vcf --recode --stdout | gzip -c > output_file.vcf.gz",
raise_errors=True,
)
def test_write_vcf_compressed_output(self, mock_run_shell_command):
augur.io.vcf.write_vcf(
"tests/data/tb_lee_2015.vcf", "output_file.vcf.gz", []
)
augur.io.vcf.run_shell_command.assert_called_once_with(
"vcftools --vcf tests/data/tb_lee_2015.vcf --recode --stdout | gzip -c > output_file.vcf.gz",
raise_errors=True,
)
def test_write_vcf_uncompressed_output(self, mock_run_shell_command):
augur.io.vcf.write_vcf(
"tests/data/tb_lee_2015.vcf", "output_file.vcf", []
)
augur.io.vcf.run_shell_command.assert_called_once_with(
"vcftools --vcf tests/data/tb_lee_2015.vcf --recode --stdout > output_file.vcf",
raise_errors=True,
)
def test_write_vcf_dropped_samples(self, mock_run_shell_command):
augur.io.vcf.write_vcf(
"tests/data/tb_lee_2015.vcf", "output_file.vcf", ["x", "y", "z"]
)
augur.io.vcf.run_shell_command.assert_called_once_with(
"vcftools --remove-indv x --remove-indv y --remove-indv z --vcf tests/data/tb_lee_2015.vcf --recode --stdout > output_file.vcf",
raise_errors=True,
)
|
