File: test_server.py

package info (click to toggle)
python-aiosmtpd 1.2.2-1%2Bdeb11u1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 1,704 kB
  • sloc: python: 3,838; makefile: 39
file content (58 lines) | stat: -rw-r--r-- 1,949 bytes parent folder | download
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)