Add teams notification feature - Enabling push notifications to teams…

… channel (#57)

* Added the user_config variables for the Teams notification

* Added teams.py, made changes for get and set methods for teams variables and also added exception for teams

* changed features.png - added Teams in notification part
added teams_plugin.md, made changes in se_flow_and_feature.pptx

* Added test documents in the subsequent folder and ran successfully

* Added pytest for the changes done

---------

Co-authored-by: BharatSahitya <[email protected]>

docs/api/teams_plugin.md

@@ -0,0 +1,11 @@
+---
+search:
+  exclude: true
+---
+
+::: spark_expectations.notifications.plugins.teams
+    handler: python
+    options:
+        filters:
+            - "!^_[^_]"
+            - "!^__[^__]"

spark_expectations/config/user_config.py

@@ -23,6 +23,12 @@ class Constants:
         "spark.expectations.notifications.slack.webhook_url"
     )
 
+    # declare const user config variables for teams notification
+    se_notifications_enable_teams = "spark.expectations.notifications.teams.enabled"
+    se_notifications_teams_webhook_url = (
+        "spark.expectations.notifications.teams.webhook_url"
+    )
+
     se_notifications_on_start = "spark.expectations.notifications.on_start"
     se_notifications_on_completion = "spark.expectations.notifications.on.completion"
     se_notifications_on_fail = "spark.expectations.notifications.on.fail"

spark_expectations/core/context.py

@@ -52,6 +52,9 @@ def __post_init__(self) -> None:
         self._enable_slack: bool = False
         self._slack_webhook_url: Optional[str] = None
 
+        self._enable_teams: bool = False
+        self._teams_webhook_url: Optional[str] = None
+
         self._table_name: Optional[str] = None
         self._input_count: int = 0
         self._error_count: int = 0
@@ -538,6 +541,45 @@ def get_slack_webhook_url(self) -> str:
             accessing it"""
         )
 
+    def set_enable_teams(self, enable_teams: bool) -> None:
+        """
+
+        Args:
+            enable_teams:
+
+        Returns:
+
+        """
+        self._enable_teams = enable_teams
+
+    @property
+    def get_enable_teams(self) -> bool:
+        """
+        This function returns whether to enable teams notification or not
+        Returns: Returns _enable_teams(bool)
+
+        """
+        return self._enable_teams
+
+    def set_teams_webhook_url(self, teams_webhook_url: str) -> None:
+        self._teams_webhook_url = teams_webhook_url
+
+    @property
+    def get_teams_webhook_url(self) -> str:
+        """
+        This function returns sack webhook url
+        Returns:
+            str: Returns _webhook_url(str)
+
+        """
+
+        if self._teams_webhook_url:
+            return self._teams_webhook_url
+        raise SparkExpectationsMiscException(
+            """The spark expectations context is not set completely, please assign '_teams_webhook_url' before 
+            accessing it"""
+        )
+
     def set_table_name(self, table_name: str) -> None:
         self._table_name = table_name
 

spark_expectations/core/exceptions.py

@@ -49,6 +49,14 @@ class SparkExpectationsSlackNotificationException(Exception):
     pass
 
 
+class SparkExpectationsTeamsNotificationException(Exception):
+    """
+    Throw this exception when spark expectations encounters miscellaneous exceptions
+    """
+
+    pass
+
+
 class SparkExpectationsEmailException(Exception):
     """
     Throw this exception when spark expectations encounters miscellaneous exceptions

spark_expectations/notifications/__init__.py

@@ -12,6 +12,9 @@
 from spark_expectations.notifications.plugins.slack import (
     SparkExpectationsSlackPluginImpl,
 )
+from spark_expectations.notifications.plugins.teams import (
+    SparkExpectationsTeamsPluginImpl,
+)
 
 
 @functools.lru_cache
@@ -30,6 +33,9 @@ def get_notifications_hook() -> pluggy.PluginManager:
     pm.register(
         SparkExpectationsSlackPluginImpl(), "spark_expectations_slack_notification"
     )
