1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58
|
"""Test other aspects of the server implementation."""
import os
import socket
import unittest
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Sink
from aiosmtpd.smtp import SMTP as Server
from smtplib import SMTP
def in_wsl():
# WSL 1.0 somehow allows more than one listener on one port.
# So when testing on WSL, we must set PLATFORM=wsl and skip the
# "test_socket_error" test.
return os.environ.get("PLATFORM") == "wsl"
class TestServer(unittest.TestCase):
def test_smtp_utf8(self):
controller = Controller(Sink())
controller.start()
self.addCleanup(controller.stop)
with SMTP(controller.hostname, controller.port) as client:
code, response = client.ehlo('example.com')
self.assertEqual(code, 250)
self.assertIn(b'SMTPUTF8', response.splitlines())
def test_default_max_command_size_limit(self):
server = Server(Sink())
self.assertEqual(server.max_command_size_limit, 512)
def test_special_max_command_size_limit(self):
server = Server(Sink())
server.command_size_limits['DATA'] = 1024
self.assertEqual(server.max_command_size_limit, 1024)
@unittest.skipIf(in_wsl(), "WSL prevents socket collisions")
# See explanation in the in_wsl() function
def test_socket_error(self):
# Testing starting a server with a port already in use
s1 = Controller(Sink(), port=8025)
s2 = Controller(Sink(), port=8025)
self.addCleanup(s1.stop)
self.addCleanup(s2.stop)
s1.start()
self.assertRaises(socket.error, s2.start)
def test_server_attribute(self):
controller = Controller(Sink())
self.assertIsNone(controller.server)
try:
controller.start()
self.assertIsNotNone(controller.server)
finally:
controller.stop()
self.assertIsNone(controller.server)
|