Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _run(params, inpath1, inpath2=None, expected1=None, expected2=None, cores=1):
assert not (inpath1 and inpath2 and expected1 and expected2)
assert not (expected2 and not expected1)
assert not (inpath2 and not inpath1)
if type(params) is str:
params = params.split()
params += ["--interleaved", "--cores", str(cores), "--buffer-size=512"]
tmp1 = str(tmpdir.join("out1-" + expected1))
params += ["-o", tmp1]
paths = [datapath(inpath1)]
if inpath2:
paths += [datapath(inpath2)]
if expected2:
tmp2 = str(tmpdir.join("out2-" + expected2))
params += ["-p", tmp2]
assert main(params + paths) is None
assert_files_equal(cutpath(expected2), tmp2)
else:
assert main(params + paths) is None
assert_files_equal(cutpath(expected1), tmp1)
def test_interleaved_neither_nor(tmpdir):
"""Option --interleaved used, but pairs of files given for input and output"""
p1 = str(tmpdir.join("temp-paired.1.fastq"))
p2 = str(tmpdir.join("temp-paired.2.fastq"))
params = "-a XX --interleaved".split()
params += ["-o", p1, "-p1", p2, "paired.1.fastq", "paired.2.fastq"]
with pytest.raises(SystemExit):
main(params)
def test_anywhere_anchored_3p():
with pytest.raises(SystemExit):
main(['-b', 'TTT$', datapath('small.fastq')])
("A", "T", True),
("C", "G", True),
("C", "T", True),
]
if discarduntrimmed:
combinations += [
("unknown", "G", False),
("A", "unknown", False),
]
params += ["--discard-untrimmed"]
else:
combinations += [
("unknown", "G", True),
("A", "unknown", True),
]
assert main(params) is None
for (name1, name2, should_exist) in combinations:
for i in (1, 2):
name = "combinatorial.{name1}_{name2}.{i}.fasta".format(name1=name1, name2=name2, i=i)
path = cutpath(os.path.join("combinatorial", name))
if should_exist:
assert tmpdir.join(name).check(), "Output file missing"
assert_files_equal(path, str(tmpdir.join(name)))
else:
assert not tmpdir.join(name).check(), "Output file should not exist"
def test_missing_file(tmpdir):
with pytest.raises(SystemExit):
main(["--paired-output", str(tmpdir.join("out.fastq")), datapath("paired.1.fastq")])
def test_second_too_short(tmpdir, cores):
# Create a truncated file in which the last read is missing
trunc2 = tmpdir.join("truncated.2.fastq")
with open(datapath("paired.2.fastq")) as f:
lines = f.readlines()
lines = lines[:-4]
trunc2.write("".join(lines))
with pytest.raises(SystemExit):
main([
"-o", "/dev/null",
"--paired-output", str(tmpdir.join("out.fastq")),
"--cores", str(cores),
datapath("paired.1.fastq"), str(trunc2)
])
record = ">r{}:{}\n{}\n".format(l1, l2, "A" * l1)
record += ">r{}:{}\n{}".format(l1, l2, "A" * l2)
with open(inpath, "w") as f:
print(record, file=f)
with open(expected, "w") as f:
if (m1 is None or func(l1, m1)) and (m2 is None or func(l2, m2)):
print(record, file=f)
assert os.path.exists(inpath)
assert os.path.exists(expected)
if m1 is None:
m1 = ""
if m2 is None:
m2 = ""
main(["--interleaved", "-o", outpath, "-" + name, "{}:{}".format(m1, m2), inpath])
assert_files_equal(expected, outpath)
def test_help():
with pytest.raises(SystemExit) as e:
main(["--help"])
assert e.value.args[0] == 0
def test_unconditional_cut_invalid_number():
with pytest.raises(SystemExit):
main(["-u", "a,b", datapath("small.fastq")])
def test_demultiplex():
tempdir = tempfile.mkdtemp(prefix='cutadapt-tests.')
multiout = os.path.join(tempdir, 'tmp-demulti.{name}.fasta')
params = ['-a', 'first=AATTTCAGGAATT', '-a', 'second=GTTCTCTAGTTCT', '-o', multiout, datapath('twoadapters.fasta')]
assert main(params) is None
assert_files_equal(cutpath('twoadapters.first.fasta'), multiout.format(name='first'))
assert_files_equal(cutpath('twoadapters.second.fasta'), multiout.format(name='second'))
assert_files_equal(cutpath('twoadapters.unknown.fasta'), multiout.format(name='unknown'))
shutil.rmtree(tempdir)