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
|
#!/usr/bin/env python
# test_sfffile.py
import os
import tempfile
from cogent.util.unit_test import TestCase, main
from cogent.parse.binary_sff import parse_binary_sff
from cogent.app.util import ApplicationError
from cogent.app.sfffile import Sfffile
__author__ = "Kyle Bittinger"
__copyright__ = "Copyright 2007-2012, The Cogent Project"
__credits__ = ["Kyle Bittinger"]
__license__ = "GPL"
__version__ = "1.5.3"
__maintainer__ = "Kyle Bittinger"
__email__ = "kylebittinger@gmail.com"
__status__ = "Prototype"
class SfffileTests(TestCase):
"""Test the Sfffile application controller.
"""
def setUp(self):
test_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
self.sff_fp = os.path.join(test_dir, 'data', 'test.sff')
def _check_unmodified_sff_contents(self, sff_file):
"""Extracting repeated code from sfffile tests"""
sff_file.seek(0)
header, reads_gen = parse_binary_sff(sff_file)
reads = list(reads_gen)
self.assertEqual(header["number_of_reads"], 1)
self.assertEqual(len(reads), 1)
self.assertEqual(reads[0]['Name'], 'FA6P1OK01CGMHQ')
def test_exit_status(self):
"""Sfffile should raise ApplicationError if exit status is nonzero."""
a = Sfffile()
self.assertRaises(ApplicationError, a, 'an_sff_file_that_does_not_exist.sff')
def test_call(self):
"""Simple sfffile call should produce expected output."""
a = Sfffile()
app_results = a(self.sff_fp)
self._check_unmodified_sff_contents(app_results['sff'])
app_results.cleanUp()
def test_call_with_output_path(self):
"""Sfffile should store output to specified filepath."""
_, output_fp = tempfile.mkstemp()
a = Sfffile()
a.Parameters['-o'].on(output_fp)
app_results = a(self.sff_fp)
self._check_unmodified_sff_contents(open(output_fp))
self._check_unmodified_sff_contents(app_results['sff'])
app_results.cleanUp()
def test_call_with_included_accession_numbers(self):
"""Sfffile should include specified accession numbers in output."""
accno_file = tempfile.NamedTemporaryFile()
accno_file.write('FA6P1OK01CGMHQ\n')
accno_file.seek(0)
a = Sfffile()
a.Parameters['-i'].on(accno_file.name)
app_results = a(self.sff_fp)
self._check_unmodified_sff_contents(app_results['sff'])
app_results.cleanUp()
def test_call_with_excluded_accession_numbers(self):
"""Sfffile should exclude specified accession numbers in output."""
accno_file = tempfile.NamedTemporaryFile()
accno_file.write('FA6P1OK01CGMHQ\n')
accno_file.seek(0)
a = Sfffile()
a.Parameters['-e'].on(accno_file.name)
app_results = a(self.sff_fp)
header, reads_gen = parse_binary_sff(app_results['sff'])
reads = list(reads_gen)
self.assertEqual(header["number_of_reads"], 0)
self.assertEqual(len(reads), 0)
app_results.cleanUp()
if __name__ == '__main__':
main()
|
