Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_raise_smtp_response_exception(code, message):
with pytest.raises(SMTPResponseException) as excinfo:
raise SMTPResponseException(code, message)
assert issubclass(excinfo.type, SMTPException)
assert excinfo.value.code == code
assert excinfo.value.message == message
async def test_sendmail_with_invalid_mail_option(
smtp_client, smtpd_server, sender_str, recipient_str, message_str
):
async with smtp_client:
with pytest.raises(SMTPResponseException) as excinfo:
await smtp_client.sendmail(
sender_str,
[recipient_str],
message_str,
mail_options=["BADDATA=0x00000000"],
)
assert excinfo.value.code == SMTPStatus.syntax_error
async def test_421_closes_connection(
smtp_client, smtpd_server, smtpd_class, smtpd_response_handler_factory, monkeypatch
):
response_handler = smtpd_response_handler_factory(
"{} Please come back in 15 seconds.".format(SMTPStatus.domain_unavailable)
)
monkeypatch.setattr(smtpd_class, "smtp_NOOP", response_handler)
await smtp_client.connect()
with pytest.raises(SMTPResponseException):
await smtp_client.noop()
assert not smtp_client.is_connected
writer.write(long_response)
await writer.drain()
server = await asyncio.start_server(
client_connected, host=bind_address, port=0, family=socket.AF_INET
)
server_port = server.sockets[0].getsockname()[1]
connect_future = event_loop.create_connection(
SMTPProtocol, host=hostname, port=server_port
)
_, protocol = await asyncio.wait_for(connect_future, timeout=1.0)
monkeypatch.setattr("aiosmtplib.protocol.MAX_LINE_LENGTH", 128)
with pytest.raises(SMTPResponseException) as exc_info:
await protocol.execute_command(b"TEST\n", timeout=1.0)
assert exc_info.value.code == 500
assert "Response too long" in exc_info.value.message
server.close()
await server.wait_closed()
def test_raise_smtp_recipient_refused(code, message, recipient):
with pytest.raises(SMTPRecipientRefused) as excinfo:
raise SMTPRecipientRefused(code, message, recipient)
assert issubclass(excinfo.type, SMTPResponseException)
assert excinfo.value.code == code
assert excinfo.value.message == message
assert excinfo.value.recipient == recipient
async def test_rset_error(
smtp_client,
smtpd_server,
smtpd_class,
smtpd_response_handler_factory,
monkeypatch,
error_code,
):
response_handler = smtpd_response_handler_factory("{} error".format(error_code))
monkeypatch.setattr(smtpd_class, "smtp_RSET", response_handler)
async with smtp_client:
with pytest.raises(SMTPResponseException) as exception_info:
await smtp_client.rset()
assert exception_info.value.code == error_code
smtp_client,
smtpd_server,
smtpd_class,
smtpd_response_handler_factory,
monkeypatch,
sender_str,
recipient_str,
message_str,
):
response_handler = smtpd_response_handler_factory(
"{} error".format(SMTPStatus.unrecognized_parameters), close_after=True
)
monkeypatch.setattr(smtpd_class, "smtp_DATA", response_handler)
async with smtp_client:
with pytest.raises(SMTPResponseException):
await smtp_client.sendmail(sender_str, [recipient_str], message_str)
async def test_mail_options_not_implemented(smtp_client, smtpd_server):
async with smtp_client:
with pytest.raises(SMTPResponseException):
await smtp_client.mail("j@example.com", options=["OPT=1"])
def test_raise_smtp_response_exception(code, message):
with pytest.raises(SMTPResponseException) as excinfo:
raise SMTPResponseException(code, message)
assert issubclass(excinfo.type, SMTPException)
assert excinfo.value.code == code
assert excinfo.value.message == message
smtpd_server,
event_loop,
smtpd_class,
smtpd_response_handler_factory,
monkeypatch,
error_code,
):
response_handler = smtpd_response_handler_factory("{} error".format(error_code))
monkeypatch.setattr(smtpd_class, "smtp_STARTTLS", response_handler)
async with smtp_client:
await smtp_client.ehlo()
old_extensions = copy.copy(smtp_client.esmtp_extensions)
with pytest.raises(SMTPResponseException) as exception_info:
await smtp_client.starttls(validate_certs=False)
assert exception_info.value.code == error_code
# Make sure our state has been _not_ been cleared
assert smtp_client.esmtp_extensions == old_extensions
assert smtp_client.supports_esmtp is True
# Make sure our connection was not upgraded. ssl protocol transport is
# private in UVloop, so just check the class name.
assert "SSL" not in type(smtp_client.transport).__name__