add prtg util

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
 
 [project]
 name = "sat-utils"
-version = "1.5.0"
+version = "1.6.0"
 authors = [
     { name="Ryan Semmler", email="[email protected]" },
     { name="Shawn Taylor", email="[email protected]" },

sat/prtg.py

@@ -0,0 +1,56 @@
+import requests
+from requests.adapters import HTTPAdapter, Retry
+
+from sat.logs import SATLogger
+
+logger = SATLogger(name=__name__)
+
+
+class PRTGHandler:
+    def __init__(
+        self,
+        base_url: str,
+        sensor_guid: str,
+        session: requests.Session = requests.Session(),
+        retry_total: int = 5,
+    ):
+        self.base_url = base_url
+        self.sensor_guid = sensor_guid
+        self.sensor_request_string = f"{self.base_url}/{self.sensor_guid}"
+
+        self.session = session
+
+        retries = Retry(
+            total=retry_total, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504]
+        )
+
+        self.session.mount("http://", HTTPAdapter(max_retries=retries))
+
+    def metric_to_prtg(self, value: float, text: str = None) -> bool:
+        """Sends a success metric to PRTG, with an optional text field explaining the result"""
+        param_dict = {"value": value}
+
+        if text:
+            param_dict["text"] = text
+
+        response = self.session.get(self.sensor_request_string, params=param_dict)
+
+        if response.status_code != 200:
+            logger.warning(f"PRTG push failed, error code of {response.status_code}")
+            return False
+
+        return True
+
+    def error_to_prtg(self, text: str = None) -> bool:
+        """Sends an error message to PRTG for errors the application"""
+        param_dict = {}
+
+        if text:
+            param_dict["text"] = text
+
+        response = self.session.get(self.sensor_request_string, params=param_dict)
+
+        if response.status_code != 200:
+            logger.warning(f"Push to PRTG failed, error code of {response.status_code}")
+            return False
+        return True

tests/test_prtg.py

@@ -0,0 +1,86 @@
+from unittest.mock import MagicMock
+
+import pytest
+from sat.prtg import PRTGHandler
+
+
+@pytest.fixture
+def request_mocks():
+    mock_session = MagicMock()
+    mock_response = MagicMock()
+    mock_session.get.return_value = mock_response
+    return mock_session, mock_response
+
+
+def test_successful_request(request_mocks):
+    """Successful prtg response should return True"""
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 200
+
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+    result = prtg_handler.metric_to_prtg(value=1)
+
+    assert result is True
+
+
+def test_failed_request(request_mocks):
+    """
+    A failed request should return false and log a warning
+
+    A failed request is any status code not equal to 200
+    """
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 500
+
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+    result = prtg_handler.metric_to_prtg(value=1)
+
+    assert result is False
+
+
+def test_value_parameter(request_mocks):
+    """The required value parameter should be used in the request params"""
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 200
+
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+
+    result = prtg_handler.metric_to_prtg(value=1)
+
+    assert result is True
+    assert mock_session.get.call_args.kwargs["params"] == {"value": 1}
+
+
+def test_text_parameter(request_mocks):
+    """Text argument should show up in params of sent request"""
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 200
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+
+    result = prtg_handler.metric_to_prtg(value=1, text="Test text string")
+
+    assert result is True
+    assert mock_session.get.call_args.kwargs["params"] == {"text": "Test text string", "value": 1}
+
+
+def test_error_to_prtg(request_mocks):
+    """Should be able to send an error to PRTG when an exception is caught"""
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 200
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+
+    result = prtg_handler.error_to_prtg()
+
+    assert result is True
+
+
+def test_error_to_prtg_with_message(request_mocks):
+    """Should be able to send an error message to provide context to a handled error"""
+    mock_session, mock_response = request_mocks
+    mock_response.status_code = 200
+    prtg_handler = PRTGHandler("test-url", "test-guid", mock_session)
+
+    result = prtg_handler.error_to_prtg(text="Process failed because reasons")
+
+    assert result is True
+    assert mock_session.get.call_args.kwargs["params"] == {"text": "Process failed because reasons"}