integration tests

tests/test_integration.py

+#! /usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# standard lib
+import os
+
+# not standard
+import pytest
+
+# local import
+from allgo import *
+
+
+# ==========================================================
+class ApiClientIntegration:
+    CLIENT = None
+    TEST_APP = 'allgo api test'
+
+    @classmethod
+    def setup_class(cls):
+        """Create a client object, and set it as attribute."""
+        raise NotImplementedError("This class has to be derived to be used.")
+
+    @classmethod
+    def _build_client(cls):
+        raise NotImplementedError
+
+    @classmethod
+    def _get_job_id(cls):
+        raise NotImplementedError
+
+    @pytest.mark.error
+    def test_create_job__user_not_exist(self):
+        """create a job with a token that refers to no one.
+            We test that we get an error 401."""
+        # The token should refer to unexisting user
+        client = self._build_client(token='0')
+
+        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
+    def test_create_job__app_not_exist(self):
+        """I try to create a job with an app that cannot exist.
+            Its name is illegal, an error 404 should be raised."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job('ç')
+
+        err = err_info.value
+        assert 404 == err.status_code
+        # ~ assert 'Application not found' == err.msg
+
+    @pytest.mark.skip(reason="I don't want to create another fake app. \
+                              Wait for version to be considered.")
+    @pytest.mark.error
+    def test_create_job__app_not_published(self):
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job('not_published')
+
+        err = err_info.value
+        assert 404 == err.status_code
+        # ~ assert 'not yet published' in err.msg
+
+    @pytest.mark.error
+    def test_create_job__queue_not_exist(self):
+        """create a job to run on a queue that does not exist.
+            Should raise an error 400."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job(self.TEST_APP, queue='no_queue')
+
+        err = err_info.value
+        assert 400 == err.status_code
+        # ~ assert 'Unknown queue' == err.msg
+
+    @pytest.mark.error
+    def test_create_job__param_error(self):
+        """create a job with an invalid param.
+            should raise an error 400."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job(self.TEST_APP, params='$')
+
+        err = err_info.value
+        assert 400 == err.status_code
+        # ~ assert err.msg.startswith('Invalid parameters')
+
+    @pytest.mark.success
+    def test_create_job(self):
+        """create a job that will succeed.
+            response should be a dict with at least keys 'id' and 'url'."""
+        response = self.CLIENT.create_job(self.TEST_APP)
+
+        assert isinstance(response, dict)
+        assert 'id'  in response
+        assert 'url' in response
+
+    # I think this test is useless as the failure is not related to the submission.
+    @pytest.mark.success
+    def test_create_job__that_will_fail(self):
+        """create a job that will fail (the job, not its submission).
+            response should be a dict with at least keys 'id' and 'url'."""
+        response = self.CLIENT.create_job(self.TEST_APP, 1)
+
+        assert isinstance(response, dict)
+        assert 'id'  in response
+        assert 'url' in response
+
+    @pytest.mark.error
+    def test_job_status__do_not_exist(self):
+        """try to get status of a job that does not exist.
+            should raise an error 404."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.job_status('fake_id')  # not found, as job id should be integer
+
+        err = err_info.value
+        assert 404 == err.status_code
+        # ~ assert 'Job not found' == err.msg
+
+    @pytest.mark.success
+    def test_job_status(self):
+        """create a job and get its status,
+            the method should return the request response as a dictionnary.
+            it should contain the 'status' key and the job id as key."""
+        response = self.CLIENT.create_job(self.TEST_APP)
+        job_id   = response['id']
+        response = self.CLIENT.job_status(job_id)
+
+        assert isinstance(response, dict)
+        assert str(job_id) in response
+        assert 'status'    in response
+
+    @pytest.mark.dbg
+    @pytest.mark.success
+    def test_download_file(self, tmp_path):
+        """Get the job status of an existing job, and try to download the log file.
+            The file should have been written in a given directory.
+            The content file is also checked.
+
+            The job queried exists and belongs to the testing user.
+        """
+        JOB_ID = self._get_job_id()
+        response = self.CLIENT.job_status(JOB_ID)
+
+        file_url = response[str(JOB_ID)]['allgo.log']
+        filepath = self.CLIENT.download_file(file_url, outdir=tmp_path)
+
+        assert filepath.startswith(str(tmp_path))
+
+        content = ""
+        with open(filepath, "r") as file_out:
+            content = file_out.read()
+        assert content.endswith("====  ALLGO JOB SUCCESS  ====\n")
+
+        # cleaning
+        os.remove(filepath)
+
+
+# ==========================================================
+class TestAllgo18(ApiClientIntegration):
+
+    @classmethod
+    def _build_client(cls, token):
+        return Client(token)
+
+    @classmethod
+    def _get_job_id(cls):
+        """return a job id that exist, with a log file named allgo.log"""
+        return 874
+
+    def setup_class(cls):
+        """define the client used in tests.
+            It uses a token defined in as env variable."""
+        TOKEN = os.environ.get('ALLGO_18_TOKEN')
+        cls.CLIENT = cls._build_client(TOKEN)
+
+
+@pytest.mark.skipif(not os.environ.get('DEV'), reason="No dev test in CI.")
+class TestAllgoDev(ApiClientIntegration):
+    TEST_APP = 'Allgo Api Test'
+
+    @classmethod
+    def _build_client(cls, token):
+        ALLGO_URL = 'https://localhost'
+
+        return Client(token, allgo_url=ALLGO_URL, verify_tls=False)
+
+    @classmethod
+    def _get_job_id(cls):
+        """return a job id that exist, with a log file named allgo.log"""
+        return 37
+
+    @classmethod
+    def setup_class(cls):
+        """define the client used in local instance.
+            It uses a token defined in as env variable."""
+        TOKEN = os.environ.get('ALLGO_DEV_TOKEN')
+        cls.CLIENT = cls._build_client(TOKEN)
+
+    # should be removed when a global solution will be found.
+    @pytest.mark.xfail(reason="published state is not yet implemented.")
+    @pytest.mark.error
+    def test_create_job__app_not_published(self):
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job('not_published')
+
+        err = err_info.value
+        assert 404 == err.status_code
+        # ~ assert 'not yet published' in err.msg
+
+
+class TestAllgoOld(ApiClientIntegration):
+    TEST_APP  = 206
+
+    @classmethod
+    def _build_client(cls, token):
+        ALLGO_URL = 'https://allgo.inria.fr'
+
+        return Client(token, allgo_url=ALLGO_URL)
+
+    @classmethod
+    def _get_job_id(cls):
+        """return a job id that exist, with a log file named allgo.log"""
+        return 71771
+
+    def setup_class(cls):
+        """define the client used in local instance.
+            It uses a token defined in as env variable."""
+        TOKEN = os.environ.get('ALLGO_OLD_TOKEN')
+        cls.CLIENT = cls._build_client(TOKEN)
+
+    # behaviour changed between old and new.
+    @pytest.mark.error
+    def test_create_job__queue_not_exist(self):
+        """create a job to run on a queue that does not exist.
+            Should raise an error 404 (code error changed between old and new version)."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job(self.TEST_APP, queue='no_queue')
+
+        err = err_info.value
+        assert 404 == err.status_code
+        # ~ assert 'Unknown queue' == err.msg
+
+    # behaviour changed between old and new.
+    # I didn't find the source that emit the error in rails code.
+    @pytest.mark.error
+    def test_create_job__param_error(self):
+        """create a job with an invalid param.
+            should raise an error 422 (code error changed between old and new version)."""
+        with pytest.raises(StatusError) as err_info:
+            self.CLIENT.create_job(self.TEST_APP, params='$')
+
+        err = err_info.value
+        assert 422 == err.status_code
+        # ~ assert err.msg.startswith('Invalid parameters')
+
+    # behaviour changed between old and new.
+    @pytest.mark.success
+    def test_job_status(self):
+        """create a job and get its status,
+            the method should return the request response as a dictionnary.
+            it should contain only the 'status' key."""
+        response = self.CLIENT.create_job(self.TEST_APP)
+        job_id   = response['id']
+        response = self.CLIENT.job_status(job_id)
+
+        assert isinstance(response, dict)
+        assert 'status'    in response