[973924]: / qiita_db / handlers / tests / test_prep_template.py

Download this file

218 lines (186 with data), 9.6 kB

  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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
# -----------------------------------------------------------------------------
# Copyright (c) 2014--, The Qiita Development Team.
#
# Distributed under the terms of the BSD 3-clause License.
#
# The full license is in the file LICENSE, distributed with this software.
# -----------------------------------------------------------------------------
from unittest import main, TestCase
from json import loads, dumps
from os.path import join
from functools import partial
from tornado.web import HTTPError
from qiita_db.handlers.tests.oauthbase import OauthTestingBase
import qiita_db as qdb
from qiita_db.handlers.prep_template import _get_prep_template
class UtilTests(TestCase):
def test_get_prep_template(self):
obs = _get_prep_template(1)
exp = qdb.metadata_template.prep_template.PrepTemplate(1)
self.assertEqual(obs, exp)
# It does not exist
with self.assertRaises(HTTPError):
_get_prep_template(100)
class PrepTemplateHandlerTests(OauthTestingBase):
def test_get_does_not_exist(self):
obs = self.get('/qiita_db/prep_template/100/', headers=self.header)
self.assertEqual(obs.code, 404)
def test_get_no_header(self):
obs = self.get('/qiita_db/prep_template/1/')
self.assertEqual(obs.code, 400)
def test_get(self):
obs = self.get('/qiita_db/prep_template/1/', headers=self.header)
self.assertEqual(obs.code, 200)
db_test_template_dir = qdb.util.get_mountpoint('templates')[0][1]
path_builder = partial(join, db_test_template_dir)
obs = loads(obs.body)
# have to check per key because since patch 51 we are updating the
# test info files
self.assertEqual(obs['data_type'], '18S')
self.assertEqual(obs['artifact'], 1)
self.assertEqual(obs['investigation_type'], 'Metagenomics')
self.assertEqual(obs['study'], 1)
self.assertEqual(obs['status'], 'private')
self.assertTrue(obs['sample-file'].startswith(
path_builder('1_')))
self.assertTrue(obs['prep-file'].startswith(
path_builder('1_prep_1_')))
class PrepTemplateDataHandlerTests(OauthTestingBase):
def test_get_does_not_exist(self):
obs = self.get('/qiita_db/prep_template/100/data/',
headers=self.header)
self.assertEqual(obs.code, 404)
def test_get_no_header(self):
obs = self.get('/qiita_db/prep_template/100/data/')
self.assertEqual(obs.code, 400)
def test_get(self):
obs = self.get('/qiita_db/prep_template/1/data/', headers=self.header)
self.assertEqual(obs.code, 200)
obs = loads(obs.body)
self.assertCountEqual(obs.keys(), ['data'])
obs = obs['data']
exp = ['1.SKB2.640194', '1.SKM4.640180', '1.SKB3.640195',
'1.SKB6.640176', '1.SKD6.640190', '1.SKM6.640187',
'1.SKD9.640182', '1.SKM8.640201', '1.SKM2.640199',
'1.SKD2.640178', '1.SKB7.640196', '1.SKD4.640185',
'1.SKB8.640193', '1.SKM3.640197', '1.SKD5.640186',
'1.SKB1.640202', '1.SKM1.640183', '1.SKD1.640179',
'1.SKD3.640198', '1.SKB5.640181', '1.SKB4.640189',
'1.SKB9.640200', '1.SKM9.640192', '1.SKD8.640184',
'1.SKM5.640177', '1.SKM7.640188', '1.SKD7.640191']
self.assertCountEqual(list(obs.keys()), exp)
obs = obs['1.SKB1.640202']
exp = {
'barcode': 'GTCCGCAAGTTA',
'center_name': 'ANL',
'center_project_name': None,
'emp_status': 'EMP',
'experiment_center': 'ANL',
'experiment_design_description':
'micro biome of soil and rhizosphere of cannabis plants '
'from CA',
'experiment_title': 'Cannabis Soil Microbiome',
'illumina_technology': 'MiSeq',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol':
'This analysis was done as in Caporaso et al 2011 Genome '
'research. The PCR primers (F515/R806) were developed against '
'the V4 region of the 16S rRNA (both bacteria and archaea), '
'which we determined would yield optimal community clustering '
'with reads of this length using a procedure similar to that '
'of ref. 15. [For reference, this primer pair amplifies the '
'region 533_786 in the Escherichia coli strain 83972 sequence '
'(greengenes accession no. prokMSA_id:470367).] The reverse '
'PCR primer is barcoded with a 12-base error-correcting Golay '
'code to facilitate multiplexing of up to 1,500 samples per '
'lane, and both PCR primers contain sequencer adapter '
'regions.',
'pcr_primers': 'FWD:GTGCCAGCMGCCGCGGTAA; REV:GGACTACHVGGGTWTCTAAT',
'platform': 'Illumina',
'primer': 'GTGCCAGCMGCCGCGGTAA',
'run_center': 'ANL',
'run_date': '8/1/12',
'run_prefix': 's_G1_L001_sequences',
'samp_size': '.25,g',
'sample_center': 'ANL',
'sequencing_meth': 'Sequencing by synthesis',
'study_center': 'CCME',
'target_gene': '16S rRNA',
'target_subfragment': 'V4',
'qiita_prep_id': '1'}
self.assertEqual(obs, exp)
def test_get_sample_information(self):
obs = self.get(
'/qiita_db/prep_template/1/data/?sample_information=true',
headers=self.header)
self.assertEqual(obs.code, 200)
obs = loads(obs.body)
self.assertCountEqual(obs.keys(), ['data'])
# let's just check that the samples and the keys from the first element
# match - this assures us that is the sample_info, the rest is
# basically the same as the regular prep_info
obs = obs['data']
exp = ['1.SKB2.640194', '1.SKM4.640180', '1.SKB3.640195',
'1.SKB6.640176', '1.SKD6.640190', '1.SKM6.640187',
'1.SKD9.640182', '1.SKM8.640201', '1.SKM2.640199',
'1.SKD2.640178', '1.SKB7.640196', '1.SKD4.640185',
'1.SKB8.640193', '1.SKM3.640197', '1.SKD5.640186',
'1.SKB1.640202', '1.SKM1.640183', '1.SKD1.640179',
'1.SKD3.640198', '1.SKB5.640181', '1.SKB4.640189',
'1.SKB9.640200', '1.SKM9.640192', '1.SKD8.640184',
'1.SKM5.640177', '1.SKM7.640188', '1.SKD7.640191']
self.assertCountEqual(list(obs.keys()), exp)
exp = ['ph', 'temp', 'depth', 'country', 'texture', 'altitude',
'latitude', 'taxon_id', 'elevation', 'env_biome', 'longitude',
'tot_nitro', 'host_taxid', 'common_name', 'description',
'env_feature', 'env_package', 'sample_type', 'tot_org_carb',
'dna_extracted', 'samp_salinity', 'anonymized_name',
'host_subject_id', 'scientific_name', 'assigned_from_geo',
'season_environment', 'water_content_soil',
'collection_timestamp', 'description_duplicate',
'physical_specimen_location', 'physical_specimen_remaining',
'qiita_study_id']
self.assertCountEqual(obs['1.SKB2.640194'].keys(), exp)
class PrepTemplateAPItestHandlerTests(OauthTestingBase):
def test_post(self):
metadata_dict = {
'SKB8.640193': {'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'GTCCGCAAGTTA',
'platform': 'Illumina',
'instrument_model': 'Illumina MiSeq'},
'SKD8.640184': {'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'GTCCGCAAGTTA',
'platform': 'Illumina',
'instrument_model': 'Illumina MiSeq'}}
data = {'prep_info': dumps(metadata_dict),
'study': 1,
'data_type': '16S'}
obs = self.post('/apitest/prep_template/', headers=self.header,
data=data)
self.assertEqual(obs.code, 200)
obs = loads(obs.body)
self.assertCountEqual(obs.keys(), ['prep'])
pt = qdb.metadata_template.prep_template.PrepTemplate(obs['prep'])
# default name, and creation_job_id
self.assertTrue(pt.name.startswith('Prep information'))
self.assertIsNone(pt.creation_job_id)
self.assertCountEqual(pt.keys(), ['1.SKB8.640193', '1.SKD8.640184'])
# testing that a new prep doesn't break the call due to empty artifact
obs = self.get('/qiita_db/prep_template/%d/' % pt.id,
headers=self.header)
self.assertEqual(obs.code, 200)
# testing that using a non default value actually works
data['name'] = 'my long and informative name'
data['job-id'] = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
obs = self.post('/apitest/prep_template/', headers=self.header,
data=data)
obs = loads(obs.body)
pt = qdb.metadata_template.prep_template.PrepTemplate(obs['prep'])
self.assertEqual(pt.name, data['name'])
self.assertEqual(pt.creation_job_id, data['job-id'])
# testing setter
jid = 'aaaaaaaa-aaaa-bbbb-aaaa-aaaaaaaaaaaa'
pt.creation_job_id = jid
self.assertEqual(pt.creation_job_id, jid)
if __name__ == '__main__':
main()