diff --git a/README.md b/README.md index ecad1727a..44f5f5cf4 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,9 @@ app: telegram: # bot_token: # chat_id: + pushover: + # user_key: + # api_token: smtp: ## If you want to receive email notifications about expired/missing 2FA credentials then uncomment # email: "user@test.com" diff --git a/config.yaml b/config.yaml index a502b9fc1..971538162 100644 --- a/config.yaml +++ b/config.yaml @@ -17,6 +17,9 @@ app: telegram: # bot_token: # chat_id: + pushover: + # user_key: + # api_token: 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 be680f7db..7e5eea047 100644 --- a/src/config_parser.py +++ b/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 diff --git a/src/notify.py b/src/notify.py index 11342181a..97a948862 100644 --- a/src/notify.py +++ b/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) diff --git a/tests/data/test_config.yaml b/tests/data/test_config.yaml index beaf98250..17bf1528c 100644 --- a/tests/data/test_config.yaml +++ b/tests/data/test_config.yaml @@ -14,6 +14,9 @@ app: discord: # webhook_url: # username: icloud-docker + pushover: + # user_key: + # api_token: smtp: # If you want to recieve email notifications about expired/missing 2FA credentials then uncomment # email: sender@test.com diff --git a/tests/test_config_parser.py b/tests/test_config_parser.py index b166e06aa..12c2cc88e 100644 --- a/tests/test_config_parser.py +++ b/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)) diff --git a/tests/test_notify.py b/tests/test_notify.py index c73469c6a..8f6f9dad4 100644 --- a/tests/test_notify.py +++ b/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, + )