update: pytest tests

Author: VsevolodX

tests/__init__.py

@@ -1,23 +0,0 @@
-import json
-import os
-import unittest
-
-
-class EndpointBaseTest(unittest.TestCase):
-    """
-    Base class for testing endpoints.
-    """
-
-    def __init__(self, *args, **kwargs):
-        super(EndpointBaseTest, self).__init__(*args, **kwargs)
-
-    def get_file_path(self, filename):
-        return os.path.join(os.path.dirname(__file__), "data", filename)
-
-    def get_content(self, filename):
-        with open(self.get_file_path(filename)) as f:
-            return f.read()
-
-    def get_content_in_json(self, filename):
-        with open(self.get_file_path(filename)) as f:
-            return json.loads(f.read())

tests/conftest.py

@@ -0,0 +1,27 @@
+"""
+Shared test fixtures and base classes for pytest.
+"""
+import json
+import os
+import unittest
+
+
+class EndpointBaseTest(unittest.TestCase):
+    """
+    Base class for testing endpoints.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(EndpointBaseTest, self).__init__(*args, **kwargs)
+
+    def get_file_path(self, filename):
+        return os.path.join(os.path.dirname(__file__), "py", "data", filename)
+
+    def get_content(self, filename):
+        with open(self.get_file_path(filename)) as f:
+            return f.read()
+
+    def get_content_in_json(self, filename):
+        with open(self.get_file_path(filename)) as f:
+            return json.loads(f.read())
+

tests/integration/__init__.py

@@ -0,0 +1,45 @@
+import os
+
+import pytest
+
+from tests.conftest import EndpointBaseTest
+
+ENV_TEST_HOST = "TEST_HOST"
+ENV_TEST_PORT = "TEST_PORT"
+ENV_TEST_ACCOUNT_ID = "TEST_ACCOUNT_ID"
+ENV_TEST_AUTH_TOKEN = "TEST_AUTH_TOKEN"
+ENV_TEST_SECURE = "TEST_SECURE"
+ENV_TEST_VERSION = "TEST_VERSION"
+DEFAULT_TEST_SECURE = "False"
+DEFAULT_TEST_VERSION = "2018-10-01"
+
+REQUIRED_ENV_VARS = [ENV_TEST_HOST, ENV_TEST_PORT, ENV_TEST_ACCOUNT_ID, ENV_TEST_AUTH_TOKEN]
+MISSING_ENV_VARS_MESSAGE = (
+    "Integration tests require environment variables: TEST_HOST, TEST_PORT, "
+    "TEST_ACCOUNT_ID, TEST_AUTH_TOKEN. Set them to run integration tests."
+)
+
+
+def _check_integration_env():
+    """Check if required integration test environment variables are set."""
+    missing = [var for var in REQUIRED_ENV_VARS if var not in os.environ]
+    if missing:
+        pytest.skip(f"{MISSING_ENV_VARS_MESSAGE} Missing: {', '.join(missing)}")
+
+
+class BaseIntegrationTest(EndpointBaseTest):
+    """
+    Base class for endpoints integration tests.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(BaseIntegrationTest, self).__init__(*args, **kwargs)
+        _check_integration_env()
+        self.endpoint_kwargs = {
+            "host": os.environ[ENV_TEST_HOST],
+            "port": os.environ[ENV_TEST_PORT],
+            "account_id": os.environ[ENV_TEST_ACCOUNT_ID],
+            "auth_token": os.environ[ENV_TEST_AUTH_TOKEN],
+            "secure": os.environ.get(ENV_TEST_SECURE, DEFAULT_TEST_SECURE).lower() == "true",
+            "version": os.environ.get(ENV_TEST_VERSION, DEFAULT_TEST_VERSION),
+        }

tests/integration/entity.py

