added automatic formatting of subject and ficex teams try catch and a…

…dded teams tests

notif/notificator.py

@@ -58,6 +58,19 @@ def send_notification_error(self, error: Exception) -> None:
         notification_error_message = self._parse_error(error)
         self.send_notification(message=notification_error_message)
 
+    @abstractmethod
+    def _format_subject(self, subject_message: str) -> str:
+        """
+        Abstract class to format a subject to create a 'title'.
+
+        Args:
+
+            subject_message (str): The message to format.
+
+        Return:
+            A formatted subject message.
+        """
+
     @staticmethod
     def _parse_error(error: Exception) -> str:
         """
@@ -67,7 +80,7 @@ def _parse_error(error: Exception) -> str:
 
             error (Exception): The exception raised during the script execution.
 
-        Returns:
+        Return:
             A formatted string base on the error message and error type.
         """
         error_type = type(error)
@@ -106,7 +119,14 @@ def __init__(self, webhook_url: str, on_error_sleep_time: int = 120):
         self.webhook_url = webhook_url
         self.headers = {'content-type': 'application/json'}
 
-        self.default_subject_message = "*Python script Slack notification*\n"
+        self.default_subject_message = "Python script Slack notification"
+
+    def _format_subject(self, subject_message: str) -> str:
+        """
+        We use Markdown formatting as specified in Slack
+        `documentation <https://api.slack.com/reference/surfaces/formatting>`_.
+        """
+        return f"*{subject_message}*\n"
 
     def send_notification(self, message: str, subject: Union[str, None] = None) -> None:
         """
@@ -117,13 +137,13 @@ def send_notification(self, message: str, subject: Union[str, None] = None) -> N
             message (str): The message to send as a notification message to the webhook url.
 
             subject (str): The subject of the notification. If None, the default message
-                '*Python script Slack notification*' is used. We use '*' to make bold the subject.
-                See `the documentation <https://api.slack.com/reference/surfaces/formatting>`_ to learn
-                how to format text using Markdown. By default None.
+                'Python script Slack notification' is used. Note that subject are formatted, text is bolded and
+                a new line is appended after the subject to create a 'title' effect. Default is None.
 
 
         """
         subject = subject if subject is not None else self.default_subject_message
+        subject = self._format_subject(subject)
         message = subject + message
 
         payload_message = {"text": message}
