Commit e94aeda7 authored by David Mendez's avatar David Mendez
Browse files

Add functional test for loading context

parent fd36b296
# pylint: disable=import-error
"""
Module that tests getting a context data
"""
import requests
from specific_tests import utils
def run_test(server_base_url, delayed_jobs_server_base_path):
"""
Tests that getting the data of a context
:param server_base_url: base url of the running server. E.g. http://127.0.0.1:5000
:param delayed_jobs_server_base_path: base path for the delayed_jobs
"""
print('-------------------------------------------')
print('Testing getting the data of a context')
print('-------------------------------------------')
job_id = utils.submit_similarity_search_job(delayed_jobs_server_base_path)
utils.wait_until_job_finished(delayed_jobs_server_base_path, job_id)
print('Now going to load this context')
context_params = {
'delayed_jobs_base_url': 'https://www.ebi.ac.uk/chembl/interface_api/delayed_jobs',
'context_type': 'SIMILARITY',
'context_id': f'{job_id}'
}
url = f'{server_base_url}/contexts/get_context_data'
request = requests.post(url, data=context_params)
status_code = request.status_code
print(f'status_code: {status_code}')
response_text = request.text
utils.print_es_response(response_text)
assert status_code == 200, 'The request failed!'
......@@ -3,7 +3,6 @@
Module that tests an es_query with context
"""
import json
import time
import requests
......@@ -21,8 +20,8 @@ def run_test(server_base_url, delayed_jobs_server_base_path):
print('Testing a es_query with context')
print('-------------------------------------------')
job_id = submit_similarity_search_job(delayed_jobs_server_base_path)
wait_until_job_finished(delayed_jobs_server_base_path, job_id)
job_id = utils.submit_similarity_search_job(delayed_jobs_server_base_path)
utils.wait_until_job_finished(delayed_jobs_server_base_path, job_id)
print('Now going to make a request with this context')
url = utils.get_url_for_get_es_data(server_base_url)
......@@ -72,58 +71,3 @@ def run_test(server_base_url, delayed_jobs_server_base_path):
print('metadata: ', metadata)
assert metadata['total_results'] > 0, 'There should be more than 0 results!'
def submit_similarity_search_job(delayed_jobs_server_base_path):
"""
:param delayed_jobs_server_base_path: base path of the delayed jobs server
:return: job_id
"""
print('Launching a similarity search job...')
submission_url = utils.get_url_for_similarity_job_submission(delayed_jobs_server_base_path)
print('submission_url: ', submission_url)
payload = {
'search_type': 'SIMILARITY',
'search_term': 'NCCc1ccc(O)c(O)c1',
'threshold': 40,
'dl__ignore_cache': False
}
print('payload: ', payload)
submit_request = requests.post(submission_url, data=payload)
submission_status_code = submit_request.status_code
print(f'submission_status_code: {submission_status_code}')
assert submission_status_code == 200, 'Job could not be submitted!'
submission_response = submit_request.json()
job_id = submission_response.get('job_id')
print('job_id: ', job_id)
return job_id
def wait_until_job_finished(delayed_jobs_server_base_path, job_id):
"""
Waits until the job finished
:param delayed_jobs_server_base_path: base path of the delayed jobs server
:param job_id: id of the job
"""
print('Waiting until job finishes...')
status_url = utils.get_url_for_job_status(delayed_jobs_server_base_path, job_id)
print('status_url: ', status_url)
job_status = None
while job_status != 'FINISHED':
status_request = requests.get(status_url)
print('Status request response code: ', status_request.status_code)
status_response = status_request.json()
job_status = status_response.get('status')
job_progress = status_response.get('progress')
print('job_status: ', job_status)
print('job_progress: ', job_progress)
time.sleep(1)
......@@ -2,6 +2,7 @@
Module with utils functions for the tests
"""
import requests
import time
def get_url_for_get_es_data(server_base_url):
......@@ -63,3 +64,57 @@ def assert_get_request_succeeds(url_to_test):
response_text = config_request.text
print_es_response(response_text)
assert status_code == 200, 'The request failed!'
def submit_similarity_search_job(delayed_jobs_server_base_path):
"""
:param delayed_jobs_server_base_path: base path of the delayed jobs server
:return: job_id
"""
print('Launching a similarity search job...')
submission_url = get_url_for_similarity_job_submission(delayed_jobs_server_base_path)
print('submission_url: ', submission_url)
payload = {
'search_type': 'SIMILARITY',
'search_term': 'NCCc1ccc(O)c(O)c1',
'threshold': 40,
'dl__ignore_cache': False
}
print('payload: ', payload)
submit_request = requests.post(submission_url, data=payload)
submission_status_code = submit_request.status_code
print(f'submission_status_code: {submission_status_code}')
assert submission_status_code == 200, 'Job could not be submitted!'
submission_response = submit_request.json()
job_id = submission_response.get('job_id')
print('job_id: ', job_id)
return job_id
def wait_until_job_finished(delayed_jobs_server_base_path, job_id):
"""
Waits until the job finished
:param delayed_jobs_server_base_path: base path of the delayed jobs server
:param job_id: id of the job
"""
print('Waiting until job finishes...')
status_url = get_url_for_job_status(delayed_jobs_server_base_path, job_id)
print('status_url: ', status_url)
job_status = None
while job_status != 'FINISHED':
status_request = requests.get(status_url)
print('Status request response code: ', status_request.status_code)
status_response = status_request.json()
job_status = status_response.get('status')
job_progress = status_response.get('progress')
print('job_status: ', job_status)
print('job_progress: ', job_progress)
time.sleep(1)
\ No newline at end of file
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