+    pm.register(
+        SparkExpectationsTeamsPluginImpl(), "spark_expectations_teams_notification"
+    )
     for name, plugin_instance in pm.list_name_plugin():
         _log.info(
             "Loaded plugin with name: %s and class: %s",

spark_expectations/notifications/plugins/teams.py

@@ -0,0 +1,64 @@
+from typing import Dict, Union
+import requests
+from spark_expectations import _log
+from spark_expectations.notifications.plugins.base_notification import (
+    SparkExpectationsNotification,
+    spark_expectations_notification_impl,
+)
+from spark_expectations.core.exceptions import (
+    SparkExpectationsTeamsNotificationException,
+)
+from spark_expectations.core.context import SparkExpectationsContext
+
+
+class SparkExpectationsTeamsPluginImpl(SparkExpectationsNotification):
+    """
+    This class implements/supports functionality to send teams notification
+    """
+
+    @spark_expectations_notification_impl
+    def send_notification(
+        self,
+        _context: SparkExpectationsContext,
+        _config_args: Dict[Union[str], Union[str, bool]],
+    ) -> None:
+        """
+        function to send the teams notification for assigned channel
+        Args:
+            _context: SparkExpectationsContext class object
+            _config_args: dict
+
+        Returns: None
+
+        """
+        try:
+            if _context.get_enable_teams is True:
+                # payload = {"token": "{token}", "channel": kwargs['channel'], "text": kwargs['message']}
+
+                message = _config_args.get("message")
+
+                # Format Message for Teams
+                if isinstance(message, str):
+                    message = message.replace("\n", "\n\n").replace("            ", "")
+
+                payload = {
+                    "title": "SE Notification",
+                    "themeColor": "008000",
+                    "text": message,
+                }
+
+                response = requests.post(
+                    _context.get_teams_webhook_url, json=payload, timeout=10
+                )
+
+                # Check the response for success or failure
+                if response.status_code == 200:
+                    _log.info("Message posted successfully!")
+                else:
+                    _log.info("Failed to post message")
+                    raise SparkExpectationsTeamsNotificationException(
+                        "error occurred while sending teams notification from spark expectations project"
+                    )
+
+        except Exception as e:
+            raise SparkExpectationsTeamsNotificationException(e)

spark_expectations/utils/reader.py

@@ -39,6 +39,8 @@ def set_notification_param(
                 user_config.se_notifications_email_subject: "spark-expectations-testing",
                 user_config.se_notifications_enable_slack: False,
                 user_config.se_notifications_slack_webhook_url: "",
+                user_config.se_notifications_enable_teams: False,
+                user_config.se_notifications_teams_webhook_url: "",
             }
 
             _notification_dict: Dict[str, Union[str, int, bool]] = (
@@ -112,6 +114,21 @@ def set_notification_param(
                         "All params/variables required for slack notification is not configured or supplied"
                     )
 
+            if _notification_dict[user_config.se_notifications_enable_teams] is True:
+                if _notification_dict[user_config.se_notifications_teams_webhook_url]:
+                    self._context.set_enable_teams(True)
+                    self._context.set_teams_webhook_url(
+                        str(
+                            _notification_dict[
+                                user_config.se_notifications_teams_webhook_url
+                            ]
+                        )
+                    )
+                else:
+                    raise SparkExpectationsMiscException(
+                        "All params/variables required for slack notification is not configured or supplied"
+                    )
+
         except Exception as e:
             raise SparkExpectationsMiscException(
                 f"error occurred while reading notification configurations {e}"

tests/core/test_context.py

@@ -49,6 +49,8 @@ def test_context_properties():
     context._mail_subject = "spark expectations"
     context._enable_slack = True
     context._slack_webhook_url = "abcedfghi"
+    context._enable_teams = True
+    context._teams_webhook_url = "abcedfghi"
     context._table_name = "test_table"
     context._input_count = 100
     context._error_count = 10
@@ -146,6 +148,8 @@ def test_context_properties():
     assert context._mail_subject == "spark expectations"
     assert context._enable_slack is True
     assert context._slack_webhook_url == "abcedfghi"
+    assert context._enable_teams is True
+    assert context._teams_webhook_url == "abcedfghi"
     assert context._table_name == "test_table"
     assert context._input_count == 100
     assert context._error_count == 10
@@ -461,6 +465,20 @@ def test_set_slack_webhook_url():
     assert context.get_slack_webhook_url == "abcdefghi"
 
 
+def test_set_enable_teams():
+    context = SparkExpectationsContext(product_id="product1", spark=spark)
+    context.set_enable_teams(True)
+    assert context._enable_teams is True
+    assert context.get_enable_teams is True
+
+
+def test_set_teams_webhook_url():
+    context = SparkExpectationsContext(product_id="product1", spark=spark)
+    context.set_teams_webhook_url("abcdefghi")
+    assert context._teams_webhook_url == "abcdefghi"
+    assert context.get_teams_webhook_url == "abcdefghi"
+
+
 def test_table_name():
     context = SparkExpectationsContext(product_id="product1", spark=spark)
     context.set_table_name("test_table")
@@ -578,6 +596,16 @@ def test_get_slack_webhook_url_exception():
         context.get_slack_webhook_url
 
 
+def test_get_teams_webhook_url_exception():
+    context = SparkExpectationsContext(product_id="product1", spark=spark)
+    context._teams_webhook_url = False
+    with pytest.raises(SparkExpectationsMiscException,
+                       match="The spark expectations context is not set completely, please assign "
+                             "'_teams_webhook_url' before \n            accessing it"):
+        context.get_teams_webhook_url
+
+
+
 def test_get_table_name_expection():
     context = SparkExpectationsContext(product_id="product1", spark=spark)
     context._table_name = ""

tests/notification/plugins/test_teams.py

@@ -0,0 +1,65 @@
+from unittest.mock import patch, Mock
+import pytest
+import requests
+from spark_expectations.core.exceptions import SparkExpectationsTeamsNotificationException
+from spark_expectations.notifications.plugins.teams import SparkExpectationsTeamsPluginImpl
+
+
+@patch('spark_expectations.notifications.plugins.teams.SparkExpectationsContext', autospec=True, spec_set=True)
+def test_send_notification_success(_mock_context):
+    # Arrange
+    teams_handler = SparkExpectationsTeamsPluginImpl()
+    _mock_context.get_enable_teams = True
+    _mock_context.get_teams_webhook_url = "http://test_webhook_url"
+
+    _config_args = {
+        "title": "SE Notification",
+        "themeColor": "008000", "message": "test message"}
+
+    # Mock requests.post to return a response with status code 200
+    with patch.object(requests, "post") as mock_post:
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_post.return_value = mock_response
+
+        # Act
+        teams_handler.send_notification(_context=_mock_context, _config_args=_config_args)
+
+        # Assert
+        mock_post.assert_called_once_with(_mock_context.get_teams_webhook_url, json={
+            "title": "SE Notification",
+            "themeColor": "008000", "text": "test message"}, timeout=10)
+
+
+@patch('spark_expectations.notifications.plugins.teams.SparkExpectationsContext', autospec=True, spec_set=True)
+def test_send_notification_exception(_mock_context):
+    # Arrange
+    teams_handler = SparkExpectationsTeamsPluginImpl()
+    _mock_context.get_enable_teams = True
+    _mock_context.get_teams_webhook_url = "http://test_webhook_url"
+    _config_args = {"message": "test message"}
+
+    # Mock requests.post to return a response with status code 404
+    with patch.object(requests, "post") as mock_post:
+        mock_response = Mock()
+        mock_response.status_code = 404
+        mock_post.return_value = mock_response
+
+        # Act and Assert
+        with pytest.raises(SparkExpectationsTeamsNotificationException):
+            teams_handler.send_notification(_context=_mock_context, _config_args=_config_args)
+
+
+@patch('spark_expectations.notifications.plugins.teams.SparkExpectationsContext', autospec=True, spec_set=True)
+def test_send_notification_teams_disabled(_mock_context):
+    # Arrange
+    teams_handler = SparkExpectationsTeamsPluginImpl()
+    _mock_context.get_enable_teams = False
+    _mock_context.get_teams_webhook_url = "http://test_webhook_url"
+    _config_args = {"message": "test message"}
+
+    with patch.object(requests, "post") as mock_post:
+        # Act
+        teams_handler.send_notification(_context=_mock_context, _config_args=_config_args)
+
+        mock_post.post.assert_not_called()

tests/notification/test__init__.py

@@ -6,6 +6,9 @@
 from spark_expectations.notifications.plugins.slack import (
     SparkExpectationsSlackPluginImpl,
 )
+from spark_expectations.notifications.plugins.teams import (
+    SparkExpectationsTeamsPluginImpl,
+)
 
 
 def test_notifications_hook():
@@ -15,11 +18,14 @@ def test_notifications_hook():
     # act
     email_plugin = pm.get_plugin("spark_expectations_email_notification")
     slack_plugin = pm.get_plugin("spark_expectations_slack_notification")
+    teams_plugin = pm.get_plugin("spark_expectations_teams_notification")
     # Check that the correct number of plugins have been registered
-    assert len(pm.list_name_plugin()) == 2
+    assert len(pm.list_name_plugin()) == 3
     # assert
     assert isinstance(pm, pluggy.PluginManager)
     assert email_plugin is not None
     assert slack_plugin is not None
+    assert teams_plugin is not None
     assert isinstance(email_plugin, SparkExpectationsEmailPluginImpl)
     assert isinstance(slack_plugin, SparkExpectationsSlackPluginImpl)
+    assert isinstance(teams_plugin, SparkExpectationsTeamsPluginImpl)

tests/utils/test_reader.py

@@ -47,6 +47,8 @@ def fixture_product_rules():
          "spark.expectations.notifications.email.subject": "Test email",
          "spark.expectations.notifications.slack.enabled": False,
          "spark.expectations.notifications.slack.webhook_url": "",
+         "spark.expectations.notifications.teams.enabled": False,
+         "spark.expectations.notifications.teams.webhook_url": "",
      }, None),
     ({
          "spark.expectations.notifications.email.enabled": True,
@@ -72,6 +74,13 @@ def fixture_product_rules():
          "spark.expectations.notifications.slack.enabled": True,
          "spark.expectations.notifications.slack.webhook_url": "",
      }, SparkExpectationsMiscException),
+    ({"spark.expectations.notifications.teams.enabled": True,
+      "spark.expectations.notifications.teams.webhook_url": "https://hooks.teams.com/services/..."},
+     None),
+    ({
+         "spark.expectations.notifications.teams.enabled": True,
+         "spark.expectations.notifications.teams.webhook_url": "",
+     }, SparkExpectationsMiscException),
 ])
 def test_set_notification_param(notification, expected_result):
     # This function helps/implements test cases for while setting notification
@@ -104,6 +113,11 @@ def test_set_notification_param(notification, expected_result):
                 notification.get("spark.expectations.notifications.slack.enabled"))
             mock_context.set_slack_webhook_url.assert_called_once_with(
                 notification.get("spark.expectations.notifications.slack.webhook_url"))
+        if notification.get("spark.expectations.notifications.teams.enabled"):
+            mock_context.set_enable_teams.assert_called_once_with(
+                notification.get("spark.expectations.notifications.teams.enabled"))
+            mock_context.set_teams_webhook_url.assert_called_once_with(
+                notification.get("spark.expectations.notifications.teams.webhook_url"))
     else:
         with pytest.raises(expected_result, match=r"All params/variables required for [a-z]+ notification "
                                                   "is not configured or supplied"):