@@ -0,0 +1,86 @@
+import time
+
+from requests.exceptions import HTTPError
+
+from tests.py.integration import BaseIntegrationTest
+
+INTEGRATION_TEST_TAG = "INTEGRATION-TEST"
+ENTITY_ID_KEY = "_id"
+ENTITY_NAME_KEY = "name"
+ENTITY_TAGS_KEY = "tags"
+TEST_NAME_PREFIX = "test-"
+UPDATED_NAME = "UPDATED"
+WARNING_ENTITY_NOT_FOUND = "WARNING: Entity with ID {} not found in the list of tagged entities: {}"
+WARNING_DELETE_FAILED = "WARNING: Failed to delete entity with ID {}: {}"
+
+
+class EntityIntegrationTest(BaseIntegrationTest):
+    """
+    Base entity integration tests class.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(EntityIntegrationTest, self).__init__(*args, **kwargs)
+        self.endpoints = None
+        self.entity_id: str = ""
+
+    def entities_selector(self):
+        """
+        Returns a selector to access entities created in tests.
+        Override upon inheritance as necessary.
+        """
+        return {ENTITY_TAGS_KEY: INTEGRATION_TEST_TAG}
+
+    def tearDown(self):
+        """Delete only the current test entity if it still exists after test.
+
+        Warn if the filtering fails, failsafe attempt to delete the entity anyways.
+        """
+        tagged_test_entity_id_list = [e[ENTITY_ID_KEY] for e in self.endpoints.list(query=self.entities_selector())]
+        try:
+            if self.entity_id not in tagged_test_entity_id_list:
+                print(WARNING_ENTITY_NOT_FOUND.format(self.entity_id, tagged_test_entity_id_list))
+            self.endpoints.delete(self.entity_id)
+
+        except HTTPError as e:
+            print(WARNING_DELETE_FAILED.format(self.entity_id, e))
+
+    def get_default_config(self):
+        """
+        Returns the default entity config.
+        Override upon inheritance.
+        """
+        raise NotImplementedError
+
+    def create_entity(self, kwargs=None):
+        entity = self.get_default_config()
+        entity.update(kwargs or {})
+        entity.setdefault(ENTITY_TAGS_KEY, []).append(INTEGRATION_TEST_TAG)
+        created_entity = self.endpoints.create(entity)
+        self.entity_id = created_entity[ENTITY_ID_KEY]
+        return created_entity
+
+    def list_entities_test(self):
+        entity = self.create_entity()
+        self.assertIn(entity[ENTITY_ID_KEY], [e[ENTITY_ID_KEY] for e in self.endpoints.list(query=self.entities_selector())])
+
+    def get_entity_by_id_test(self):
+        entity = self.create_entity()
+        self.assertEqual(self.endpoints.get(entity[ENTITY_ID_KEY])[ENTITY_ID_KEY], entity[ENTITY_ID_KEY])
+
+    def create_entity_test(self):
+        name = f"{TEST_NAME_PREFIX}{time.time()}"
+        entity = self.create_entity({ENTITY_NAME_KEY: name})
+        self.assertEqual(entity[ENTITY_NAME_KEY], name)
+        self.assertIsNotNone(entity[ENTITY_ID_KEY])
+        self.assertIn(INTEGRATION_TEST_TAG, entity[ENTITY_TAGS_KEY])
+
+    def delete_entity_test(self):
+        entity = self.create_entity()
+        self.endpoints.delete(entity[ENTITY_ID_KEY])
+        self.assertNotIn(entity[ENTITY_ID_KEY], [e[ENTITY_ID_KEY] for e in self.endpoints.list(query=self.entities_selector())])
+
+    def update_entity_test(self):
+        entity = self.create_entity()
+        updated_entity = self.endpoints.update(entity[ENTITY_ID_KEY], {ENTITY_NAME_KEY: UPDATED_NAME})
+        self.assertEqual(updated_entity[ENTITY_NAME_KEY], UPDATED_NAME)

tests/data/login.json tests/py/data/login.jsontests/data/login.json renamed to tests/py/data/login.json

@@ -2,16 +2,30 @@
 import time
 
 from exabyte_api_client.endpoints.jobs import JobEndpoints
-from tests.integration.entity import EntityIntegrationTest
+from tests.py.integration.entity import EntityIntegrationTest
+
+KNOWN_COMPLETED_JOB_ID = "9gyhfncWDhnSyzALv"
+JOB_NAME_PREFIX = "API-CLIENT TEST JOB"
+DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
+DEFAULT_NODES = 1
+DEFAULT_NOTIFY = "n"
+DEFAULT_PPN = 1
+DEFAULT_QUEUE = "D"
+DEFAULT_TIME_LIMIT = "00:05:00"
+TEST_TIME_LIMIT = "00:10:00"
+TEST_NOTIFY = "abe"
+JOB_STATUS_SUBMITTED = "submitted"
+JOB_STATUS_FINISHED = "finished"
+JOB_WAIT_TIMEOUT = 900
+JOB_WAIT_SLEEP_INTERVAL = 5
+JOB_TIMEOUT_ERROR = "job with ID {} did not finish within {} seconds"
 
 
 class JobEndpointsIntegrationTest(EntityIntegrationTest):
     """
     Job endpoints integration tests.
     """
 
-    KNOWN_COMPLETED_JOB_ID = "9gyhfncWDhnSyzALv"
-
     def __init__(self, *args, **kwargs):
         super(JobEndpointsIntegrationTest, self).__init__(*args, **kwargs)
         self.endpoints = JobEndpoints(**self.endpoint_kwargs)
@@ -21,10 +35,10 @@ def get_default_config(self):
         Returns the default entity config.
         Override upon inheritance.
         """
-        now_time = datetime.datetime.today().strftime("%Y-%m-%d %H:%M:%S")
-        return {"name": "API-CLIENT TEST JOB {}".format(now_time)}
+        now_time = datetime.datetime.today().strftime(DATETIME_FORMAT)
+        return {"name": f"{JOB_NAME_PREFIX} {now_time}"}
 
-    def get_compute_params(self, nodes=1, notify="n", ppn=1, queue="D", time_limit="00:05:00"):
+    def get_compute_params(self, nodes=DEFAULT_NODES, notify=DEFAULT_NOTIFY, ppn=DEFAULT_PPN, queue=DEFAULT_QUEUE, time_limit=DEFAULT_TIME_LIMIT):
         return {"nodes": nodes, "notify": notify, "ppn": ppn, "queue": queue, "timeLimit": time_limit}
 
     def test_list_jobs(self):
@@ -42,34 +56,33 @@ def test_delete_job(self):
     def test_update_job(self):
         self.update_entity_test()
 
-    def _wait_for_job_to_finish(self, id_, timeout=900):
+    def _wait_for_job_to_finish(self, id_, timeout=JOB_WAIT_TIMEOUT):
         end = time.time() + timeout
         while time.time() < end:
             job = self.endpoints.get(id_)
-            if job["status"] == "finished":
+            if job["status"] == JOB_STATUS_FINISHED:
                 return
-            time.sleep(5)
-        raise BaseException("job with ID {} did not finish within {} seconds".format(id_, timeout))
+            time.sleep(JOB_WAIT_SLEEP_INTERVAL)
+        raise BaseException(JOB_TIMEOUT_ERROR.format(id_, timeout))
 
     def test_submit_job_and_wait_to_finish(self):
         job = self.create_entity()
         self.endpoints.submit(job["_id"])
-        self.assertEqual("submitted", self.endpoints.get(job["_id"])["status"])
+        self.assertEqual(JOB_STATUS_SUBMITTED, self.endpoints.get(job["_id"])["status"])
         self._wait_for_job_to_finish(job["_id"])
 
     def test_create_job_timeLimit(self):
-        time_limit = "00:10:00"
-        job = self.create_entity({"compute": self.get_compute_params(time_limit=time_limit)})
+        job = self.create_entity({"compute": self.get_compute_params(time_limit=TEST_TIME_LIMIT)})
         self.assertEqual(self.endpoints.get(job["_id"])["_id"], job["_id"])
-        self.assertEqual(self.endpoints.get(job["_id"])["compute"]["timeLimit"], time_limit)
+        self.assertEqual(self.endpoints.get(job["_id"])["compute"]["timeLimit"], TEST_TIME_LIMIT)
 
     def test_create_job_notify(self):
-        job = self.create_entity({"compute": self.get_compute_params(notify="abe")})
+        job = self.create_entity({"compute": self.get_compute_params(notify=TEST_NOTIFY)})
         self.assertEqual(self.endpoints.get(job["_id"])["_id"], job["_id"])
-        self.assertEqual(self.endpoints.get(job["_id"])["compute"]["notify"], "abe")
+        self.assertEqual(self.endpoints.get(job["_id"])["compute"]["notify"], TEST_NOTIFY)
 
     def test_list_files(self):
-        http_response_data = self.endpoints.list_files(self.KNOWN_COMPLETED_JOB_ID)
+        http_response_data = self.endpoints.list_files(KNOWN_COMPLETED_JOB_ID)
         self.assertIsInstance(http_response_data, list)
         self.assertGreater(len(http_response_data), 0)
         self.assertIsInstance(http_response_data[0], dict)