Add Pushover Notification Support and Unit Tests (#286)

* Update config_parser.py

* Update notify.py

* Update README.md

* Update config.yaml

* Update config_parser.py

* Update config_parser.py

* Update notify.py

* Update test_notify.py

* Update test_config_parser.py

* Update test_config.yaml

* Update test_config_parser.py

* Update test_notify.py

* Update test_notify.py

* Update test_notify.py

* Update notify.py

* Update notify.py

---------

Co-authored-by: Mandar Patil <mandarons@pm.me>

Author: john-pierre

README.md

@@ -83,6 +83,9 @@ app:
   telegram:
   # bot_token: <your Telegram bot token>
   # chat_id: <your Telegram user or chat ID>
+  pushover:
+  # user_key: <your Pushover user key>
+  # api_token: <your Pushover api token>
   smtp:
     ## If you want to receive email notifications about expired/missing 2FA credentials then uncomment
     # email: "user@test.com"

config.yaml

@@ -17,6 +17,9 @@ app:
   telegram:
   # bot_token: <your Telegram bot token>
   # chat_id: <your Telegram user or chat ID>
+  pushover:
+  # user_key: <your Pushover user key>
+  # api_token: <your Pushover api token>
   smtp:
   # If you want to receive email notifications about expired/missing 2FA credentials then uncomment
   # email: "sender@test.com"

src/config_parser.py

@@ -409,3 +409,25 @@ def get_discord_username(config):
     else:
         username = get_config_value(config=config, config_path=config_path)
     return username
+
+# Get pushover user key
+def get_pushover_user_key(config):
+    """Return Pushover user key from config."""
+    user_key = None
+    config_path = ["app", "pushover", "user_key"]
+    if not traverse_config_path(config=config, config_path=config_path):
+        LOGGER.warning(f"Warning: user_key is not found in {config_path_to_string(config_path)}.")
+    else:
+        user_key = get_config_value(config=config, config_path=config_path)
+    return user_key
+
+# Get pushover api token
+def get_pushover_api_token(config):
+    """Return Pushover API token from config."""
+    api_token = None
+    config_path = ["app", "pushover", "api_token"]
+    if not traverse_config_path(config=config, config_path=config_path):
+        LOGGER.warning(f"Warning: api_token is not found in {config_path_to_string(config_path)}.")
+    else:
+        api_token = get_config_value(config=config, config_path=config_path)
+    return api_token

src/notify.py

@@ -77,6 +77,33 @@ def notify_discord(config, message, last_send=None, dry_run=False):
         LOGGER.warning("Not sending 2FA notification because Discord is not configured.")
     return sent_on
 
+def notify_pushover(config, message, last_send=None, dry_run=False):
+    """Send Pushover notification."""
+    sent_on = None
+    user_key = config_parser.get_pushover_user_key(config=config)
+    api_token = config_parser.get_pushover_api_token(config=config)
+
+    if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24):
+        LOGGER.info("Throttling Pushover to once a day")
+        sent_on = last_send
+    elif user_key and api_token:
+        sent_on = datetime.datetime.now()
+        if not dry_run:
+            if not post_message_to_pushover(api_token, user_key, message):
+                sent_on = None
+    else:
+        LOGGER.warning("Not sending 2FA notification because Pushover is not configured.")
+    return sent_on
+
+def post_message_to_pushover(api_token, user_key, message):
+    """Post message to Pushover API."""
+    url = "https://api.pushover.net/1/messages.json"
+    data = {"token": api_token, "user": user_key, "message": message}
+    response = requests.post(url, data=data, timeout=10)
+    if response.status_code == 200:
+        return True
+    LOGGER.error(f"Failed to send Pushover notification. Response: {response.text}")
+    return False
 
 def send(config, username, last_send=None, dry_run=False, region="global"):
     """Send notifications."""
@@ -91,6 +118,7 @@ def send(config, username, last_send=None, dry_run=False, region="global"):
     subject = f"icloud-docker: Two step authentication is required for {username}"
     notify_telegram(config=config, message=message, last_send=last_send, dry_run=dry_run)
     notify_discord(config=config, message=message, last_send=last_send, dry_run=dry_run)
+    notify_pushover(config=config, message=message, last_send=last_send, dry_run=dry_run)
     email = config_parser.get_smtp_email(config=config)
     to_email = config_parser.get_smtp_to_email(config=config)
     host = config_parser.get_smtp_host(config=config)

tests/data/test_config.yaml

@@ -14,6 +14,9 @@ app:
   discord:
   # webhook_url: <server webhook>
   # username: icloud-docker
+  pushover:
+  # user_key: <your Pushover user key>
+  # api_token: <your Pushover api token>
   smtp:
   # If you want to recieve email notifications about expired/missing 2FA credentials then uncomment
   # email: sender@test.com

tests/test_config_parser.py

@@ -503,3 +503,29 @@ def test_get_discord_username(self):
     def test_get_discord_username_none_config(self):
         """None config."""
         self.assertIsNone(config_parser.get_discord_username(config=None))
