[#543] Add base tests for EDI

legion/legion/external/edi.py

@@ -35,7 +35,7 @@ class EdiClient:
     EDI client
     """
 
-    def __init__(self, base=None, token=None, retries=3, http_client=None, use_relative_url=False):
+    def __init__(self, base, token=None, retries=3):
         """
         Build client
 
@@ -54,11 +54,24 @@ def __init__(self, base=None, token=None, retries=3, http_client=None, use_relat
         self._token = token
         self._version = legion.edi.server.EDI_VERSION
         self._retries = retries
-        if http_client:
-            self._http_client = http_client
-        else:
-            self._http_client = requests
-        self._use_relative_url = use_relative_url
+
+
+    def _request(self, action, url, data=None, headers=None):
+        """
+        Make HTTP request
+
+        :param action: request action, e.g. get / post / delete
+        :type action: str
+        :param url: target URL
+        :type url: str
+        :param data: (Optional) data to be placed in body of request
+        :param data: dict[str, str] or None
+        :param headers: (Optional) additional HTTP headers
+        :param headers: dict[str, str] or None
+        :return: :py:class:`requests.Response` -- response
+        """
+        return requests.request(action.lower(), url, data=data, headers=headers)
+
 
     def _query(self, url_template, payload=None, action='GET'):
         """
@@ -73,20 +86,16 @@ def _query(self, url_template, payload=None, action='GET'):
         :return: dict[str, any] -- response content
         """
         sub_url = url_template.format(version=self._version)
-        if self._use_relative_url:
-            target_url = sub_url
-        else:
-            target_url = self._base.strip('/') + sub_url
-
-        headers = {}
+        target_url = self._base.strip('/') + sub_url
 
         left_retries = self._retries if self._retries > 0 else 1
         raised_exception = None
 
         while left_retries > 0:
             try:
-                LOGGER.debug('Requesting {}'.format(full_url))
-                response = requests.request(action.lower(), full_url, data=payload, headers=headers, auth=auth)
+                LOGGER.debug('Requesting {}'.format(target_url))
+                # TODO: Add sending token (LEGION #313)
+                response = self._request(action, target_url, payload)
             except requests.exceptions.ConnectionError as exception:
                 LOGGER.error('Failed to connect to {}: {}. Retrying'.format(self._base, exception))
                 raised_exception = exception
@@ -100,20 +109,12 @@ def _query(self, url_template, payload=None, action='GET'):
                 self._base, raised_exception
             ))
 
-        if hasattr(response, 'text'):
-            response_data = response.text
-        else:
-            response_data = response.data
-
-        if isinstance(response_data, bytes):
-            response_data = response_data.decode('utf-8')
-
         try:
-            answer = json.loads(response_data)
+            answer = json.loads(response.text)
             LOGGER.debug('Got answer: {!r} with code {} for URL {!r}'
                          .format(answer, response.status_code, target_url))
         except ValueError as json_decode_exception:
-            raise ValueError('Invalid JSON structure {!r}: {}'.format(response_data, json_decode_exception))
+            raise ValueError('Invalid JSON structure {!r}: {}'.format(response.text, json_decode_exception))
 
         if isinstance(answer, dict) and answer.get('error', False):
             exception = answer.get('exception')

legion/tests/legion_test_utils.py

@@ -13,6 +13,7 @@
 import docker.types
 import docker.errors
 import docker.client
+import requests
 
 import legion.config
 import legion.containers.docker
@@ -506,6 +507,41 @@ def __exit__(self, *args, exception=None):
             raise exception
 
 
