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_mock_server_starttls_with_smtplib(preset_server):
smtp = smtplib.SMTP()
smtp._host = preset_server.hostname # Hack around smtplib SNI bug
smtp.connect(host=preset_server.hostname, port=preset_server.port)
preset_server.responses.append(b'\n'.join([
b'250-localhost, hello',
b'250-SIZE 100000',
b'250 STARTTLS',
]))
code, message = smtp.ehlo()
assert code == SMTPStatus.completed
preset_server.responses.append(b'220 ready for TLS')
code, message = smtp.starttls()
assert code == SMTPStatus.ready
# make sure our connection was actually upgraded
assert isinstance(smtp.sock, ssl.SSLSocket)
preset_server.responses.append(b'250 all good')
code, message = smtp.ehlo()
assert code == SMTPStatus.completed
async def test_helo_ok(smtp_client, smtpd_server):
async with smtp_client:
response = await smtp_client.helo()
assert response.code == SMTPStatus.completed
async def test_starttls_gmail():
client = SMTP(hostname="smtp.gmail.com", port=587, use_tls=False)
await client.connect(timeout=1.0)
await client.ehlo()
await client.starttls(validate_certs=False)
response = await client.ehlo()
assert response.code == SMTPStatus.completed
assert "smtp.gmail.com at your service" in response.message
with pytest.raises(SMTPAuthenticationError):
await client.login("test", "test")
async def test_sendmail_without_size_option(
smtp_client,
smtpd_server,
smtpd_class,
smtpd_response_handler_factory,
monkeypatch,
sender_str,
recipient_str,
message_str,
received_commands,
):
response_handler = smtpd_response_handler_factory(
"{} done".format(SMTPStatus.completed)
)
monkeypatch.setattr(smtpd_class, "smtp_EHLO", response_handler)
async with smtp_client:
errors, response = await smtp_client.sendmail(
sender_str, [recipient_str], message_str
)
assert not errors
assert response != ""
async def test_many_commands_with_gather(
monkeypatch, smtp_client, smtpd_server, smtpd_class, smtpd_response_handler_factory
):
"""
Tests that appropriate locks are in place to prevent commands confusing each other.
"""
response_handler = smtpd_response_handler_factory(
"{} Alice Smith ".format(SMTPStatus.completed)
)
monkeypatch.setattr(smtpd_class, "smtp_EXPN", response_handler)
async with smtp_client:
tasks = [
smtp_client.ehlo(),
smtp_client.helo(),
smtp_client.noop(),
smtp_client.vrfy("foo@bar.com"),
smtp_client.expn("users@example.com"),
smtp_client.mail("alice@example.com"),
smtp_client.help(),
]
results = await asyncio.gather(*tasks)
for result in results[:-1]:
async def test_rset_after_sendmail_error_response_to_mail(
smtp_client, smtpd_server, received_commands
):
"""
If an error response is given to the MAIL command in the sendmail method,
test that we reset the server session.
"""
async with smtp_client:
response = await smtp_client.ehlo()
assert response.code == SMTPStatus.completed
with pytest.raises(SMTPResponseException) as excinfo:
await smtp_client.sendmail(">foobar<", ["test@example.com"], "Hello World")
assert excinfo.value.code == SMTPStatus.unrecognized_parameters
assert received_commands[-1][0] == "RSET"
async with smtp_client:
response = await smtp_client.starttls(validate_certs=False)
assert response.code == SMTPStatus.ready
# Make sure our state has been cleared
assert not smtp_client.esmtp_extensions
assert not smtp_client.supported_auth_methods
assert not smtp_client.supports_esmtp
# Make sure our connection was actually upgraded. ssl protocol transport is
# private in UVloop, so just check the class name.
assert "SSL" in type(smtp_client.transport).__name__
response = await smtp_client.ehlo()
assert response.code == SMTPStatus.completed
async def test_helo_with_hostname(smtp_client, smtpd_server):
async with smtp_client:
response = await smtp_client.helo(hostname="example.com")
assert response.code == SMTPStatus.completed
async def test_context_manager(smtp_client, smtpd_server):
async with smtp_client:
assert smtp_client.is_connected
response = await smtp_client.noop()
assert response.code == SMTPStatus.completed
assert not smtp_client.is_connected
async def test_starttls_not_supported(
smtp_client, smtpd_server, smtpd_class, smtpd_response_handler_factory, monkeypatch
):
response_handler = smtpd_response_handler_factory(
"{} HELP".format(SMTPStatus.completed)
)
monkeypatch.setattr(smtpd_class, "smtp_EHLO", response_handler)
async with smtp_client:
await smtp_client.ehlo()
with pytest.raises(SMTPException):
await smtp_client.starttls(validate_certs=False)