@@ -198,6 +218,12 @@ def __init__(self,
 
         self.default_subject_message = "Python script notification email"
 
+    def _format_subject(self, subject_message: str) -> str:
+        """
+        None since subject is the subject of the email.
+        """
+        pass
+
     def send_notification(self, message: str, subject: Union[str, None] = None) -> None:
         """
         Send a notification message to the destination email.
@@ -206,7 +232,7 @@ def send_notification(self, message: str, subject: Union[str, None] = None) -> N
 
             message (str): The message of the email.
             subject (str): The subject of the email. If None, the default message 'Python script notification email'
-                is used. By default None.
+                is used. Default is None.
 
         """
         subject = subject if subject is not None else self.default_subject_message
@@ -254,7 +280,13 @@ def __init__(self, channel_url: str, on_error_sleep_time: int = 120) -> None:
             raise ImportError("package notify_run need to be installed to use this class.")
         self.notifier = ChannelNotify(endpoint=channel_url)
 
-        self.default_subject_message = "**Python script notification**\n"
+        self.default_subject_message = "Python script notification"
+
+    def _format_subject(self, subject_message: str) -> str:
+        """
+        We use a similar logic as Markdown formatting as specified to highlight the subject.
+        """
+        return f"**{subject_message}**\n"
 
     def send_notification(self, message: str, subject: Union[str, None] = None) -> None:
         """
@@ -264,11 +296,12 @@ def send_notification(self, message: str, subject: Union[str, None] = None) -> N
 
             message (str): The message to send as a notification message to the channel.
             subject (str): The subject of the notification. If None, the default message
-                '**Python script notification**' is used. We use '**' create a highlight pattern and make a sort
-                of title. By default None.
+                'Python script notification' is used. Note that subject are formatted, text are surrounded with '*' and
+                a new line is appended after the subject to create a 'title' effect. Default is None.
 
         """
         subject = subject if subject is not None else self.default_subject_message
+        subject = self._format_subject(subject)
         message = subject + message
         try:
             self.notifier.send(message)
@@ -310,7 +343,14 @@ def __init__(self, webhook_url: str, on_error_sleep_time: int = 120):
             raise ImportError("package pymsteams need to be installed to use this class.")
         self.teams_hook = pymsteams.connectorcard(webhook_url)
 
-        self.default_subject_message = "**Python script Teams notification**\n"
+        self.default_subject_message = "Python script Teams notification"
+
+    def _format_subject(self, subject_message: str) -> str:
+        """
+        We use a similar logic as Markdown formatting as specified in Microsoft Teams
+        `documentation <https://docs.microsoft.com/en-us/microsoftteams/platform/task-modules-and-cards/cards/cards-format?tabs=adaptive-md%2Cconnector-html>`_.
+        """
+        return f"**{subject_message}**\n"
 
     def send_notification(self, message: str, subject: Union[str, None] = None) -> None:
         # pylint: disable=line-too-long
@@ -321,12 +361,12 @@ def send_notification(self, message: str, subject: Union[str, None] = None) -> N
 
             message (str): The message to send as a notification message to the webhook url.
             subject (str): The subject of the notification. If None, the default message
-                '**Python script Teams notification**' is used. We use '**' to bold the subject. See
-                `this documentation <https://docs.microsoft.com/en-us/microsoftteams/platform/task-modules-and-cards/cards/cards-format?tabs=adaptive-md%2Cconnector-html>`_
-                to learn how to format text using Markdown. By default None.
+                'Python script Teams notification' is used. Note that subject are formatted, text is bolded and
+                a new line is appended after the subject to create a 'title' effect. Default is None.
 
         """
         subject = subject if subject is not None else self.default_subject_message
+        subject = self._format_subject(subject)
         message = subject + message
 
         self.teams_hook.text(message)
@@ -338,6 +378,6 @@ def send_notification(self, message: str, subject: Union[str, None] = None) -> N
                 Warning)
             sleep(self.on_error_sleep_time)
             try:
-                self.teams_hook.text(message)
+                self.teams_hook.send()
             except pymsteams.TeamsWebhookException:
                 warnings.warn("Second error when trying to send notification, will abort sending message.", Warning)

notif/version.py

@@ -1 +1 @@
-__version__ = '0.5.3'
+__version__ = '0.5.4'

tests/test_notificator.py

@@ -3,9 +3,10 @@
 from smtplib import SMTPRecipientsRefused
 from unittest.mock import patch, call, MagicMock
 
+import pymsteams
 import requests
 
-from notif.notificator import SlackNotificator, EmailNotificator, ChannelNotificator
+from notif.notificator import SlackNotificator, EmailNotificator, ChannelNotificator, TeamsNotificator
 
 
 class SlackNotificatorTest(unittest.TestCase):
@@ -30,10 +31,10 @@ def test_givenASlackNotificator_whenSendNotification_thenSendMessageDefaultSubje
 
     @patch("notif.notificator.requests")
     def test_givenASlackNotificator_whenSendNotificationWithSubject_thenSendMessageWithSubject(self, requests_mock):
-        a_user_formatted_subject = "**Here a subject**"
+        a_user_formatted_subject = "Here a subject"
         self.slack_notificator.send_notification(self.a_notification, subject=a_user_formatted_subject)
 
-        expected_payload_message = {"text": a_user_formatted_subject + self.a_notification}
+        expected_payload_message = {"text": f"*{a_user_formatted_subject}*\n" + self.a_notification}
         post_call = [call.post(self.a_fake_web_hook, data=json.dumps(expected_payload_message), headers=self.headers)]
 
         requests_mock.assert_has_calls(post_call)
@@ -88,7 +89,7 @@ def test_givenAEmailNotificator_whenSendNotificationWithSubject_thenSendMessageW
                                              a_mock_smtp_server,
                                              on_error_sleep_time=1)  # 1 second since normal is 120
 
-        a_user_formatted_subject = "**Here a subject**"
+        a_user_formatted_subject = "Here a subject"
         email_notificator.send_notification(self.a_notification, subject=a_user_formatted_subject)
 
         expected_content = 'Subject: %s\n\n%s' % (a_user_formatted_subject, self.a_notification)
@@ -146,14 +147,14 @@ def test_givenAChannelNotificator_whenSendNotification_thenSendMessageDefaultSub
     @patch("notif.notificator.ChannelNotify")
     def test_givenAChannelNotificator_whenSendNotificationWithSubject_thenSendMessageWithSubject(
             self, channel_notify_mock):
-        a_user_formatted_subject = "**Here a subject**"
+        a_user_formatted_subject = "Here a subject"
 
         channel_notificator = ChannelNotificator(self.a_fake_channel_url,
                                                  on_error_sleep_time=1)  # 1 second since normal is 120
 
         channel_notificator.send_notification(self.a_notification, subject=a_user_formatted_subject)
 
-        expected_message = a_user_formatted_subject + self.a_notification
+        expected_message = f"**{a_user_formatted_subject}**\n" + self.a_notification
         post_call = [call(endpoint=self.a_fake_channel_url), call().send(expected_message)]
 
         channel_notify_mock.assert_has_calls(post_call)
@@ -178,5 +179,53 @@ def test_givenAChannelNotificator_whenSendNotificationDoesNotWork_thenWaitTimer(
             channel_notify_mock.assert_has_calls(post_call)
 
 
+class TeamsNotificatorTest(unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.a_fake_web_hook = "a_web_hook"
+        self.a_notification = "A normal text."
+        self.default_subject_message = "**Python script Teams notification**\n"
+        self.headers = {'content-type': 'application/json'}
+
+    @patch("notif.pymsteams.connectorcard")
+    def test_givenATeamsNotificator_whenSendNotification_thenSendMessageDefaultSubject(self, pymsteams_mock):
+        self.teams_notificator = TeamsNotificator(self.a_fake_web_hook,
+                                                  on_error_sleep_time=1)  # 1 second since normal is 120
+        self.teams_notificator.send_notification(self.a_notification)
+
+        expected_message = self.default_subject_message + self.a_notification
+        post_call = [call(self.a_fake_web_hook), call().text(expected_message), call().send()]
+
+        pymsteams_mock.assert_has_calls(post_call)
+
+    @patch("notif.pymsteams.connectorcard")
+    def test_givenATeamsNotificator_whenSendNotificationWithSubject_thenSendMessageWithSubject(self, pymsteams_mock):
+        self.teams_notificator = TeamsNotificator(self.a_fake_web_hook,
+                                                  on_error_sleep_time=1)  # 1 second since normal is 120
+
+        a_user_formatted_subject = "Here a subject"
+        self.teams_notificator.send_notification(self.a_notification, subject=a_user_formatted_subject)
+
+        expected_message = f"**{a_user_formatted_subject}**\n" + self.a_notification
+        post_call = [call(self.a_fake_web_hook), call().text(expected_message), call().send()]
+
+        pymsteams_mock.assert_has_calls(post_call)
+
+    @patch("notif.pymsteams.connectorcard")
+    def test_givenATeamsNotificator_whenSendNotificationDoesNotWork_thenWaitTimer(self, pymsteams_mock):
+        pymsteams_mock().send.side_effect = pymsteams.TeamsWebhookException
+
+        self.teams_notificator = TeamsNotificator(self.a_fake_web_hook,
+                                                  on_error_sleep_time=1)  # 1 second since normal is 120
+
+        with self.assertWarns(Warning):
+            self.teams_notificator.send_notification(self.a_notification)
+
+            expected_message = self.default_subject_message + self.a_notification
+            post_call = [call(self.a_fake_web_hook), call().text(expected_message), call().send(), call().send()]
+
+            pymsteams_mock.assert_has_calls(post_call)
+
+
 if __name__ == "__main__":
     unittest.main()