+def build_requests_reponse_from_flask_test_response(test_response, url):
+    """
+    Build requests.Response object from Flask test client response
+
+    :param test_response: Flask test client response
+    :type test_response: :py:class:`flask.wrappers.Response`
+    :param url: requested URL
+    :type url: str
+    :return: :py:class:`requests.Response` -- response object
+    """
+    response = requests.Response()
+    response.status_code = test_response.status_code
+    response.url = url
+    response._content = test_response.data
+    for header, value in test_response.headers.items():
+        response.headers[header] = value
+    response._encoding = test_response.charset
+    response._content_consumed = True
+    return response
+
+
+def build_requests_mock_function(test_client):
+    """
+    Build function that shoul replace requests.request function in tests
+
+    :param test_client: test flask client
+    :type test_client: :py:class:`flask.test.FlaskClient`
+    :return: Callable[[str, str, dict[str, str], dict[str, str]], requests.Response]
+    """
+    def func(action, url, data=None, headers=None):
+        test_response = test_client.open(url, method=action, data=data, headers=headers)
+        return build_requests_reponse_from_flask_test_response(test_response, url)
+    return func
+
+
 class EDITestServer:
     """
     Context manager for testing EDI server
@@ -519,7 +555,6 @@ def __init__(self, enclave_name='debug-enclave'):
 
         self.application = None
         self.http_client = None
-        self.edi_client = None
 
     def __enter__(self):
         """
@@ -535,15 +570,16 @@ def __enter__(self):
         test_enclave = legion.k8s.enclave.Enclave(self._enclave_name)
         test_enclave._data_loaded = True
 
-        with patch('legion.k8s.get_current_namespace', lambda *x: self._enclave_name):
-            with patch('legion.edi.server.get_application_enclave', lambda *x: test_enclave):
-                with patch('legion.edi.server.get_application_grafana', lambda *x: None):
-                    with patch_environ(additional_environment):
-                        self.application = ediserve.init_application(None)
-                        self.application.testing = True
-                        self.http_client = self.application.test_client()
-                        self.edi_client = legion.external.edi.EdiClient(http_client=self.http_client,
-                                                                        use_relative_url=True)
+        with patch('legion.k8s.get_current_namespace', lambda *x: self._enclave_name), \
+             patch('legion.edi.server.get_application_enclave', lambda *x: test_enclave), \
+             patch('legion.edi.server.get_application_grafana', lambda *x: None), \
+             patch_environ(additional_environment):
+            self.application = ediserve.init_application(None)
+
+        self.application.testing = True
+        self.http_client = self.application.test_client()
+        self.edi_client = legion.external.edi.EdiClient('')
+        self.edi_client._request = build_requests_mock_function(self.http_client)
 
         return self
 

legion/tests/test_edi.py

@@ -37,15 +37,29 @@
 import legion.serving.pyserve as pyserve
 
 
+def get_models_mock_empty(model_id, model_version):
+    return []
+
 class TestEDI(unittest2.TestCase):
-    def test_model_info_with_typed_columns(self):
+    def test_edi_inspect_all_empty(self):
         with EDITestServer() as edi:
-            with unittest.mock.patch('legion.k8s.enclave.Enclave.get_models', side_effect=lambda *x, **y: []) as get_models_patched:
+            with unittest.mock.patch('legion.k8s.enclave.Enclave.get_models', side_effect=get_models_mock_empty) as get_models_patched:
                 models_info = edi.edi_client.inspect()
                 self.assertIsInstance(models_info, list)
                 self.assertEqual(len(models_info), 0)
                 self.assertTrue(len(get_models_patched.call_args_list) == 1, 'should be exactly one call')
 
+    def test_edi_inspect_model_id_and_version_empty(self):
+        TEST_MODEL_ID = 'test-id'
+        TEST_MODEL_VERSION = 'test-version'
+
+        with EDITestServer() as edi:
+            with unittest.mock.patch('legion.k8s.enclave.Enclave.get_models', side_effect=get_models_mock_empty) as get_models_patched:
+                models_info = edi.edi_client.inspect(TEST_MODEL_ID, TEST_MODEL_VERSION)
+                self.assertIsInstance(models_info, list)
+                self.assertEqual(len(models_info), 0)
+                self.assertTrue(len(get_models_patched.call_args_list) == 1, 'should be exactly one call')
+                self.assertTupleEqual(get_models_patched.call_args_list[0][0], (TEST_MODEL_ID, TEST_MODEL_VERSION))
 
 
 if __name__ == '__main__':