Add health service

CHANGELOG.md

@@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased (0.5.0)
+
+### Added
+* Health Service with variuos functions that allow user's to inspect the health status of the system
+### Fixed
+### Changed
+* Health status information is included when an instantiated unity object is printed.
+### Removed
+### Security
+### Deprecated
+
+
 ## [0.4.0] - 2024-03-29
 
 ### Added

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "unity-sds-client"
-version = "0.4.0"
+version = "0.5.0"
 description = "Unity-Py is a Python client to simplify interactions with NASA's Unity Platform."
 authors = ["Anil Natha, Mike Gangl"]
 readme = "README.md"

tests/test_unity_health_service.py

@@ -0,0 +1,50 @@
+"""
+This module contains a set of tests is to ensure that the
+Unity Health Service is functional.
+"""
+
+import pytest
+
+from unity_sds_client.unity import Unity
+from unity_sds_client.unity_services import UnityServices
+
+
+@pytest.mark.regression
+def test_health_service_client_creation():
+    """
+    Test that an instance of the health service can be instantiated.
+    """
+    s = Unity()
+    health_service = s.client(UnityServices.HEALTH_SERVICE)
+
+@pytest.mark.regression
+def test_health_status_retrieval():
+    """
+    Test that health statuses can be retrieved using the health service.
+    """
+    print("Example health status check")
+    s = Unity()
+    health_service = s.client(UnityServices.HEALTH_SERVICE)
+    health_statuses = health_service.get_health_status()
+    assert health_statuses is not None
+
+@pytest.mark.regression
+def test_health_status_printing():
+    """
+    Test that health statuses can be printed using the health service.
+    """
+    print("Example health status check")
+    s = Unity()
+    health_service = s.client(UnityServices.HEALTH_SERVICE)
+    health_service.print_health_status()
+
+@pytest.mark.regression
+def test_health_service_printing():
+    """
+    Test that when the health service client is printed, it outputs
+    the health status information
+    """
+    print("Example health status printing of health service object.")
+    s = Unity()
+    health_service = s.client(UnityServices.HEALTH_SERVICE)
+    print(health_service)

unity_sds_client/services/health_service.py

@@ -0,0 +1,96 @@
+from unity_sds_client.unity_session import UnitySession
+
+class HealthService(object):
+    """
+    The HealthService class is a wrapper to Unity's Health API endpoints.
+    """
+
+    def __init__(
+        self,
+        session:UnitySession
+    ):
+        """
+        Initialize the HealthService class.
+
+        Parameters
+        ----------
+        session : UnitySession
+            The Unity Session that will be used to facilitate making calls to the Health endpoints.
+
+        Returns
+        -------
+        List
+            List of applications and their health statses
+        """
+
+        self._health_statuses = None
+
+    def __str__(self):
+        return self.generate_health_status_report()
+
+    def get_health_status(self):
+        """
+        Returns a list of services and their respective health status
+        """
+
+        # Get Health Information
+        # Stubbed in health data until Health API endpoint is available
+        self._health_statuses = [
+          {
+            "service": "airflow",
+            "landingPage":"https://unity.jpl.nasa.gov/project/venue/processing/ui",
+            "healthChecks": [
+              {
+                "status": "HEALTHY",
+                "date": "2024-04-09T18:01:08Z"
+              }
+            ]
+          },
+          {
+            "service": "jupyter",
+            "landingPage":"https://unity.jpl.nasa.gov/project/venue/ads/jupyter",
+            "healthChecks": [
+              {
+                "status": "HEALTHY",
+                "date": "2024-04-09T18:01:08Z"
+              }
+            ]
+          },
+          {
+            "service": "other_service",
+            "landingPage":"https://unity.jpl.nasa.gov/project/venue/other_service",
+            "healthChecks": [
+              {
+                "status": "UNHEALTHY",
+                "date": "2024-04-09T18:01:08Z"
+              }
+            ]
+          }
+        ]
+
+        return self._health_statuses
+
+    def generate_health_status_report(self):
+        """
+        Return a generated report of health status information
+        """
+
+        if self._health_statuses is None:
+            self.get_health_status()
+
+        health_status_title = "HEALTH STATUS REPORT"
+        report = f"\n\n{health_status_title}\n"
+        report = report + len(health_status_title) * "-" + "\n\n"
+        for service in self._health_statuses:
+            report = report + f"{service["service"]}\n"
+            for status in service["healthChecks"]:
+                report = report + f"{status["date"]}: {status["status"]}\n"
+            report = report + "\n"
+
+        return report
+
+    def print_health_status(self):
+        """
+        Print the health status report
+        """
+        print(f"{self.generate_health_status_report()}")

unity_sds_client/unity.py

@@ -2,6 +2,7 @@
 from configparser import ConfigParser, ExtendedInterpolation
 from unity_sds_client.services.data_service import DataService
 from unity_sds_client.services.process_service import ProcessService
+from unity_sds_client.services.health_service import HealthService
 from unity_sds_client.unity_session import UnitySession
 from unity_sds_client.unity_exception import UnityException
 from unity_sds_client.unity_environments import UnityEnvironments
@@ -55,22 +56,26 @@ def client(self, service_name: UnityServices):
         """
         if service_name == UnityServices.DATA_SERVICE:
             return DataService(session=self._session)
+        if service_name == UnityServices.HEALTH_SERVICE:
+            return HealthService(session=self._session)
         elif service_name == UnityServices.PROCESS_SERVICE:
             return ProcessService(session=self._session)
         else:
             raise UnityException("Invalid service name: " + str(service_name))
 
     def __str__(self):
-        response = "UNITY CONFIGURATION"
-        response = response + "\n\n" + len(response) * "-" + "\n"
+        response = "\nUNITY CONFIGURATION"
+        response = response + "\n" + len(response) * "-" + "\n"
 
         config = self._session.get_config()
         config_sections = config.sections()
         for section in config_sections:
-            response = response + "\n{}\n".format(section)
+            response = response + "\n[{}]\n".format(section)
             for setting in dict(config[section]):
                 response = response + "{}: {}\n".format(setting, dict(config[section])[setting])
 
+        response = response + self.client(UnityServices.HEALTH_SERVICE).generate_health_status_report()
+
         return response
 
 

unity_sds_client/unity_services.py

@@ -6,6 +6,7 @@ class UnityServices(Enum):
     The UnityServices class is used to specify a service, when needed, when interacting with the unity_sds_client package.
     """
 
-    DATA_SERVICE = "data_service"
     APPLICATION_SERVICE = "app_service"
+    DATA_SERVICE = "data_service"
+    HEALTH_SERVICE = "health_service"
     PROCESS_SERVICE = "process_service"