forked from johnjh/vamps_mobedac_ws
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_gast_taxtable.py
124 lines (108 loc) · 4.83 KB
/
test_gast_taxtable.py
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
import httplib, urllib, urllib2
import json
from dbconn import test_engine
from threading import Thread
import time
import sys
def createObject(data_json, object_str):
headers = {'content-type': 'application/json'}
conn = httplib.HTTPConnection("localhost:8080")
conn.request("POST", "/mobedac_ws/" + object_str, data_json, headers)
response = conn.getresponse()
print response.status, response.reason
data = response.read()
print data
conn.close()
return data
def getObject(object_type, obj_id):
conn = httplib.HTTPConnection("localhost:8080")
conn.request("GET", "/mobedac_ws/" + object_type + "/" + str(obj_id))
response = conn.getresponse()
data = response.read()
conn.close()
return data
def getListenerResults():
conn = httplib.HTTPConnection("localhost:8081")
conn.request("GET", "/get_requests")
response = conn.getresponse()
data = response.read()
conn.close()
return data
def clearListenerResults():
conn = httplib.HTTPConnection("localhost:8081")
conn.request("GET", "/clear_requests")
response = conn.getresponse()
conn.close()
# from mobedac_blank.py
#file_data = {"analysis_system":"VAMPS","library_ids":["139824","139820"],
# "analysis_params":{"auth":"TkzmLuiSuwQEhivEveZ7tvYiB","user":"mobedac","vamps_user":"plustig"}}
#file_contents = json.dumps(file_data)
submission_data_l1_only = {"library_ids" : ['l1'
#,
#'l2',
# 'l3', 'l0', 'ls0'
],
"analysis_params": {
"user":"mobedac",
"auth" : "asdfasdfasdf"}
}
submission_data_l123 = {"library_ids" : ['l1','l2','l3'],
"analysis_params": {
"user":"mobedac",
"auth" : "asdfasdfasdf"}
}
def postSubmission(submission_object):
return createObject(json.dumps(submission_object), "submission")
def change_sequence_file_type(new_type):
urllib2.urlopen("http://localhost:8081/mobedac/set_sequence_set_file_type/" + new_type)
# main routine for each submission request we want to make
# first clears the db
# then tells the mobedac_vamps_listener process to clear out any data it has accumulated
def test_by_type(seq_file_type, submission_object, expected_names):
print "***** Starting test for file type: " + seq_file_type
test_engine.execute("Delete from submission_details;")
test_engine.execute("Delete from submission;")
# clear all requests from listener
clearListenerResults()
# now create a request
change_sequence_file_type(seq_file_type)
submission_id = postSubmission(submission_object)
# now we need to loop and check status
while True:
# get the status of the submission
submission_str = getObject("submission", submission_id)
submission_json = json.loads(submission_str)
overall_status = submission_json['status_code']
if overall_status == 1:
# call over to the mobedac_vamps_listener.py process and get
# the list of all calls that it registered made to it from our API service
listener_calls = getListenerResults()
print "Calls made to Mobedac: " + listener_calls
print "Expected Calls Mobedac: " + str(expected_names)
# are they ok?
listener_calls_array = json.loads(listener_calls)
if len(listener_calls_array) != len(expected_names ):
print "Incorrect number of calls to VAMPS and Mobedac got: " + str(len(listener_calls_array))
error = False
for actual,expected in zip(listener_calls_array,expected_names):
if actual != expected:
error = True
print "Incorrect call, expected: " + expected + " got: " + actual
break
if error == False:
print "Success!!! for " + seq_file_type
break
else:
print "Fail for " + seq_file_type
break
time.sleep(10)
# should post 3 times, gast twice, and tax table once and 1 data return
#expected_names_123 = ['upload_data_post', 'upload_data_post', 'upload_data_post', 'upload_data_gast', 'generate_taxonomy_table', 'POST results']
#test_by_type('fasta', submission_data_l123, expected_names_123)
#===============
# should post, gast, tax table and data return 1 time each
expected_names = ['upload_data_post', 'upload_data_gast', 'generate_taxonomy_table', 'POST results']
#test_by_type('fasta', submission_data_l1_only, expected_names)
#test_by_type('fastq_small', submission_data_l1_only, expected_names)
test_by_type('sff_small',submission_data_l1_only, expected_names)
#test_by_type('fastq_large', submission_data_l1_only, expected_names)