Some unit tests for the module.

pytest.ini

+[pytest]
+markers =
+    cstr: related to constructor
+    getter: test getters
+    error: test when an error occured
+    success: test output on successfull request
+    dbg: for debug.

requirements.txt

+pytest
+sphinx
+nbsphinx
+pandoc

tests/test_allgo.py

+#! /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
+
+
+# ------------------------------------------------