feat(alert): Add notify to Alert class that returns notified channels

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-22.04
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v3

CHANGES

@@ -1,6 +1,12 @@
+1.23.0
+-------
+- Add notify to Alert class that returns notified channels
+- Add support for Python 3.13
+- Drop support for Python 3.8
+
 1.22.2
 -------
-- enforce providing an environemnt for alerts and incidents
+- enforce providing an environment for alerts and incidents
 
 1.22.1
 -------

intezer_sdk/__init__.py

@@ -1 +1 @@
-__version__ = '1.22.2'
+__version__ = '1.23.0'

intezer_sdk/_api.py

@@ -28,6 +28,9 @@ def __init__(self, api: IntezerApiClient):
     def on_premise_version(self) -> Optional[OnPremiseVersion]:
         return self.api.on_premise_version
 
+    def _get_tenant_id(self) -> Optional[str]:
+        return os.environ.get('INTEZER_TENANT_ID')
+
     def analyze_by_hash(self,
                         file_hash: str,
                         disable_dynamic_unpacking: Optional[bool],
@@ -699,6 +702,10 @@ def get_alerts_by_alert_ids(self, alert_ids: List[str], environments: List[str]
         data = dict(alert_ids=alert_ids)
         if environments:
             data['environments'] = environments
+
+        if tenant_id := self._get_tenant_id():
+            data['tenant_id'] = tenant_id
+
         response = self.api.request_with_refresh_expired_access_token(method='GET',
                                                                       path='/alerts/search',
                                                                       data=data)
@@ -718,6 +725,10 @@ def get_alert_by_alert_id(self, alert_id: str, environment: Optional[str] = None
         data = dict(alert_id=alert_id)
         if environment:
             data['environment'] = environment
+
+        if tenant_id := self._get_tenant_id():
+            data['tenant_id'] = tenant_id
+
         response = self.api.request_with_refresh_expired_access_token(method='GET',
                                                                       path='/alerts/get-by-id',
                                                                       data=data)
@@ -748,6 +759,32 @@ def get_device_by_id(self, device_id: str) -> dict:
 
         return data_response['result']
 
+    def notify_alert(self, alert_id: str, environment: Optional[str] = None) -> dict:
+        """
+        Send a notification for an alert.
+
+        :param alert_id: The alert id to notify.
+        :param environment: The environment of the alert.
+        :raises: :class:`requests.HTTPError` if the request failed for any reason.
+        :return: The notification response containing notified_channels.
+        """
+        self.assert_any_on_premise('notify-alert')
+
+        data = {}
+        if environment:
+            data['environment'] = environment
+
+        if tenant_id := self._get_tenant_id():
+            data['tenant_id'] = tenant_id
+
+        response = self.api.request_with_refresh_expired_access_token(
+            method='POST',
+            path=f'/alerts/{alert_id}/notify',
+            data=data if data else None
+        )
+        raise_for_status(response)
+        return response.json()
+
     def get_index_response(self, index_id: str) -> Response:
         """
         Get the index response by its id.
@@ -791,7 +828,10 @@ def get_raw_alert_data(
         :return: The raw alert data.
         """
 
-        data = {"environment": environment, "raw_data_type": raw_data_type}
+        data = {'environment': environment, 'raw_data_type': raw_data_type}
+
+        if tenant_id := self._get_tenant_id():
+            data['tenant_id'] = tenant_id
 
         response = self.api.request_with_refresh_expired_access_token(
             method='GET',

intezer_sdk/alerts.py

@@ -26,7 +26,6 @@
 from intezer_sdk.endpoint_analysis import EndpointAnalysis
 from intezer_sdk.util import add_filter
 
-
 DEFAULT_LIMIT = 100
 DEFAULT_OFFSET = 0
 ALERTS_SEARCH_REQUEST = '/alerts/search'
@@ -556,8 +555,8 @@ def _fetch_scan(scan_: dict,
             elif scan_type == 'url':
                 _fetch_scan(scan, 'url_analysis', UrlAnalysis)
 
-    def get_raw_data(self, 
-                     environment: Optional[str] = None, 
+    def get_raw_data(self,
+                     environment: Optional[str] = None,
                      raw_data_type: str = 'raw_alert') -> dict:
         """
         Get raw alert data.
@@ -568,9 +567,26 @@ def get_raw_data(self,
         """
         if not environment and not self.environment:
             raise ValueError('Environment is required to get raw data.')
-        
+
         return self._api.get_raw_alert_data(
             alert_id=self.alert_id,
             environment=environment or self.environment,
             raw_data_type=raw_data_type
         )
+
+    def notify(self) -> List[str]:
+        """
+        Send a notification for this alert.
+
+        :raises intezer_sdk.errors.AlertNotFoundError: If the alert was not found.
+        :raises intezer_sdk.errors.AlertInProgressError: If the alert is still being processed.
+        :raises: :class:`requests.HTTPError` if the request failed for any reason.
+        :return: List of notified channels.
+        """
+        if self.status == AlertStatusCode.NOT_FOUND:
+            raise errors.AlertNotFoundError(self.alert_id)
+        elif self.status == AlertStatusCode.IN_PROGRESS:
+            raise errors.AlertInProgressError(self.alert_id)
+
+        response = self._api.notify_alert(self.alert_id, self.environment)
+        return response.get('notified_channels', [])

setup.py

@@ -36,15 +36,16 @@ def rel(*xs):
     ],
     keywords='intezer',
     tests_requires=[
-        'responses == 0.25.0',
-        'pytest == 8.0.1'
+        'responses == 0.25.8',
+        'pytest == 8.1.1'
     ],
-    python_requires='!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,!=3.7.*',
+    python_requires='!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,!=3.7.*,!=3.8.*',
     classifiers=[
         'License :: OSI Approved :: Apache Software License',
-        'Programming Language :: Python :: 3.8',
         'Programming Language :: Python :: 3.9',
         'Programming Language :: Python :: 3.10',
         'Programming Language :: Python :: 3.11',
-        'Programming Language :: Python :: 3.12']
+        'Programming Language :: Python :: 3.12',
+        'Programming Language :: Python :: 3.13',
+    ]
 )

test_requirements.txt

@@ -1,4 +1,4 @@
 # This is not used by the project, but is used by the CI/CD pipeline to install dependencies, update setup.py for package dependencies.
 requests>=2.29.0,<3
-responses==0.25.0
-pytest==8.1.1
+responses==0.25.8
+pytest==8.4.1

tests/unit/test_alerts.py

@@ -4,8 +4,10 @@
 
 import responses
 
+from intezer_sdk import errors
 from intezer_sdk.alerts import Alert
 from intezer_sdk.alerts import get_alerts_by_alert_ids
+from intezer_sdk.consts import AlertStatusCode
 from tests.unit.base_test import BaseTest
 from tests.utils import load_binary_file_from_resources
 
@@ -141,17 +143,76 @@ def test_get_raw_alert_data(self):
             'result_url': 'https://example.com/download/alert-data',
             'metadata': {'environment': environment, 'raw_data_type': 'raw_alert'}
         }
-        
+
         with responses.RequestsMock() as mock:
             mock.add('GET',
                      url=f'{self.full_url}/alerts/{alert_id}/raw-data',
                      json=expected_raw_data,
                      status=HTTPStatus.OK)
-            
+
             alert = Alert(alert_id=alert_id)
-            
+
             # Act
             result_data = alert.get_raw_data(environment=environment)
-            
+
             # Assert
             self.assertEqual(result_data, expected_raw_data)
+
+    def test_alert_notify_success(self):
+        # Arrange
+        alert_id = 'test_alert_id'
+        expected_channels = ['channel-123-456', 'channel-789-012']
+        with responses.RequestsMock() as mock:
+            mock.add('GET',
+                     url=f'{self.full_url}/alerts/get-by-id',
+                     status=HTTPStatus.OK,
+                     json={'result': {'environment': 'environment'}, 'status': 'success'})
+            mock.add('POST',
+                     url=f'{self.full_url}/alerts/{alert_id}/notify',
+                     status=HTTPStatus.OK,
+                     json={'notified_channels': expected_channels})
+
+            # Act
+            alert = Alert.from_id(alert_id)
+            notified_channels = alert.notify()
+
+            # Assert
+            self.assertEqual(notified_channels, expected_channels)
+
+    def test_alert_notify_returns_empty_list_when_no_channels_in_response(self):
+        # Arrange
+        alert_id = 'test_alert_id'
+        with responses.RequestsMock() as mock:
+            mock.add('GET',
+                     url=f'{self.full_url}/alerts/get-by-id',
+                     status=HTTPStatus.OK,
+                     json={'result': {'environment': 'environment'}, 'status': 'success'})
+            mock.add('POST',
+                     url=f'{self.full_url}/alerts/{alert_id}/notify',
+                     status=HTTPStatus.OK,
+                     json={'result': True})
+
+            # Act
+            alert = Alert.from_id(alert_id)
+            notified_channels = alert.notify()
+
+            # Assert
+            self.assertEqual(notified_channels, [])
+
+    def test_alert_notify_raises_alert_not_found_error_when_alert_not_found(self):
+        # Arrange
+        alert = Alert(alert_id='test_alert_id')
+        alert.status = AlertStatusCode.NOT_FOUND
+
+        # Act & Assert
+        with self.assertRaises(errors.AlertNotFoundError):
+            alert.notify()
+
+    def test_alert_notify_raises_alert_in_progress_error_when_alert_in_progress(self):
+        # Arrange
+        alert = Alert(alert_id='test_alert_id')
+        alert.status = AlertStatusCode.IN_PROGRESS
+
+        # Act & Assert
+        with self.assertRaises(errors.AlertInProgressError):
+            alert.notify()