diff --git a/NOTIFICATION_CONFIG.md b/NOTIFICATION_CONFIG.md index 4c3222712..198786478 100644 --- a/NOTIFICATION_CONFIG.md +++ b/NOTIFICATION_CONFIG.md @@ -58,6 +58,7 @@ app: pushover: user_key: "your-30-char-user-key" api_token: "your-30-char-app-token" + priority: "your-notification-priority" ``` **Setup Steps:** @@ -185,6 +186,7 @@ app: pushover: user_key: "user-key" api_token: "app-token" + priority: 1 smtp: email: "icloud@domain.com" to: "admin@domain.com" diff --git a/config.yaml b/config.yaml index 3ffe66490..3fadddb8f 100644 --- a/config.yaml +++ b/config.yaml @@ -37,6 +37,7 @@ app: pushover: # user_key: # api_token: + # priority: smtp: # If you want to receive email notifications about expired/missing 2FA credentials then uncomment # email: "sender@test.com" diff --git a/src/config_parser.py b/src/config_parser.py index 76eea4fa9..88f71b279 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -907,6 +907,17 @@ def get_pushover_api_token(config: dict) -> str | None: """ return get_notification_config_value(config, "pushover", "api_token") +def get_pushover_notification_priority(config: dict) -> int | None: + """Return Pushover notification priority from config. + + Args: + config: Configuration dictionary + + Returns: + Pushover notification priority if configured, None otherwise + """ + return get_notification_config_value(config, "pushover", "priority") + # ============================================================================= # Sync Summary Notification Configuration Functions diff --git a/src/notify.py b/src/notify.py index 463b4673c..62cd27ef8 100644 --- a/src/notify.py +++ b/src/notify.py @@ -195,7 +195,7 @@ def notify_discord(config, message, last_send=None, dry_run=False): return None -def _get_pushover_config(config) -> tuple[str | None, str | None, bool]: +def _get_pushover_config(config) -> tuple[str | None, str | None, int | None, bool]: """ Extract Pushover configuration from config. @@ -203,21 +203,23 @@ def _get_pushover_config(config) -> tuple[str | None, str | None, bool]: config: The configuration dictionary Returns: - Tuple of (user_key, api_token, is_configured) + Tuple of (user_key, api_token, priority, is_configured) """ user_key = config_parser.get_pushover_user_key(config=config) api_token = config_parser.get_pushover_api_token(config=config) + priority = config_parser.get_pushover_notification_priority(config=config) is_configured = bool(user_key and api_token) - return user_key, api_token, is_configured + return user_key, api_token, priority, is_configured -def post_message_to_pushover(api_token: str, user_key: str, message: str) -> bool: +def post_message_to_pushover(api_token: str, user_key: str, priority: int | None, message: str) -> bool: """ Post message to Pushover API. Args: api_token: Pushover API token user_key: Pushover user key + priority: Pushover notification priority (-2 to 2, optional) message: Message to send Returns: @@ -225,6 +227,8 @@ def post_message_to_pushover(api_token: str, user_key: str, message: str) -> boo """ url = "https://api.pushover.net/1/messages.json" data = {"token": api_token, "user": user_key, "message": message} + if priority is not None: + data["priority"] = priority response = requests.post(url, data=data, timeout=10) if response.status_code == 200: return True @@ -249,7 +253,7 @@ def notify_pushover(config, message, last_send=None, dry_run=False): LOGGER.info("Throttling Pushover to once a day") return last_send - user_key, api_token, is_configured = _get_pushover_config(config) + user_key, api_token, priority, is_configured = _get_pushover_config(config) if not is_configured: LOGGER.warning("Not sending 2FA notification because Pushover is not configured.") return None @@ -259,7 +263,7 @@ def notify_pushover(config, message, last_send=None, dry_run=False): return sent_on # user_key and api_token are guaranteed to be non-None due to is_configured check - if post_message_to_pushover(api_token, user_key, message): # type: ignore[arg-type] + if post_message_to_pushover(api_token, user_key, priority, message): # type: ignore[arg-type] return sent_on return None @@ -663,14 +667,14 @@ def _send_pushover_no_throttle(config, message: str, dry_run: bool) -> bool: Returns: True if sent successfully, False otherwise """ - user_key, api_token, is_configured = _get_pushover_config(config) + user_key, api_token, priority, is_configured = _get_pushover_config(config) if not is_configured: return False if dry_run: return True - return post_message_to_pushover(api_token, user_key, message) # type: ignore[arg-type] + return post_message_to_pushover(api_token, user_key, priority, message) # type: ignore[arg-type] def _send_email_no_throttle(config, message: str, subject: str, dry_run: bool) -> bool: diff --git a/tests/test_config_parser.py b/tests/test_config_parser.py index 9b08059e3..2283f5177 100644 --- a/tests/test_config_parser.py +++ b/tests/test_config_parser.py @@ -581,6 +581,19 @@ 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)) + def test_get_pushover_notification_priority(self): + """Test for getting Pushover notification priority.""" + config = read_config(config_path=tests.CONFIG_PATH) + config["app"]["pushover"] = {"priority": 1} + self.assertEqual( + config["app"]["pushover"]["priority"], + config_parser.get_pushover_notification_priority(config=config), + ) + + def test_get_pushover_notification_priority_none_config(self): + """None config for Pushover notification priority.""" + self.assertIsNone(config_parser.get_pushover_notification_priority(config=None)) + def test_get_app_max_threads_default(self): """Test for getting app max threads with default value.""" config = read_config(config_path=tests.CONFIG_PATH) diff --git a/tests/test_notify.py b/tests/test_notify.py index e4900ee8a..00a8281d9 100644 --- a/tests/test_notify.py +++ b/tests/test_notify.py @@ -794,6 +794,7 @@ def test_notify_pushover_success(self): post_message_mock.assert_called_once_with( self.config["app"]["pushover"]["api_token"], self.config["app"]["pushover"]["user_key"], + None, self.message_body, ) @@ -807,6 +808,7 @@ def test_notify_pushover_fail(self): post_message_mock.assert_called_once_with( self.config["app"]["pushover"]["api_token"], self.config["app"]["pushover"]["user_key"], + None, self.message_body, ) @@ -848,7 +850,7 @@ 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") + post_message_to_pushover("pushover_api_token", "pushover_user_key", None, "message") # Verify that post is called with the correct arguments post_mock.assert_called_once_with( @@ -861,7 +863,7 @@ 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") + post_message_to_pushover("pushover_api_token", "pushover_user_key", None, "message") # Verify that post is called with the correct arguments post_mock.assert_called_once_with( @@ -870,6 +872,32 @@ def test_post_message_to_pushover_fail(self): timeout=10, ) + def test_post_message_to_pushover_with_priority(self): + """Test for post to Pushover with priority.""" + with patch("requests.post") as post_mock: + post_mock.return_value.status_code = 200 + post_message_to_pushover("pushover_api_token", "pushover_user_key", 1, "message") + + # Verify that post is called with the correct arguments including priority + post_mock.assert_called_once_with( + "https://api.pushover.net/1/messages.json", + data={"token": "pushover_api_token", "user": "pushover_user_key", "message": "message", "priority": 1}, + timeout=10, + ) + + def test_post_message_to_pushover_with_priority_zero(self): + """Test for post to Pushover with priority zero.""" + with patch("requests.post") as post_mock: + post_mock.return_value.status_code = 200 + post_message_to_pushover("pushover_api_token", "pushover_user_key", 0, "message") + + # Verify that post is called with priority=0 (should not be excluded due to truthiness check) + post_mock.assert_called_once_with( + "https://api.pushover.net/1/messages.json", + data={"token": "pushover_api_token", "user": "pushover_user_key", "message": "message", "priority": 0}, + timeout=10, + ) + def test_format_sync_summary_message_with_many_errors(self): """Test formatting message with more than 10 errors.""" from src.notify import _format_sync_summary_message