Mercurial > repos > fubar > tool_factory_2

comparison toolfactory/testclass.py @ 3:1c652687a08f draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author fubar
date Fri, 30 Apr 2021 07:06:57 +0000
parents
children
comparison
equal deleted inserted replaced
2:5fc0c9a93072 3:1c652687a08f
1 import argparse
2 import json
3 import os
4 import requests
5
6
7 from bioblend import galaxy
8
9 class ToolTester():
10 # test a newly installed tool using bioblend
11 """
12
13 https://github.com/nsoranzo/bioblend-tutorial/blob/master/historical_exercises/api-scripts.exercises/run_tool.py
14 import sys
15 import json
16 import requests
17 import output
18
19 BASE_URL = 'http://localhost:8080'
20
21 # -----------------------------------------------------------------------------
22 def run_tool( tool_id, history_id, **kwargs ):
23 full_url = BASE_URL + '/api/tools'
24
25 #EXERCISE: POST...
26
27 # -----------------------------------------------------------------------------
28 if __name__ == '__main__':
29 # e.g. ./run_tool.py Filter1 ebfb8f50c6abde6d '{ "input" : { "src": "hda", "id": "77f74776fd03cbc5" }, "cond" : "c6>=100.0" }'
30 # e.g. ./run_tool.py sort1 f597429621d6eb2b '{ "input": { "src": "hda", "id": "b472e2eb553fa0d1" }, "column": "c6", "style": "alpha", "column_set_0|other_column" : "c2", "column_set_0|other_style": "num" }'
31 tool_id, history_id = sys.argv[1:3]
32 params = json.loads( sys.argv[3] ) if len( sys.argv ) >= 4 else {}
33 response = run_tool( tool_id, history_id, **params )
34 output.output_response( response )
35
36
37 def get_testdata(self,urlin,fout):
38 '''
39 grab a test file
40 GET /api/tools/{tool_id}/test_data_download?tool_version={tool_version}&filename={filename}
41 http://localhost:8080/api/tools/tacrev/test_data_download?tool_version=2.00&filename=in
42 '''
43 """
44 def __init__(self, args):
45 self.galaxy = args.galaxy
46 self.key = args.key
47 self.tool_id = args.tool_id
48
49 def run_test(self):
50 """
51 GET /api/tools/{tool_id}/test_data_download?tool_version={tool_version}&filename={filename}
52 http://localhost:8080/api/tools/tacrev/test_data_download?tool_version=2.00&filename=input1
53 """
54 inputs = {}
55 gi = galaxy.GalaxyInstance(url=self.galaxy, key=self.key, verify=False)
56 chistory = gi.histories.get_most_recently_used_history()
57 chistory_id = chistory['id']
58 #contents = gi.histories.show_history(chistory_id, contents=True)
59 #print('####chistory',chistory,'\n#### contents=',contents)
60 #history = gi.histories.create_history(name=f"{self.tool_id}_test_history")
61 #new_hist_id = history['id']
62 fapi = ''.join([self.galaxy, '/api/tools/', self.tool_id, '/build'])
63 build = gi.make_get_request(url=fapi,params={"history_id":chistory_id}).json()
64 fapi = ''.join([self.galaxy, '/api/tools/', self.tool_id, '/test_data'])
65 test_data = requests.get(fapi, params={'key':self.key, 'history_id':chistory_id})# gi.make_get_request(url=fapi,params={"history_id":chistory_id,'key':self.key}).json()
66 print(test_data)
67 testinputs = test_data.json()[0].get('inputs',None)
68 print('testinputs',testinputs)
69 stateinputs = build.get('state_inputs',None) # 'input1': {'values': [{'id': '7b326180327c3fcc', 'src': 'hda'}]}}
70 if testinputs:
71 for k in testinputs.keys():
72 v = testinputs[k]
73 if '|' in k:
74 nk = k.split('|')[-1]
75 inputs[nk] = v
76 else:
77 inputs[k] = v
78 if stateinputs:
79 print('stateinputs',stateinputs)
80 for k in stateinputs.keys():
81 inp = stateinputs[k]
82 if isinstance(inp,dict):
83 if inp.get('values',None):
84 for anin in inp['values']:
85 if anin.get('id', None) and anin.get('src', None):
86 gi.histories.copy_dataset(chistory_id, anin['id'], source=anin['src'])
87 print('******copied id', anin['id'])
88 up = {k:anin}
89 print(up)
90 inputs.update(up) # replace the input def
91 print('after state inputs', inputs)
92 fapi = ''.join([self.galaxy, '/api/tools'])
93 r = gi.tools.run_tool(chistory_id, self.tool_id, inputs, input_format='legacy')
94 print(f"Called test on {self.tool_id} - got {r}")
95
96 def _parser():
97 parser = argparse.ArgumentParser()
98 parser.add_argument("-g", "--galaxy", help='URL of target galaxy',default="http://localhost:8080")
99 parser.add_argument("-a", "--key", help='Galaxy admin key', default="13073fde17d06591ce36e596e3c29904")
100 parser.add_argument("-t", "--tool_id", help='Tool id to test', default="plotter")
101 return parser
102
103
104 if __name__ == "__main__":
105 args = _parser().parse_args()
106 tt = ToolTester(args)
107 tt.run_test()
108