+
+    def test_get_pushover_user_key(self):
+        """Test for getting Pushover user key."""
+        config = read_config(config_path=tests.CONFIG_PATH)
+        config["app"]["pushover"] = {"user_key": "pushover_user_key"}
+        self.assertEqual(
+            config["app"]["pushover"]["user_key"],
+            config_parser.get_pushover_user_key(config=config),
+        )
+
+    def test_get_pushover_user_key_none_config(self):
+        """None config for Pushover user key."""
+        self.assertIsNone(config_parser.get_pushover_user_key(config=None))
+
+    def test_get_pushover_api_token(self):
+        """Test for getting Pushover API token."""
+        config = read_config(config_path=tests.CONFIG_PATH)
+        config["app"]["pushover"] = {"api_token": "pushover_api_token"}
+        self.assertEqual(
+            config["app"]["pushover"]["api_token"],
+            config_parser.get_pushover_api_token(config=config),
+        )
+
+    def test_get_pushover_api_token_none_config(self):
+        """None config for Pushover API token."""
+        self.assertIsNone(config_parser.get_pushover_api_token(config=None))

tests/test_notify.py

@@ -8,8 +8,10 @@
 from src.email_message import EmailMessage as Message
 from src.notify import (
     notify_discord,
+    notify_pushover,
     notify_telegram,
     post_message_to_discord,
+    post_message_to_pushover,
     post_message_to_telegram,
 )
 
@@ -29,6 +31,7 @@ def setUp(self) -> None:
                     "password": "password",
                 },
                 "telegram": {"bot_token": "bot_token", "chat_id": "chat_id"},
+                "pushover": {"user_key": "pushover_user_key", "api_token": "pushover_api_token"},
             },
         }
         self.message_body = "message body"
@@ -332,3 +335,88 @@ def test_post_message_to_discord_fail(self):
                 data={"content": message, "username": "username"},
                 timeout=10,
             )
+
+    def test_notify_pushover_success(self):
+        """Test for successful Pushover notification."""
+        with patch("src.notify.post_message_to_pushover") as post_message_mock:
+            notify_pushover(self.config, self.message_body, None, False)
+
+            # Verify that post_message_to_pushover is called with the correct arguments
+            post_message_mock.assert_called_once_with(
+                self.config["app"]["pushover"]["api_token"],
+                self.config["app"]["pushover"]["user_key"],
+                self.message_body,
+            )
+
+    def test_notify_pushover_fail(self):
+        """Test for failed Pushover notification."""
+        with patch("src.notify.post_message_to_pushover") as post_message_mock:
+            post_message_mock.return_value = False
+            notify_pushover(self.config, self.message_body, None, False)
+
+            # Verify that post_message_to_pushover is called with the correct arguments
+            post_message_mock.assert_called_once_with(
+                self.config["app"]["pushover"]["api_token"],
+                self.config["app"]["pushover"]["user_key"],
+                self.message_body,
+            )
+
+    def test_notify_pushover_throttling(self):
+        """Test for throttled Pushover notification."""
+        last_send = datetime.datetime.now() - datetime.timedelta(hours=2)
+        dry_run = False
+
+        with patch("src.notify.post_message_to_pushover") as post_message_mock:
+            notify_pushover(self.config, self.message_body, last_send, dry_run)
+
+            # Verify that post_message_to_pushover is not called when throttled
+            post_message_mock.assert_not_called()
+
+    def test_notify_pushover_dry_run(self):
+        """Test for dry run mode in Pushover notification."""
+        last_send = datetime.datetime.now()
+        dry_run = True
+
+        with patch("src.notify.post_message_to_pushover") as post_message_mock:
+            notify_pushover(self.config, self.message_body, last_send, dry_run)
+
+            # Verify that post_message_to_pushover is not called in dry run mode
+            post_message_mock.assert_not_called()
+
+    def test_notify_pushover_no_config(self):
+        """Test for missing Pushover configuration."""
+        config = {}
+        last_send = None
+        dry_run = False
+
+        with patch("src.notify.post_message_to_pushover") as post_message_mock:
+            notify_pushover(config, self.message_body, last_send, dry_run)
+
+            # Verify that post_message_to_pushover is not called when Pushover configuration is missing
+            post_message_mock.assert_not_called()
+
+    def test_post_message_to_pushover(self):
+        """Test for successful post to Pushover."""
+        with patch("requests.post") as post_mock:
+            post_mock.return_value.status_code = 200
+            post_message_to_pushover("pushover_api_token", "pushover_user_key", "message")
+
+            # Verify that post is called with the correct arguments
+            post_mock.assert_called_once_with(
+                "https://api.pushover.net/1/messages.json",
+                data={"token": "pushover_api_token", "user": "pushover_user_key", "message": "message"},
+                timeout=10,
+            )
+
+    def test_post_message_to_pushover_fail(self):
+        """Test for failed post to Pushover."""
+        with patch("requests.post") as post_mock:
+            post_mock.return_value.status_code = 400
+            post_message_to_pushover("pushover_api_token", "pushover_user_key", "message")
+
+            # Verify that post is called with the correct arguments
+            post_mock.assert_called_once_with(
+                "https://api.pushover.net/1/messages.json",
+                data={"token": "pushover_api_token", "user": "pushover_user_key", "message": "message"},
+                timeout=10,
+            )