From a7f1f6720dd06667db2f4f731848aa8093f01697 Mon Sep 17 00:00:00 2001 From: LETORT Sebastien <sebastien.letort@irisa.fr> Date: Fri, 6 Dec 2019 14:44:09 +0100 Subject: [PATCH] Some unit tests for the module. --- pytest.ini | 7 + requirements.txt | 4 + tests/test_allgo.py | 367 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 378 insertions(+) create mode 100644 pytest.ini create mode 100644 requirements.txt create mode 100644 tests/test_allgo.py diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..db32819 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,7 @@ +[pytest] +markers = + cstr: related to constructor + getter: test getters + error: test when an error occured + success: test output on successfull request + dbg: for debug. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..80c7eaf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +pytest +sphinx +nbsphinx +pandoc diff --git a/tests/test_allgo.py b/tests/test_allgo.py new file mode 100644 index 0000000..9500d9d --- /dev/null +++ b/tests/test_allgo.py @@ -0,0 +1,367 @@ +#! /usr/bin/env python3 + +# standard lib +import errno +import os +import sys +import tempfile + +# not standard +import pytest + +if 2 == sys.version_info.major: + from mock import Mock, patch +else: + from unittest.mock import Mock, patch + +# local import +import allgo # I need to modify constant +from allgo import * + + +# ================================================ +@pytest.fixture(scope="module") +def no_env_token(): + # setup + try: + os.environ.pop('ALLGO_TOKEN') # , default=None) <= python3 + except KeyError: + pass + + yield + + # teardown + + +# ================================================ +def test_constants(): + assert MAIN_INSTANCE_URL.startswith("https://") + assert TOKEN_FILE.endswith(".allgo_token") + + +# ------------------------------------------------ +@pytest.mark.cstr +def test_cstr__no_token(no_env_token, tmp_path): + allgo.TOKEN_FILE = os.path.join(str(tmp_path), 'fake') + + with pytest.raises(TokenError): + Client() + + +@pytest.mark.cstr +def test_cstr__mini(no_env_token): + assert isinstance(Client(token="jeton"), Client) + + +# ------------------------------------------------ +@pytest.mark.getter +def test_token_getter__arg(): + token = "un_jeton" + c = Client(token=token) + assert token == c.token + + +@pytest.mark.getter +def test_token_getter__envvar(): + token = "un_jeton" + os.environ['ALLGO_TOKEN'] = token + c = Client() + assert token == c.token + + +@pytest.mark.getter +def test_token_getter__file(no_env_token, tmp_path): + token = "un_jeton" + allgo.TOKEN_FILE = os.path.join(str(tmp_path), 'fake') + with open(TOKEN_FILE, "w") as fs: + fs.write(token) + c = Client() + assert token == c.token + + +# ------------------------------------------------ +@pytest.fixture +def client(): + return Client(token="fake") + + +# ------------------------------------------------ +@pytest.mark.getter +def test_allgo_url__default(client): + assert MAIN_INSTANCE_URL == client.allgo_url + + +@pytest.mark.getter +def test_allgo_url__arg(): + token = "un_jeton" + url = "a wanted url" + c = Client(token=token, allgo_url=url) + assert url == c.allgo_url + + +# ------------------------------------------------ +@pytest.mark.getter +def test_verify_tls__default(client): + assert True is client.verify_tls + + +@pytest.mark.getter +def test_verify_tls__arg(): + token = "un_jeton" + verify = False + c = Client(token=token, verify_tls=verify) + assert verify == c.verify_tls + + +# ------------------------------------------------ +# -- Mock usage => https://realpython.com/testing-third-party-apis-with-mocks/ +# -- json returned values are directly copied from allgo repository +# -- -> django/allgo/api/v1/views.py +# ------------------------------------------------ +@pytest.mark.error +@patch('allgo.requests.post') +def test_create_job__user_not_exist(mock_post, client): + # -- mock + mock_post.return_value = Mock(status_code=401) + # ~ mock_post.return_value.json.return_value = \ + # ~ {'error': 'API request without http authorisation'} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.create_job('fake') + + err = err_info.value + assert 401 == err.status_code + # ~ assert '' == err.msg + + +@pytest.mark.error +@patch('allgo.requests.post') +def test_create_job__app_not_exist(mock_post, client): + # -- mock + mock_post.return_value = Mock(status_code=404) + mock_post.return_value.json.return_value = \ + {'error': 'Application not found'} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.create_job('fake') + + err = err_info.value + assert 404 == err.status_code + assert 'Application not found' == err.msg + + +@pytest.mark.error +@patch('allgo.requests.post') +def test_create_job__app_not_published(mock_post, client): + # -- mock + mock_post.return_value = Mock(status_code=404) + mock_post.return_value.json.return_value = \ + {'error': "This app is not yet published"} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.create_job('fake') + + err = err_info.value + assert 404 == err.status_code + assert 'not yet published' in err.msg + + +@pytest.mark.error +@patch('allgo.requests.post') +def test_create_job__queue_not_exist(mock_post, client): + # -- mock + mock_post.return_value = Mock(status_code=400) + mock_post.return_value.json.return_value = \ + {'error': 'Unknown queue'} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.create_job('fake') + + err = err_info.value + assert 400 == err.status_code + assert 'Unknown queue' == err.msg + + +@pytest.mark.error +@patch('allgo.requests.post') +def test_create_job__param_error(mock_post, client): + # -- mock + mock_post.return_value = Mock(status_code=400) + mock_post.return_value.json.return_value = \ + {'error': "Invalid parameters: blabla"} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.create_job('fake') + + err = err_info.value + assert 400 == err.status_code + assert err.msg.startswith('Invalid parameters') + + +@pytest.mark.success +@patch('allgo.requests.post') +def test_create_job(mock_post, client): + # -- mock + job_id = 33 + job_url = "http://job_url" + post_output = { + "avg_time": 0, # legacy, not relevant anymore + "id" : job_id, + "url" : job_url, + } + mock_post.return_value = Mock(status_code=200) + mock_post.return_value.json.return_value = post_output + + # -- tests + response = client.create_job('fake') + + assert job_id == response['id'] + assert job_url == response['url'] + + +# ------------------------------------------------ +@pytest.mark.error +@patch('allgo.requests.get') +def test_job_status__user_not_exist(mock_get, client): + # -- mock + mock_get.return_value = Mock(status_code=401) + # ~ mock_get.return_value.json.return_value = \ + # ~ {'error': 'API request without http authorisation'} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.job_status('fake_id') + + err = err_info.value + assert 401 == err.status_code + # ~ assert '' == err.msg + + +@pytest.mark.error +@patch('allgo.requests.get') +def test_job_status__error(mock_get, client): + # -- mock + mock_get.return_value = Mock(status_code=404) + mock_get.return_value.json.return_value = \ + {'error': 'Job not found'} + + # -- tests + with pytest.raises(StatusError) as err_info: + client.job_status('fake_id') + + err = err_info.value + assert 404 == err.status_code + assert 'Job not found' == err.msg + + +@pytest.mark.success +@patch('allgo.requests.get') +def test_job_status(mock_get, client): + """The method return the request response as a dictionnary.""" + # -- mock + job_id = 33 + files = { 'f1': "file1_url", 'f2': "file2_url" } + status = "bored" + mock_get.return_value = Mock(status_code=200) + mock_get.return_value.json.return_value = \ + { + job_id: files, + "status": status, + } + + # -- tests + response = client.job_status(job_id) + + assert files == response[job_id] + assert status == response['status'] + + +# ------------------------------------------------ +@pytest.mark.success +@patch('allgo.Client.create_job') +@patch('allgo.Client.job_status') +@patch('time.sleep') +def test_run_job__default(mock_sleep, mock_status, mock_create, client, capsys): + """Here, I only want to test that both methods create_job and job_status are called. + and job_status called several times. + Each call to job_status will return the next element of the side_effect list. + The Mock on sleep only intend to speed up the test.""" + # -- mock + app = 'fake' + fake_id = 6 + mock_create.return_value = { 'id': fake_id } + job_status_returns = [ + { 'status': 'new' }, + { 'status': 'waiting' }, + { 'status': 'running' }, + { 'status': 'aborting' }, + { 'status': 'done' }, + { 'status': 'never_called' }, # should not be called + ] + mock_status.side_effect = job_status_returns + + output = client.run_job(app) + + # tests + mock_create.assert_called_once() + mock_status.assert_called_with(fake_id) + + x_calls = len(job_status_returns) - 1 # -1 because of 'never_called' + assert x_calls == mock_status.call_count + + # run_job add the id to the job_status output + x_output = { 'status': 'done', 'id': fake_id } + assert x_output == output + + # without verbose, no output + assert '' == capsys.readouterr().out + + +@pytest.mark.success +@patch('allgo.Client.create_job') +@patch('allgo.Client.job_status') +@patch('time.sleep') +def test_run_job__args(mock_sleep, mock_status, mock_create, client, capsys): + """Here, I only want to test that both methods create_job and job_status are called. + and job_status called several times. + Each call to job_status will return the next element of the side_effect list.""" + # -- mock + fake_id = 6 + mock_create.return_value = { 'id': fake_id } + job_status_returns = [ + { 'status': 'new' }, + { 'status': 'waiting' }, + { 'status': 'waiting' }, + { 'status': 'waiting' }, + { 'status': 'done' }, + { 'status': 'never_called' }, # should not be called + ] + mock_status.side_effect = job_status_returns + + app = 'fake' + kwargs = { + 'version': '666', + 'params' : '7', + 'files' : ['truc', 'chose'], + } + sleep_d = 1 + client.run_job(app, sleep_duration=sleep_d, verbose=True, **kwargs) + + # tests + mock_create.assert_called_once_with(app, **kwargs) + mock_status.assert_called_with(fake_id) + mock_sleep.assert_called_with(sleep_d) + + x_calls = len(job_status_returns) - 1 # -1 because of 'never_called' + assert x_calls == mock_status.call_count + + x_output = "\nnew\t\nwaiting\t..\ndone\t\n" + assert x_output == capsys.readouterr().out + + +# ------------------------------------------------ -- GitLab