Commit 6ca94b34 authored by Mikaël Salson's avatar Mikaël Salson Committed by Vidjil Team

get_analysis: Simplify the function in controllers/ by splitting it.

The function could not be properly tested because unit test cannot (in an easy way?)
work on files stored by the DB.
The get_analysis function has been split in several functions.
parent 192211b3
......@@ -308,42 +308,15 @@ def get_analysis():
if not auth.can_view_patient(request.vars["patient"]):
error += "you do not have permission to consult this patient ("+str(request.vars["patient"])+")"
## empty analysis file
res = {"samples": {"number": 0,
"original_names": [],
"order": [],
"info_sequence_file" : []
},
"custom": [],
"clusters": [],
"clones" : [],
"tags": {},
"vidjil_json_version" : "2014.09"
}
if "custom" in request.vars :
return gluon.contrib.simplejson.dumps(res, separators=(',',':'))
return gluon.contrib.simplejson.dumps(get_default_analysis(), separators=(',',':'))
if error == "" :
## récupération des infos se trouvant dans le fichier .analysis
analysis_query = db(db.analysis_file.patient_id == request.vars["patient"]).select(orderby=~db.analysis_file.analyze_date)
if len(analysis_query) > 0 :
row = analysis_query.first()
f = open(defs.DIR_RESULTS+'/'+row.analysis_file, "r")
analysis = gluon.contrib.simplejson.loads(f.read())
f.close()
if 'cluster' in analysis:
res["clusters"] = analysis["cluster"]
if 'clusters' in analysis :
res["clusters"] = analysis["clusters"]
res["clones"] = analysis["clones"]
res["tags"] = analysis["tags"]
res["samples"]= analysis["samples"]
res["info_patient"] = db.patient[request.vars["patient"]].info
return gluon.contrib.simplejson.dumps(res, separators=(',',':'))
analysis_data = get_analysis_data(request.vars['patient'])
analysis_data["info_patient"] = db.patient[request.vars["patient"]].info
return gluon.contrib.simplejson.dumps(analysis_data, separators=(',',':'))
else :
res = {"success" : "false",
......
import defs
import vidjil_utils
def get_analysis_from_patient(patient_id):
'''
Returns the data from the DB corresponding to the analysis of this patient
'''
return db(db.analysis_file.patient_id == patient_id).select(orderby=~db.analysis_file.analyze_date)
def get_analysis_data(patient_id):
'''
Return an analysis file (if any, or a default file) for the given patient.
'''
result = get_default_analysis()
analysis_query = get_analysis_from_patient(patient_id)
if len(analysis_query) > 0:
row = analysis_query.first()
f = open(defs.DIR_RESULTS+'/'+row.analysis_file, "r")
return get_clean_analysis(f)
return result
def get_clean_analysis(filehandle):
'''
Take a filehandle in parameter and return a dictionary made of
the analysis in the correct format.
'''
result = get_default_analysis()
analysis = gluon.contrib.simplejson.loads(filehandle.read())
result = vidjil_utils.put_properties_in_dict(analysis, result, {'cluster': 'clusters',
'clusters': 'clusters',
'clones': 'clones',
'tags': 'tags',
'samples': 'samples'})
return result
def get_analysis_info(json_paths, patient_id):
'''Return the information in the analysis files for the given patient
under the provided json paths.
'''
analysis_file = get_analysis_from_patient(patient_id)
results = []
for analysis in analysis_file:
filename=defs.DIR_RESULTS + analysis.analysis_file
results.append(vidjil_utils.extract_fields_from_json(json_paths, None, filename))
return results
def get_default_analysis():
'''
return an empty analysis file
'''
return {"samples": {"number": 0,
"original_names": [],
"order": [],
"info_sequence_file" : []
},
"custom": [],
"clusters": [],
"clones" : [],
"tags": {},
"vidjil_json_version" : "2014.09"
}
......@@ -35,7 +35,12 @@ class DefaultController(unittest.TestCase):
# for defs
current.db = db
current.auth = auth
def _get_fake_analysis_file(self):
file = tempfile.TemporaryFile()
file.write('{"toto": 1, "bla": [], "clones": {"id": "AATA", "tag": 0}}')
file.seek(0)
return file
def testIndex(self):
resp = index()
......@@ -92,14 +97,6 @@ class DefaultController(unittest.TestCase):
self.assertTrue(defs.PORT_FUSE_SERVER is None, 'get_custom_data returns error without fuse server')
else:
self.assertEqual(resp['reads']['segmented'][0], resp['reads']['segmented'][1], "get_custom_data doesn't return a valid json")
def testGetAnalysis(self):
request.vars['config'] = fake_config_id
request.vars['patient'] = fake_patient_id
resp = get_analysis()
self.assertNotEqual(resp.find('"info_patient":"plop"'), -1, "get_analysis doesn't return a valid json")
def testSaveAnalysis(self):
class emptyClass( object ):
......@@ -115,5 +112,13 @@ class DefaultController(unittest.TestCase):
request.vars['patient'] = fake_patient_id
resp = save_analysis()
self.assertNotEqual(resp.find('analysis saved","success":"true"'), -1, "save_analysis failed")
resp = get_analysis_from_patient(fake_patient_id)
self.assertEqual(len(resp), 1, "should have one analysis for that patient %d"%len(resp))
self.assertEqual(resp[0].patient_id, fake_patient_id, "get_analysis doesn't have the correct patient")
def testGetCleanAnalysis(self):
analysis = get_clean_analysis(self._get_fake_analysis_file())
self.assertEqual(analysis['clones']['id'], 'AATA', 'Bad clone id')
self.assertEqual(analysis['tags'], {}, 'Bad tags entry in analysis')
self.assertEqual(analysis['vidjil_json_version'], '2014.09', 'Bad vidjil_json_version string')
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment