From 31a4bfc594538b6c14a68c05b8029e840116ac54 Mon Sep 17 00:00:00 2001 From: Andrea Pierleoni Date: Sat, 10 Dec 2016 07:21:47 +0000 Subject: [PATCH] restored missing test file --- tests/test_client.py | 307 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 307 insertions(+) create mode 100644 tests/test_client.py diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..c03f728 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,307 @@ +import json +import logging +import os +import unittest + +from opentargets import Connection +from opentargets import OpenTargetsClient +from opentargets.statistics import HarmonicSumScorer + +logger = logging.getLogger(__name__) +logger.setLevel(logging.WARNING) + + +class OpenTargetClientTest(unittest.TestCase): + _AUTO_GET_TOKEN = 'auto' + + def setUp(self): + + self.client = OpenTargetsClient() + self.http2_client = OpenTargetsClient(use_http2=True) + self.auth_client = OpenTargetsClient(auth_app_name='test', + auth_secret='test', ) + + def tearDown(self): + self.client.close() + + def testSearchTargetCorrectResult(self): + target_symbol = 'BRAF' + response = self.client.search(target_symbol) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['type'], 'search-object-target') + self.assertEqual(result['id'], 'ENSG00000157764') + self.assertEqual(result['data']['approved_symbol'], target_symbol) + + def testSearchTargetFetchAllResults(self): + target_symbol = 'BRAF' + response = self.client.search(target_symbol) + total_results = len(response) + self.assertGreater(total_results, 0) + c = 0 + for i in response: + c += 1 + self.assertEqual(total_results, c) + + def testSearchTargetFetchAllResultsAuth(self): + target_symbol = 'BRAF' + response = self.auth_client.search(target_symbol) + total_results = len(response) + self.assertGreater(total_results, 0) + c = 0 + for i in response: + c += 1 + self.assertEqual(total_results, c) + + def testSearchTargetCorrectResultHTTP2(self): + target_symbol = 'BRAF' + response = self.http2_client.search(target_symbol) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['type'], 'search-object-target') + self.assertEqual(result['id'], 'ENSG00000157764') + self.assertEqual(result['data']['approved_symbol'], target_symbol) + + def testSearchTargetFetchAllResultsHTTP2(self): + target_symbol = 'BRAF' + response = self.http2_client.search(target_symbol) + total_results = len(response) + self.assertGreater(total_results, 0) + c = 0 + for i in response: + c += 1 + self.assertEqual(total_results, c) + + def testSearchDiseaseCorrectResult(self): + disease_label = 'cancer' + response = self.client.search(disease_label) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['type'], 'search-object-disease') + self.assertEqual(result['id'], 'EFO_0000311') + + # #this takes a lot to run + # def testSearchDiseaseFetchAllResults(self): + # disease_label = 'cancer' + # response = self.client.search(disease_label, size = 100) + # total_results = len(response) + # self.assertGreater(total_results,0) + # c=0 + # for i in response: + # c+=1 + # self.assertEqual(total_results, c) + # print(total_results, c) + + def testGetAssociation(self): + association_id = "ENSG00000157764-EFO_0005803" + response = self.client.get_association(association_id) + self.assertEquals(len(response), 1) + self.assertEquals(association_id, response[0]['id']) + + def testFilterAssociations(self): + response = self.client.filter_associations() + self.assertGreater(len(response), 0) + total = response.info.total + print(response) + response.filter(target='ENSG00000157764') + self.assertLess(len(response), total) + print(response) + total = response.info.total + response.filter(direct=True) + self.assertLess(len(response), total) + print(response) + total = response.info.total + response.filter(scorevalue_min=0.2) + self.assertLess(len(response), total) + print(response) + total = response.info.total + response.filter(therapeutic_area='efo_0000701') + self.assertLess(len(response), total) + print(response) + results = [] + for i, r in enumerate(response): + print(i, r['id'], r['association_score']['overall'], r['disease']['efo_info']['label']) + results.append(r) + response_multi = self.client.filter_associations(target='ENSG00000157764', direct=True, scorevalue_min=0.2, + therapeutic_area='efo_0000701') + self.assertEqual(len(response_multi), response.info.total) + for i, r in enumerate(response_multi): + self.assertEqual(results[i]['id'], r['id']) + response_chained = self.client.filter_associations().filter(target='ENSG00000157764').filter( + direct=True).filter(therapeutic_area='efo_0000701').filter(scorevalue_min=0.2) + self.assertEqual(len(response_chained), response.info.total) + for i, r in enumerate(response_chained): + self.assertEqual(results[i]['id'], r['id']) + + def testGetAssociationsForTarget(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['target']['gene_info']['symbol'], target_symbol) + + def testGetAssociationsForDisease(self): + disease_label = 'cancer' + response = self.client.get_associations_for_disease(disease_label) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['disease']['efo_info']['label'], disease_label) + + def testGetEvidence(self): + evidence_id = "5cf863da265c32d112ff4fc3bfc25ab3" + response = self.client.get_evidence(evidence_id) + self.assertEquals(len(response), 1) + self.assertEquals(evidence_id, response[0]['id']) + + def testFilterEvidence(self): + response = self.client.filter_evidence() + self.assertGreater(len(response), 0) + + def testGetEvidenceForTarget(self): + target_symbol = 'BRAF' + response = self.client.get_evidence_for_target(target_symbol) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['target']['gene_info']['symbol'], target_symbol) + + def testGetEvidenceForDisease(self): + disease_label = 'medulloblastoma' + response = self.client.get_evidence_for_disease(disease_label) + self.assertGreater(len(response), 0) + result = next(response) + self.assertEqual(result['disease']['efo_info']['label'], disease_label) + + def testSerialiseToJson(self): + target_symbol = 'BRAF' + '''test iterable version''' + response = self.client.get_associations_for_target(target_symbol) + items = len(response) + self.assertGreater(len(response), 0) + json_output = response.to_json() + parsed_json = [json.loads(i) for i in json_output] + self.assertEqual(items, len(parsed_json)) + '''test non iterable version''' + response = self.client.get_associations_for_target(target_symbol) + items = len(response) + self.assertGreater(len(response), 0) + json_output = response.to_json(iterable=False) + parsed_json = json.loads(json_output) + self.assertEqual(items, len(parsed_json)) + + def testResultToPandasDataFrame(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol) + items = len(response) + self.assertGreater(len(response), 0) + dataframe = response.to_dataframe() + self.assertEqual(len(dataframe), items) + + def testResultToPandasCSV(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol, + fields=['association_score.*', + 'target.gene_info.symbol', + 'disease.efo_info.*'] + ) + items = len(response) + self.assertGreater(len(response), 0) + csv = response.to_csv() + filename = 'braf_associations.csv' + open(filename, 'wb').write(csv.encode('utf-8')) + self.assertEqual(len(csv.split('\n')), items + 2) + self.assertTrue(os.path.isfile(filename)) + os.remove(filename) + + def testResultToPandasExcel(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol, + fields=['association_score.*', + 'target.gene_info.symbol', + 'disease.efo_info.*'] + ) + self.assertGreater(len(response), 0) + filename = 'braf_associations.xls' + response.to_excel(filename) + self.assertTrue(os.path.isfile(filename)) + os.remove(filename) + + def testResultToFile(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol, + fields=['association_score.*', + 'target.gene_info.symbol', + 'disease.efo_info.*'] + ) + self.assertGreater(len(response), 0) + filename = 'braf_associations.json.gz' + response.to_file(filename) + self.assertTrue(os.path.isfile(filename)) + os.remove(filename) + + + def testSerialiseToNamedtuple(self): + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol) + items = len(response) + self.assertGreater(len(response), 0) + nt_output = list(response.to_namedtuple()) + for i, result in enumerate(nt_output): + self.assertIsNotNone(result.target.id) + self.assertEqual(items, i + 1) + + def testGetStats(self): + response = self.client.get_stats() + self.assertEquals(len(response), 0) + + def testAutodetectPost(self): + self.assertFalse(Connection._auto_detect_post({'target': ['ENSG00000157764']})) + self.assertTrue(Connection._auto_detect_post({'target': ['ENSG00000157764', + 'ENSG00000171862', + 'ENSG00000136997', + 'ENSG00000012048', + 'ENSG00000139618', + ]})) + + def testGetToPost(self): + response = self.client.conn.get('/public/association/filter', params={'target': ['ENSG00000157764', + 'ENSG00000171862', + 'ENSG00000136997', + 'ENSG00000012048', + 'ENSG00000139618', + ]}) + self.assertGreater(len(response), 0) + + def testCustomScore(self): + def score_with_datatype_subset(datatypes, results): + for r in results: + datatype_scores = r['association_score']['datatypes'] + filtered_scores = [datatype_scores[dt] for dt in datatypes] + custom_score = HarmonicSumScorer.harmonic_sum(filtered_scores) + if custom_score: + yield (round(custom_score, 3), r['disease']['id'], dict(zip(datatypes, filtered_scores))) + + target_symbol = 'BRAF' + response = self.client.get_associations_for_target(target_symbol) + self.assertGreater(len(response), 0) + for i, filtered_data in enumerate(score_with_datatype_subset(['genetic_association', + 'known_drug', + 'somatic_mutation'], + response)): + self.assertGreater(filtered_data[0], 0.) + + self.assertLess(i, len(response)) + + def testGetAvailableEndpoints(self): + endpoints = self.client.conn.get_api_endpoints() + self.assertTrue('/public/search' in endpoints) + + def testGetEndpointDocs(self): + docs = self.client.conn.api_endpoint_docs('/public/search') + self.assertGreater(len(docs['get']['parameters']),0) + + def testPing(self): + response = self.client.conn.ping() + if isinstance(response, bool): + self.assertTrue(response) + else: + self.assertIsNotNone(response) \ No newline at end of file