Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions NOTIFICATION_CONFIG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:**
Expand Down Expand Up @@ -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"
Expand Down
1 change: 1 addition & 0 deletions config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ app:
pushover:
# user_key: <your Pushover user key>
# api_token: <your Pushover api token>
# priority: <your notification priority (optional)>
smtp:
# If you want to receive email notifications about expired/missing 2FA credentials then uncomment
# email: "sender@test.com"
Expand Down
11 changes: 11 additions & 0 deletions src/config_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 12 additions & 8 deletions src/notify.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,36 +195,40 @@ 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.

Args:
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:
True if message was sent successfully, False otherwise
"""
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
Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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:
Expand Down
13 changes: 13 additions & 0 deletions tests/test_config_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
32 changes: 30 additions & 2 deletions tests/test_notify.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand All @@ -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,
)

Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand All @@ -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
Expand Down
Loading