From c0095cd1101aec1717d755887b00887f8aa3dd4e Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 6 Dec 2025 15:56:54 -0500 Subject: [PATCH] Added Brevo Support (#1455) --- README.md | 1 + apprise/plugins/brevo.py | 562 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 2 +- pyproject.toml | 1 + tests/test_plugin_brevo.py | 264 +++++++++++++ 5 files changed, 829 insertions(+), 1 deletion(-) create mode 100644 apprise/plugins/brevo.py create mode 100644 tests/test_plugin_brevo.py diff --git a/README.md b/README.md index d32ab0cf..9feaac50 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,7 @@ The table below identifies the services this tool supports and some example serv | [AWS SES](https://github.com/caronc/apprise/wiki/Notify_ses) | ses:// | (TCP) 443 | ses://user@domain/AccessKeyID/AccessSecretKey/RegionName
ses://user@domain/AccessKeyID/AccessSecretKey/RegionName/email1/email2/emailN | [Bark](https://github.com/caronc/apprise/wiki/Notify_bark) | bark:// | (TCP) 80 or 443 | bark://hostname
bark://hostname/device_key
bark://hostname/device_key1/device_key2/device_keyN
barks://hostname
barks://hostname/device_key
barks://hostname/device_key1/device_key2/device_keyN | [BlueSky](https://github.com/caronc/apprise/wiki/Notify_bluesky) | bluesky:// | (TCP) 443 | bluesky://Handle:AppPw
bluesky://Handle:AppPw/TargetHandle
bluesky://Handle:AppPw/TargetHandle1/TargetHandle2/TargetHandleN +| [Brevo](https://github.com/caronc/apprise/wiki/Notify_brevo) | brevo:// | (TCP) 443 | brevo://APIToken:FromEmail/
brevo://APIToken:FromEmail/ToEmail
brevo://APIToken:FromEmail/ToEmail1/ToEmail2/ToEmailN/ | [Chanify](https://github.com/caronc/apprise/wiki/Notify_chanify) | chantify:// | (TCP) 443 | chantify://token | [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token
discord://avatar@webhook_id/webhook_token | [Dot.](https://github.com/caronc/apprise/wiki/Notify_dot) | dot:// | (TCP) 443 | dot://apikey@device_id/text/
dot://apikey@device_id/image/
**Note**: `device_id` is the Quote/0 hardware serial diff --git a/apprise/plugins/brevo.py b/apprise/plugins/brevo.py new file mode 100644 index 00000000..f17b7f92 --- /dev/null +++ b/apprise/plugins/brevo.py @@ -0,0 +1,562 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +# API Reference: https://developers.brevo.com/reference/getting-started-1 + +from json import dumps +from os.path import splitext + +import requests + +from .. import exception +from ..common import NotifyFormat, NotifyType +from ..conversion import convert_between +from ..locale import gettext_lazy as _ +from ..utils.parse import is_email, parse_list, validate_regex +from .base import NotifyBase + +# Extend HTTP Error Messages (most common Brevo SMTP errors) +BREVO_HTTP_ERROR_MAP = { + 400: "Bad Request - Invalid payload or missing parameters.", + 401: "Unauthorized - Invalid Brevo API key.", + 402: "Payment Required - Plan limitation or credit issue.", + 429: "Too Many Requests - Rate limit exceeded.", +} + +# Comprehensive list of Brevo-supported extensions for Transactional Emails +# Source: Brevo API Documentation & Transactional Attachment Guidelines +BREVO_VALID_EXTENSIONS = ( + # Documents & Text + "xlsx", "xls", "ods", "docx", "docm", "doc", "csv", "pdf", "txt", + "rtf", "msg", "pub", "mobi", "ppt", "pptx", "eps", "odt", "ics", + "xml", "css", "html", "htm", "shtml", + # Images + "gif", "jpg", "jpeg", "png", "tif", "tiff", "bmp", "cgm", + # Archives + "zip", "tar", "ez", "pkpass", + # Audio + "mp3", "m4a", "m4v", "wma", "ogg", "flac", "wav", "aif", "aifc", "aiff", + # Video + "mp4", "mov", "avi", "mkv", "mpeg", "mpg", "wmv" +) + + +class NotifyBrevo(NotifyBase): + """A wrapper for Notify Brevo Notifications.""" + + # The default descriptive name associated with the Notification + service_name = "Brevo" + + # The services URL + service_url = "https://www.brevo.com/" + + # The default secure protocol + secure_protocol = "brevo" + + # A URL that takes you to the setup/help of the specific protocol + setup_url = "https://github.com/caronc/apprise/wiki/Notify_brevo" + + # Default to markdown + notify_format = NotifyFormat.HTML + + # The default Email API URL to use + notify_url = "https://api.brevo.com/v3/smtp/email" + + # Support attachments + attachment_support = True + + # Allow 300 requests per minute. + # 60/300 = 0.2 + request_rate_per_sec = 0.2 + + # The default subject to use if one isn't specified. + default_empty_subject = "" + + # Define object templates + templates = ( + "{schema}://{apikey}:{from_email}", + "{schema}://{apikey}:{from_email}/{targets}", + ) + + # Define our template arguments + template_tokens = dict( + NotifyBase.template_tokens, + **{ + "apikey": { + "name": _("API Key"), + "type": "string", + "private": True, + "required": True, + "regex": (r"^[a-zA-Z0-9._-]+$", "i"), + }, + "from_email": { + "name": _("Source Email"), + "type": "string", + "required": True, + }, + "target_email": { + "name": _("Target Email"), + "type": "string", + "map_to": "targets", + }, + "targets": { + "name": _("Targets"), + "type": "list:string", + }, + }, + ) + + # Define our template arguments + template_args = dict( + NotifyBase.template_args, + **{ + "to": { + "alias_of": "targets", + }, + "cc": { + "name": _("Carbon Copy"), + "type": "list:string", + }, + "bcc": { + "name": _("Blind Carbon Copy"), + "type": "list:string", + }, + "reply": { + "name": _("Reply To Email"), + "type": "string", + "map_to": "reply_to", + }, + }, + ) + + def __init__( + self, + apikey, + from_email, + targets=None, + reply_to=None, + cc=None, + bcc=None, + **kwargs, + ): + """Initialize Notify Brevo Object.""" + super().__init__(**kwargs) + + # API Key (associated with project) + self.apikey = validate_regex( + apikey, *self.template_tokens["apikey"]["regex"] + ) + if not self.apikey: + msg = f"An invalid Brevo API Key ({apikey}) was specified." + self.logger.warning(msg) + raise TypeError(msg) + + result = is_email(from_email) + if not result: + msg = f"Invalid ~From~ email specified: {from_email}" + self.logger.warning(msg) + raise TypeError(msg) + + # Store email address + self.from_email = result["full_email"] + + # Reply-to + self.reply_to = None + if reply_to: + result = is_email(reply_to) + if not result: + msg = "An invalid Brevo Reply To ({}) was specified.".format( + f"{reply_to}") + self.logger.warning(msg) + raise TypeError(msg) + + self.reply_to = ( + result["name"] if result["name"] else False, + result["full_email"], + ) + + # Acquire Targets (To Emails) + self.targets = [] + + # Acquire Carbon Copies + self.cc = set() + + # Acquire Blind Carbon Copies + self.bcc = set() + + # Validate recipients (to:) and drop bad ones: + if targets: + for recipient in parse_list(targets): + + result = is_email(recipient) + if result: + self.targets.append(result["full_email"]) + continue + + self.logger.warning( + f"Dropped invalid email ({recipient}) specified.", + ) + else: + # add ourselves + self.targets.append(self.from_email) + + # Validate recipients (cc:) and drop bad ones: + for recipient in parse_list(cc): + + result = is_email(recipient) + if result: + self.cc.add(result["full_email"]) + continue + + self.logger.warning( + f"Dropped invalid Carbon Copy email ({recipient}) specified.", + ) + + # Validate recipients (bcc:) and drop bad ones: + for recipient in parse_list(bcc): + + result = is_email(recipient) + if result: + self.bcc.add(result["full_email"]) + continue + + self.logger.warning( + "Dropped invalid Blind Carbon Copy email " + f"({recipient}) specified.", + ) + + return + + @property + def url_identifier(self): + """Returns all of the identifiers that make this URL unique from + another simliar one. + + Targets or end points should never be identified here. + """ + return (self.secure_protocol, self.apikey, self.from_email) + + def url(self, privacy=False, *args, **kwargs): + """Returns the URL built dynamically based on specified arguments.""" + + # Our URL parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + if len(self.cc) > 0: + # Handle our Carbon Copy Addresses + params["cc"] = ",".join(self.cc) + + if len(self.bcc) > 0: + # Handle our Blind Carbon Copy Addresses + params["bcc"] = ",".join(self.bcc) + + if self.reply_to: + # Handle our reply to address + params["reply"] = ( + "{} <{}>".format(*self.reply_to) + if self.reply_to[0] + else self.reply_to[1] + ) + + # a simple boolean check as to whether we display our target emails + # or not + has_targets = not ( + len(self.targets) == 1 and self.targets[0] == self.from_email + ) + + return "{schema}://{apikey}:{from_email}/{targets}?{params}".format( + schema=self.secure_protocol, + apikey=self.pprint(self.apikey, privacy, safe=""), + # never encode email since it plays a huge role in our hostname + from_email=self.from_email, + targets=( + "" + if not has_targets + else "/".join( + [NotifyBrevo.quote(x, safe="") for x in self.targets] + ) + ), + params=NotifyBrevo.urlencode(params), + ) + + def __len__(self): + """Returns the number of targets associated with this notification.""" + return max(len(self.targets), 1) + + def send( + self, + body, + title="", + notify_type=NotifyType.INFO, + attach=None, + **kwargs, + ): + """Perform Brevo Notification.""" + + if not self.targets: + # There is no one to email; we're done + self.logger.warning( + "There are no Brevo email recipients to notify") + return False + + headers = { + "User-Agent": self.app_id, + "Content-Type": "application/json", + "Accept": "application/json", + "api-key": self.apikey, + } + + # error tracking (used for function return) + has_error = False + + # A Simple Email Payload Template + _payload = { + "sender": { + "email": self.from_email, + }, + # Placeholder, filled per target + "to": [{"email": None}], + "subject": title if title else self.default_empty_subject, + } + # Body selection + use_html = self.notify_format == NotifyFormat.HTML + + if use_html: + # body already normalised; keep your existing logic + _payload["htmlContent"] = body + _payload["textContent"] = convert_between( + NotifyFormat.HTML, NotifyFormat.TEXT, body + ) + else: + # Plain text requested, but Brevo still wants HTML + _payload["textContent"] = body + _payload["htmlContent"] = convert_between( + NotifyFormat.TEXT, NotifyFormat.HTML, body + ) + + if attach and self.attachment_support: + attachments = [] + + # Send our attachments + for no, attachment in enumerate(attach, start=1): + # Perform some simple error checking + if not attachment: + # We could not access the attachment + self.logger.error( + "Could not access Brevo attachment" + f" {attachment.url(privacy=True)}." + ) + return False + + # Brevo does not track content/mime type and relies 100% + # entirely on the filename extension as to whether or not it + # will accept it or not. + # + # The below prepares a safe_name (which can't be .dat like + # other plugins since Brevo rejects that type). For this + # reason .txt is chosen intentionally for this circumstance. + + # Use the attachment name if available, otherwise default to a + # generic name + raw_name = attachment.name \ + if attachment.name else f"file{no:03}.txt" + + # If the filename does NOT match a supported extension, append + # .txt + _, ext = splitext(raw_name) + safe_name = f"{raw_name}.txt" if ( + not ext or ext[1:].lower() + not in BREVO_VALID_EXTENSIONS) else raw_name + + try: + attachments.append({ + "content": attachment.base64(), + "name": safe_name, + }) + + except exception.AppriseException: + # We could not access the attachment + self.logger.error( + "Could not access Brevo attachment" + f" {attachment.url(privacy=True)}." + ) + return False + + self.logger.debug( + "Appending Brevo attachment" + f" {attachment.url(privacy=True)}" + ) + + # Append our attachments to the payload + _payload.update({ + "attachment": attachments, + }) + + if self.reply_to: + _payload["replyTo"] = {"email": self.reply_to[1]} + + targets = list(self.targets) + while len(targets) > 0: + target = targets.pop(0) + + # Create a copy of our template + payload = _payload.copy() + + # the cc, bcc, to field must be unique or SendMail will fail, the + # below code prepares this by ensuring the target isn't in the cc + # list or bcc list. It also makes sure the cc list does not contain + # any of the bcc entries + cc = self.cc - self.bcc - {target} + bcc = self.bcc - {target} + + # Set our main recipient + payload["to"] = [{"email": target}] + + if len(cc): + payload["cc"] = [{"email": email} for email in cc] + + if len(bcc): + payload["bcc"] = [{"email": email} for email in bcc] + + self.logger.debug( + "Brevo POST URL:" + f" {self.notify_url} (cert_verify={self.verify_certificate!r})" + ) + self.logger.debug(f"Brevo Payload: {payload!s}") + + # Always call throttle before any remote server i/o is made + self.throttle() + try: + r = requests.post( + self.notify_url, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + if r.status_code not in ( + requests.codes.ok, + requests.codes.accepted, + requests.codes.created, + ): + # We had a problem + status_str = NotifyBrevo.http_response_code_lookup( + r.status_code, BREVO_HTTP_ERROR_MAP + ) + + self.logger.warning( + "Failed to send Brevo notification to {}: " + "{}{}error={}.".format( + target, + status_str, + ", " if status_str else "", + r.status_code, + ) + ) + + self.logger.debug(f"Response Details:\r\n{r.content}") + + # Mark our failure + has_error = True + continue + + else: + self.logger.info( + f"Sent Brevo notification to {target}." + ) + + except requests.RequestException as e: + self.logger.warning( + "A Connection error occurred sending Brevo " + f"notification to {target}." + ) + self.logger.debug(f"Socket Exception: {e!s}") + + # Mark our failure + has_error = True + continue + + return not has_error + + @staticmethod + def parse_url(url): + """Parses the URL and returns enough arguments that can allow us to re- + instantiate this object.""" + + results = NotifyBase.parse_url(url) + if not results: + # We're done early as we couldn't load the results + return results + + # Our URL looks like this: + # {schema}://{apikey}:{from_email}/{targets} + # + # which actually equates to: + # {schema}://{user}:{password}@{host}/{email1}/{email2}/etc.. + # ^ ^ ^ + # | | | + # apikey -from addr- + + if not results.get("user"): + # An API Key as not properly specified + return None + + if not results.get("password"): + # A From Email was not correctly specified + return None + + # Prepare our API Key + results["apikey"] = NotifyBrevo.unquote(results["user"]) + + # Prepare our From Email Address + results["from_email"] = "{}@{}".format( + NotifyBrevo.unquote(results["password"]), + NotifyBrevo.unquote(results["host"]), + ) + + # Acquire our targets + results["targets"] = NotifyBrevo.split_path(results["fullpath"]) + + # The 'to' makes it easier to use yaml configuration + if "to" in results["qsd"] and len(results["qsd"]["to"]): + results["targets"] += NotifyBrevo.parse_list( + results["qsd"]["to"] + ) + + # Handle Carbon Copy Addresses + if "cc" in results["qsd"] and len(results["qsd"]["cc"]): + results["cc"] = NotifyBrevo.parse_list(results["qsd"]["cc"]) + + # Handle Blind Carbon Copy Addresses + if "bcc" in results["qsd"] and len(results["qsd"]["bcc"]): + results["bcc"] = NotifyBrevo.parse_list(results["qsd"]["bcc"]) + + # Handle Reply To Address + if "reply" in results["qsd"] and len(results["qsd"]["reply"]): + results["reply_to"] = NotifyBrevo.unquote(results["qsd"]["reply"]) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index fd0e666f..4d3e468a 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -59,7 +59,7 @@ Apprise is a Python package that simplifies access to many popular \ notification services. It supports sending alerts to platforms such as: \ \ `46elks`, `AfricasTalking`, `Apprise API`, `APRS`, `AWS SES`, `AWS SNS`, \ -`Bark`, `BlueSky`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, \ +`Bark`, `BlueSky`, `Brevo`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, \ `Clickatell`, `ClickSend`, `DAPNET`, `DingTalk`, `Discord`, \ `Dot. (Quote/0)`, `E-Mail`, `Emby`, `FCM`, `Feishu`, `Flock`, \ `Free Mobile`, `Google Chat`, `Gotify`, `Growl`, `Guilded`, \ diff --git a/pyproject.toml b/pyproject.toml index b40219b9..5ef48395 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,7 @@ keywords = [ "AWS", "Bark", "BlueSky", + "Brevo", "BulkSMS", "BulkVS", "Burst SMS", diff --git a/tests/test_plugin_brevo.py b/tests/test_plugin_brevo.py new file mode 100644 index 00000000..0c4993f0 --- /dev/null +++ b/tests/test_plugin_brevo.py @@ -0,0 +1,264 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +# Disable logging for a cleaner testing output +import logging +import os +from unittest import mock + +from helpers import AppriseURLTester +import pytest +import requests + +from apprise import Apprise, AppriseAttachment, NotifyType +from apprise.plugins.brevo import NotifyBrevo + +logging.disable(logging.CRITICAL) + +# Attachment Directory +TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), "var") + +# a test UUID we can use +UUID4 = "8b799edf-6f98-4d3a-9be7-2862fb4e5752" + +# Our Testing URLs +apprise_url_tests = ( + ( + "brevo://", + { + "instance": None, + }, + ), + ( + "brevo://:@/", + { + "instance": None, + }, + ), + ( + "brevo://abcd", + { + # Just an broken email (no api key or email) + "instance": None, + }, + ), + ( + "brevo://abcd@host", + { + # Just an Email specified, no API Key + "instance": None, + }, + ), + ( + "brevo://invalid-api-key+*-d:user@example.com", + { + # An invalid API Key + "instance": TypeError, + }, + ), + ( + ("brevo://abcd:user@example.com/newuser@example.com" + "?reply=%20!"), + { + # An invalid Reply-To address + "instance": TypeError, + }, + ), + ( + "brevo://abcd:user@example.com?format=text", + { + # No To/Target Address(es) specified; so we sub in the same From + # address + "instance": NotifyBrevo, + }, + ), + ( + ("brevo://abcd:user@example.com/newuser@example.com" + "?reply=user@example.ca"), + { + # A good email + "instance": NotifyBrevo, + }, + ), + ( + "brevo://abcd:user@example.com/bademailaddress", + { + # won't be able to send email + "instance": NotifyBrevo, + "notify_response": False, + }, + ), + ( + ( + "brevo://abcd:user@example.com/newuser@example.com" + "?bcc=l2g@nuxref.com" + ), + { + # A good email with Blind Carbon Copy + "instance": NotifyBrevo, + }, + ), + ( + ( + "brevo://abcd:user@example.com/newuser@example.com" + "?cc=l2g@nuxref.com" + ), + { + # A good email with Carbon Copy + "instance": NotifyBrevo, + }, + ), + ( + ( + "brevo://abcd:user@example.com/newuser@example.com" + "?to=l2g@nuxref.com" + ), + { + # A good email with Carbon Copy + "instance": NotifyBrevo, + }, + ), + ( + "brevo://abcd:user@example.ca/newuser@example.ca", + { + "instance": NotifyBrevo, + # force a failure + "response": False, + "requests_response_code": requests.codes.internal_server_error, + }, + ), + ( + "brevo://abcd:user@example.uk/newuser@example.uk", + { + "instance": NotifyBrevo, + # throw a bizzare code forcing us to fail to look it up + "response": False, + "requests_response_code": 999, + }, + ), + ( + "brevo://abcd:user@example.au/newuser@example.au", + { + "instance": NotifyBrevo, + # Throws a series of i/o exceptions with this flag + # is set and tests that we gracfully handle them + "test_requests_exceptions": True, + }, + ), +) + + +def test_plugin_brevo_urls(): + """NotifyBrevo() Apprise URLs.""" + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch("requests.get") +@mock.patch("requests.post") +def test_plugin_brevo_edge_cases(mock_post, mock_get): + """NotifyBrevo() Edge Cases.""" + + # no apikey + with pytest.raises(TypeError): + NotifyBrevo(apikey=None, from_email="user@example.com") + + # invalid from email + with pytest.raises(TypeError): + NotifyBrevo(apikey="abcd", from_email="!invalid") + + # no email + with pytest.raises(TypeError): + NotifyBrevo(apikey="abcd", from_email=None) + + # Invalid To email address + NotifyBrevo( + apikey="abcd", from_email="user@example.com", targets="!invalid" + ) + + # Test invalid bcc/cc entries mixed with good ones + assert isinstance( + NotifyBrevo( + apikey="abcd", + from_email="l2g@example.com", + bcc=("abc@def.com", "!invalid"), + cc=("abc@test.org", "!invalid"), + ), + NotifyBrevo, + ) + + +@mock.patch("requests.post") +def test_plugin_brevo_attachments(mock_post): + """NotifyBrevo() Attachments.""" + + request = mock.Mock() + request.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = request + + path = os.path.join(TEST_VAR_DIR, "apprise-test.gif") + attach = AppriseAttachment(path) + obj = Apprise.instantiate("brevo://abcd:user@example.com") + assert isinstance(obj, NotifyBrevo) + assert ( + obj.notify( + body="body", + title="title", + notify_type=NotifyType.INFO, + attach=attach, + ) + is True + ) + + mock_post.reset_mock() + + # Try again in a use case where we can't access the file + with mock.patch("os.path.isfile", return_value=False): + assert ( + obj.notify( + body="body", + title="title", + notify_type=NotifyType.INFO, + attach=attach, + ) + is False + ) + + # Try again in a use case where we can't access the file + with mock.patch("builtins.open", side_effect=OSError): + assert ( + obj.notify( + body="body", + title="title", + notify_type=NotifyType.INFO, + attach=attach, + ) + is False + )