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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
"""Tests that run the program in a subprocess"""
import subprocess
import sys
import os
import pytest
from utils import datapath, assert_files_equal, cutpath
def test_run_cutadapt_process():
subprocess.check_call(["cutadapt", "--version"])
def test_run_as_module():
"""Check that "python3 -m cutadapt ..." works"""
from cutadapt import __version__
with subprocess.Popen(
[sys.executable, "-m", "cutadapt", "--version"], stdout=subprocess.PIPE
) as py:
assert py.communicate()[0].decode().strip() == __version__
@pytest.mark.skipif(sys.platform == "win32", reason="Perhaps this can be fixed")
def test_standard_input_pipe(tmp_path, cores):
"""Read FASTQ from standard input"""
out_path = os.fspath(tmp_path / "out.fastq")
in_path = datapath("small.fastq")
# Simulate that no file name is available for stdin
with subprocess.Popen(["cat", in_path], stdout=subprocess.PIPE) as cat:
with subprocess.Popen(
[
sys.executable,
"-m",
"cutadapt",
"--cores",
str(cores),
"-a",
"TTAGACATATCTCCGTCG",
"-o",
out_path,
"-",
],
stdin=cat.stdout,
) as py:
_ = py.communicate()
cat.stdout.close()
_ = py.communicate()[0]
assert_files_equal(cutpath("small.fastq"), out_path)
def test_standard_output(tmp_path, cores):
"""Write FASTQ to standard output (not using --output/-o option)"""
out_path = os.fspath(tmp_path / "out.fastq")
with open(out_path, "w") as out_file:
py = subprocess.Popen(
[
sys.executable,
"-m",
"cutadapt",
"--cores",
str(cores),
"-a",
"TTAGACATATCTCCGTCG",
datapath("small.fastq"),
],
stdout=out_file,
)
_ = py.communicate()
assert_files_equal(cutpath("small.fastq"), out_path)
def test_errors_are_printed_to_stderr(tmp_path):
out_path = os.fspath(tmp_path / "out.fastq")
py = subprocess.Popen(
[
sys.executable,
"-m",
"cutadapt",
"-o",
out_path,
tmp_path / "does-not-exist.fastq",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout_bytes, stderr_bytes = py.communicate()
assert b"No such file or directory" in stderr_bytes
assert b"No such file or directory" not in stdout_bytes
def test_explicit_standard_output(tmp_path, cores):
"""Write FASTQ to standard output (using "-o -")"""
out_path = os.fspath(tmp_path / "out.fastq")
with open(out_path, "w") as out_file:
py = subprocess.Popen(
[
sys.executable,
"-m",
"cutadapt",
"-o",
"-",
"--cores",
str(cores),
"-a",
"TTAGACATATCTCCGTCG",
datapath("small.fastq"),
],
stdout=out_file,
)
_ = py.communicate()
assert_files_equal(cutpath("small.fastq"), out_path)
def test_force_fasta_output(tmp_path, cores):
"""Write FASTA to standard output even on FASTQ input"""
out_path = os.fspath(tmp_path / "out.fasta")
with open(out_path, "w") as out_file:
py = subprocess.Popen(
[
sys.executable,
"-m",
"cutadapt",
"--fasta",
"-o",
"-",
"--cores",
str(cores),
"-a",
"TTAGACATATCTCCGTCG",
datapath("small.fastq"),
],
stdout=out_file,
)
_ = py.communicate()
assert_files_equal(cutpath("small.fasta"), out_path)
@pytest.mark.skipif(sys.platform == "win32", reason="Maybe this can be made to work")
def test_non_utf8_locale():
subprocess.check_call(
[sys.executable, "-m", "cutadapt", "-o", os.devnull, datapath("small.fastq")],
env={"LC_CTYPE": "C"},
)
|
