add multiples to
This commit is contained in:
@@ -1,29 +1,41 @@
|
|||||||
from email.mime.application import MIMEApplication
|
from email.mime.application import MIMEApplication
|
||||||
from email.mime.multipart import MIMEMultipart
|
from email.mime.multipart import MIMEMultipart
|
||||||
from email.mime.text import MIMEText
|
from email.mime.text import MIMEText
|
||||||
from email.utils import formataddr
|
from email.utils import formataddr as formataddr_
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
type Email = str
|
||||||
|
type NamedEmail = tuple[str, Email]
|
||||||
|
type Address = Email | list[Email] | NamedEmail | list[NamedEmail]
|
||||||
|
|
||||||
|
|
||||||
class Message:
|
class Message:
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
from_: tuple[str | None, str],
|
from_: NamedEmail,
|
||||||
to: tuple[str | None, str],
|
to: Address,
|
||||||
subject: str,
|
subject: str,
|
||||||
reply_to: tuple[str | None, str] | None = None,
|
cc: Address | None = None,
|
||||||
|
bcc: Address | None = None,
|
||||||
|
reply_to: NamedEmail | None = None,
|
||||||
content: str | None = None,
|
content: str | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self._references = set()
|
self._references = set()
|
||||||
self._body = MIMEMultipart('alternative')
|
self._body = MIMEMultipart('alternative')
|
||||||
self._message = MIMEMultipart('mixed')
|
self._message = MIMEMultipart('mixed')
|
||||||
self._message['From'] = formataddr(from_)
|
self._message['From'] = formataddr_(from_)
|
||||||
self._message['To'] = formataddr(to)
|
self._message['To'] = formataddr(to)
|
||||||
self._message['Subject'] = subject
|
self._message['Subject'] = subject
|
||||||
self._message.attach(self._body)
|
self._message.attach(self._body)
|
||||||
|
|
||||||
if reply_to:
|
if reply_to:
|
||||||
self._message['Reply-To'] = formataddr(reply_to)
|
self._message['Reply-To'] = formataddr_(reply_to)
|
||||||
|
|
||||||
|
if cc:
|
||||||
|
self._message['Cc'] = formataddr(cc)
|
||||||
|
|
||||||
|
if bcc:
|
||||||
|
self._message['Bcc'] = formataddr(bcc)
|
||||||
|
|
||||||
if content:
|
if content:
|
||||||
self.add_alternative(content, subtype='plain')
|
self.add_alternative(content, subtype='plain')
|
||||||
@@ -62,3 +74,13 @@ class Message:
|
|||||||
self._message['References'] = ' '.join(self._references)
|
self._message['References'] = ' '.join(self._references)
|
||||||
|
|
||||||
return self._message.as_bytes()
|
return self._message.as_bytes()
|
||||||
|
|
||||||
|
|
||||||
|
def formataddr(address: Address) -> str:
|
||||||
|
if isinstance(address, str):
|
||||||
|
return formataddr_((None, address))
|
||||||
|
|
||||||
|
if isinstance(address, tuple):
|
||||||
|
return formataddr_(address)
|
||||||
|
|
||||||
|
return ', '.join([formataddr(addr) for addr in address])
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "layercake"
|
name = "layercake"
|
||||||
version = "0.8.0"
|
version = "0.8.1"
|
||||||
description = "Packages shared dependencies to optimize deployment and ensure consistency across functions."
|
description = "Packages shared dependencies to optimize deployment and ensure consistency across functions."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
authors = [
|
authors = [
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from pathlib import Path
|
|||||||
import boto3
|
import boto3
|
||||||
import pytest
|
import pytest
|
||||||
from layercake.dateutils import now
|
from layercake.dateutils import now
|
||||||
from layercake.email_ import Message
|
from layercake.email_ import Message, formataddr
|
||||||
from moto import mock_aws
|
from moto import mock_aws
|
||||||
|
|
||||||
|
|
||||||
@@ -18,31 +18,65 @@ def aws_credentials():
|
|||||||
os.environ['AWS_DEFAULT_REGION'] = 'sa-east-1'
|
os.environ['AWS_DEFAULT_REGION'] = 'sa-east-1'
|
||||||
|
|
||||||
|
|
||||||
def test_send_email(aws_credentials):
|
def test_formataddr():
|
||||||
from_ = 'pytest@eduseg.com.br'
|
assert (
|
||||||
|
formataddr(('Sérgio', 'sergio@somosbeta.com.br'))
|
||||||
|
== '=?utf-8?q?S=C3=A9rgio?= <sergio@somosbeta.com.br>'
|
||||||
|
)
|
||||||
|
|
||||||
|
assert formataddr('sergio@somosbeta.com.br') == 'sergio@somosbeta.com.br'
|
||||||
|
|
||||||
|
assert (
|
||||||
|
formataddr(
|
||||||
|
[
|
||||||
|
('Sérgio', 'sergio@somosbeta.com.br'),
|
||||||
|
('Tiago', 'tiago@somosbeta.com.br'),
|
||||||
|
]
|
||||||
|
)
|
||||||
|
== '=?utf-8?q?S=C3=A9rgio?= <sergio@somosbeta.com.br>, Tiago <tiago@somosbeta.com.br>'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_email():
|
||||||
|
from_ = ('EDUSEG®', 'pytest@eduseg.com.br')
|
||||||
|
to = [
|
||||||
|
('Sérgio', 'sergio+pytest1@somosbeta.com.br'),
|
||||||
|
('Sergio', 'sergio+pytest2@somosbeta.com.br'),
|
||||||
|
]
|
||||||
|
emailmsg = Message(
|
||||||
|
to=to,
|
||||||
|
from_=from_,
|
||||||
|
bcc='osergiosiqueira@gmail.com',
|
||||||
|
subject='áéíóúnõ',
|
||||||
|
reply_to=('', 'osergiosiqueira@gmail.com'),
|
||||||
|
content='<b>Lorem ipsum dolor sit amet, consectetur adipiscing elit</b>',
|
||||||
|
)
|
||||||
|
emailmsg.add_header('X-Feedback', f'uid=123; oid=123; date={now().isoformat()}')
|
||||||
|
emailmsg.add_alternative(
|
||||||
|
"""
|
||||||
|
<h2>áéíóúnõ</h2>
|
||||||
|
html
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
emailmsg.attach(Path('tests/samples/test.pdf'), 'relatório.pdf')
|
||||||
|
|
||||||
with mock_aws():
|
with mock_aws():
|
||||||
ses = boto3.client('ses', region_name='sa-east-1')
|
ses = boto3.client('ses', region_name='sa-east-1')
|
||||||
ses.verify_email_identity(EmailAddress=from_)
|
ses.verify_email_identity(EmailAddress=from_[1])
|
||||||
to = ('Sérgio', 'sergio@somosbeta.com.br')
|
|
||||||
msg = Message(
|
|
||||||
to=to,
|
|
||||||
from_=('Beta Educação', from_),
|
|
||||||
subject='áéíóúnõ',
|
|
||||||
reply_to=('', 'osergiosiqueira@gmail.com'),
|
|
||||||
content='<b>Lorem ipsum dolor sit amet, consectetur adipiscing elit</b>',
|
|
||||||
)
|
|
||||||
msg.add_header('X-Feedback', f'uid=123; oid=123; date={now().isoformat()}')
|
|
||||||
msg.add_alternative(
|
|
||||||
"""
|
|
||||||
<h2>áéíóúnõ</h2>
|
|
||||||
html
|
|
||||||
"""
|
|
||||||
)
|
|
||||||
msg.attach(Path('tests/samples/test.pdf'), 'relatório.pdf')
|
|
||||||
|
|
||||||
assert ses.send_raw_email(
|
assert ses.send_raw_email(
|
||||||
Source=from_,
|
Source=from_[1],
|
||||||
Destinations=[to[1]],
|
Destinations=[to[0][1]],
|
||||||
RawMessage={'Data': msg.as_bytes()},
|
RawMessage={'Data': emailmsg.as_bytes()},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# sesv2 = boto3.client('sesv2')
|
||||||
|
# r = sesv2.send_email(
|
||||||
|
# Content={
|
||||||
|
# 'Raw': {
|
||||||
|
# 'Data': emailmsg.as_bytes(),
|
||||||
|
# },
|
||||||
|
# }
|
||||||
|
# )
|
||||||
|
|
||||||
|
# assert 'MessageId' in r
|
||||||
|
|||||||
Reference in New Issue
Block a user