From 1a1c1bb15d4659f1076c7e14a064721761d81aa6 Mon Sep 17 00:00:00 2001
From: Pandu E POLUAN <pepoluan@gmail.com>
Date: Tue, 23 Mar 2021 13:31:32 +0700
Subject: [PATCH 2/4] Code Hygiene (#259)
* Activate LOTS of flake8 plugins to enforce code hygiene
* Tune Annotation Thresholds
* Add Annotation
* Add pytest-mock to the deps of "docs"
* Fix post-rebase flake8 complaints
* Update NEWS.rst
* Move flake8 plugins into a pseudo-section in tox.ini
* Create concrete class for MessageHandler
* Bump Version to 1.5.0a2
* Experimentally enable tox-ing on 3.10
* Use typing.ByteString instead of custom AnyBytes
---
.../workflows/unit-testing-and-coverage.yml | 10 +-
aiosmtpd/__init__.py | 2 +-
aiosmtpd/controller.py | 10 +-
aiosmtpd/docs/NEWS.rst | 5 +-
aiosmtpd/docs/_exts/autoprogramm.py | 64 ++++---
aiosmtpd/docs/conf.py | 9 +-
aiosmtpd/docs/proxyprotocol.rst | 6 +-
aiosmtpd/docs/smtp.rst | 2 +-
aiosmtpd/handlers.py | 158 +++++++++++-------
aiosmtpd/lmtp.py | 6 +-
aiosmtpd/main.py | 11 +-
aiosmtpd/proxy_protocol.py | 32 ++--
aiosmtpd/qa/test_0packaging.py | 38 ++++-
aiosmtpd/qa/test_1testsuite.py | 6 +-
aiosmtpd/smtp.py | 124 ++++++++------
aiosmtpd/testing/helpers.py | 10 +-
aiosmtpd/tests/conftest.py | 48 +++---
aiosmtpd/tests/test_handlers.py | 90 +++++++---
aiosmtpd/tests/test_main.py | 16 +-
aiosmtpd/tests/test_proxyprotocol.py | 67 +++++---
aiosmtpd/tests/test_server.py | 26 +--
aiosmtpd/tests/test_smtp.py | 89 +++++-----
aiosmtpd/tests/test_starttls.py | 17 +-
housekeep.py | 3 +-
setup.cfg | 52 +++++-
tox.ini | 57 ++++++-
26 files changed, 627 insertions(+), 331 deletions(-)
diff --git a/.github/workflows/unit-testing-and-coverage.yml b/.github/workflows/unit-testing-and-coverage.yml
index f7b0e32..ebc2248 100644
--- a/.github/workflows/unit-testing-and-coverage.yml
+++ b/.github/workflows/unit-testing-and-coverage.yml
@@ -38,9 +38,17 @@ jobs:
python -m pip install --upgrade pip setuptools wheel
python setup.py develop
- name: "flake8 Style Checking"
+ shell: bash
# language=bash
run: |
- pip install colorama flake8 flake8-bugbear
+ # A bunch of flake8 plugins...
+ grab_f8_plugins=(
+ "from configparser import ConfigParser;"
+ "config = ConfigParser();"
+ "config.read('tox.ini');"
+ "print(config['flake8_plugins']['deps']);"
+ )
+ pip install colorama flake8 $(python -c "${grab_f8_plugins[*]}")
python -m flake8 aiosmtpd setup.py housekeep.py release.py
- name: "Docs Checking"
# language=bash
diff --git a/aiosmtpd/__init__.py b/aiosmtpd/__init__.py
index 9c7b938..e96d0ee 100644
--- a/aiosmtpd/__init__.py
+++ b/aiosmtpd/__init__.py
@@ -1,4 +1,4 @@
# Copyright 2014-2021 The aiosmtpd Developers
# SPDX-License-Identifier: Apache-2.0
-__version__ = "1.5.0a1"
+__version__ = "1.5.0a2"
diff --git a/aiosmtpd/controller.py b/aiosmtpd/controller.py
index d3345b8..79bdbd0 100644
--- a/aiosmtpd/controller.py
+++ b/aiosmtpd/controller.py
@@ -85,7 +85,7 @@ class _FakeServer(asyncio.StreamReaderProtocol):
factory() failed to instantiate an SMTP instance.
"""
- def __init__(self, loop):
+ def __init__(self, loop: asyncio.AbstractEventLoop):
# Imitate what SMTP does
super().__init__(
asyncio.StreamReader(loop=loop),
@@ -93,7 +93,9 @@ class _FakeServer(asyncio.StreamReaderProtocol):
loop=loop,
)
- def _client_connected_cb(self, reader, writer):
+ def _client_connected_cb(
+ self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+ ) -> None:
pass
@@ -143,7 +145,7 @@ class BaseController(metaclass=ABCMeta):
"""Subclasses can override this to customize the handler/server creation."""
return SMTP(self.handler, **self.SMTP_kwargs)
- def _factory_invoker(self):
+ def _factory_invoker(self) -> Union[SMTP, _FakeServer]:
"""Wraps factory() to catch exceptions during instantiation"""
try:
self.smtpd = self.factory()
@@ -223,7 +225,7 @@ class BaseThreadedController(BaseController, metaclass=ABCMeta):
"""
raise NotImplementedError
- def _run(self, ready_event: threading.Event):
+ def _run(self, ready_event: threading.Event) -> None:
asyncio.set_event_loop(self.loop)
try:
# Need to do two-step assignments here to ensure IDEs can properly
diff --git a/aiosmtpd/docs/NEWS.rst b/aiosmtpd/docs/NEWS.rst
index ce627a7..eea911d 100644
--- a/aiosmtpd/docs/NEWS.rst
+++ b/aiosmtpd/docs/NEWS.rst
@@ -3,8 +3,8 @@
###################
-1.5.0 (aiosmtpd-next-next)
-==========================
+1.5.0 (aiosmtpd-next)
+=====================
Added
-----
@@ -13,6 +13,7 @@ Added
Fixed/Improved
--------------
* All Controllers now have more rationale design, as they are now composited from a Base + a Mixin
+* A whole bunch of annotations
1.4.2 (2021-03-08)
diff --git a/aiosmtpd/docs/_exts/autoprogramm.py b/aiosmtpd/docs/_exts/autoprogramm.py
index 69088be..c23bd2f 100644
--- a/aiosmtpd/docs/_exts/autoprogramm.py
+++ b/aiosmtpd/docs/_exts/autoprogramm.py
@@ -32,6 +32,7 @@ import argparse
import builtins
import collections
import os
+import sphinx
from docutils import nodes
from docutils.parsers.rst import Directive
@@ -39,13 +40,13 @@ from docutils.parsers.rst.directives import unchanged
from docutils.statemachine import StringList
from functools import reduce
from sphinx.util.nodes import nested_parse_with_titles
-from typing import List
+from typing import Any, Dict, List, Optional, Tuple
__all__ = ("AutoprogrammDirective", "import_object", "scan_programs", "setup")
-def get_subparser_action(parser):
+def get_subparser_action(parser: argparse.ArgumentParser) -> argparse._SubParsersAction:
neg1_action = parser._actions[-1]
if isinstance(neg1_action, argparse._SubParsersAction):
@@ -56,7 +57,13 @@ def get_subparser_action(parser):
return a
-def scan_programs(parser, command=None, maxdepth=0, depth=0, groups=False):
+def scan_programs(
+ parser: argparse.ArgumentParser,
+ command: List[str] = None,
+ maxdepth: int = 0,
+ depth: int = 0,
+ groups: bool = False,
+):
if command is None:
command = []
@@ -79,6 +86,7 @@ def scan_programs(parser, command=None, maxdepth=0, depth=0, groups=False):
subp_action = get_subparser_action(parser)
if subp_action:
+ # noinspection PyUnresolvedReferences
choices = subp_action.choices.items()
if not (
@@ -89,11 +97,10 @@ def scan_programs(parser, command=None, maxdepth=0, depth=0, groups=False):
for cmd, sub in choices:
if isinstance(sub, argparse.ArgumentParser):
- for program in scan_programs(sub, command + [cmd], maxdepth, depth + 1):
- yield program
+ yield from scan_programs(sub, command + [cmd], maxdepth, depth + 1)
-def scan_options(actions):
+def scan_options(actions: list):
for arg in actions:
if not (arg.option_strings or isinstance(arg, argparse._SubParsersAction)):
yield format_positional_argument(arg)
@@ -103,13 +110,13 @@ def scan_options(actions):
yield format_option(arg)
-def format_positional_argument(arg):
+def format_positional_argument(arg: argparse.Action) -> Tuple[List[str], str]:
desc = (arg.help or "") % {"default": arg.default}
name = arg.metavar or arg.dest
return [name], desc
-def format_option(arg):
+def format_option(arg: argparse.Action) -> Tuple[List[str], str]:
desc = (arg.help or "") % {"default": arg.default}
if not isinstance(arg, (argparse._StoreAction, argparse._AppendAction)):
@@ -131,7 +138,7 @@ def format_option(arg):
return names, desc
-def import_object(import_name):
+def import_object(import_name: str) -> Any:
module_name, expr = import_name.split(":", 1)
try:
mod = __import__(module_name)
@@ -151,7 +158,8 @@ def import_object(import_name):
with open(f[0]) as fobj:
codestring = fobj.read()
foo = imp.new_module("foo")
- exec(codestring, foo.__dict__) # nosec
+ # noinspection BuiltinExec
+ exec(codestring, foo.__dict__) # noqa: DUO105 # nosec
sys.modules["foo"] = foo
mod = __import__("foo")
@@ -163,7 +171,7 @@ def import_object(import_name):
globals_ = builtins
if not isinstance(globals_, dict):
globals_ = globals_.__dict__
- return eval(expr, globals_, mod.__dict__) # nosec
+ return eval(expr, globals_, mod.__dict__) # noqa: DUO104 # nosec
class AutoprogrammDirective(Directive):
@@ -204,13 +212,16 @@ class AutoprogrammDirective(Directive):
if start_command:
- def get_start_cmd_parser(p):
+ def get_start_cmd_parser(
+ p: argparse.ArgumentParser,
+ ) -> argparse.ArgumentParser:
looking_for = start_command.pop(0)
action = get_subparser_action(p)
if not action:
raise ValueError("No actions for command " + looking_for)
+ # noinspection PyUnresolvedReferences
subp = action.choices[looking_for]
if start_command:
@@ -263,7 +274,7 @@ class AutoprogrammDirective(Directive):
options_adornment=options_adornment,
)
- def run(self):
+ def run(self) -> list:
node = nodes.section()
node.document = self.state.document
result = StringList()
@@ -274,17 +285,17 @@ class AutoprogrammDirective(Directive):
def render_rst(
- title,
- options,
- is_program,
- is_subgroup,
- description,
- usage,
- usage_strip,
- usage_codeblock,
- epilog,
- options_title,
- options_adornment,
+ title: str,
+ options: List[Tuple[List[str], str]],
+ is_program: bool,
+ is_subgroup: bool,
+ description: str,
+ usage: Optional[str],
+ usage_strip: bool,
+ usage_codeblock: bool,
+ epilog: str,
+ options_title: str,
+ options_adornment: str,
):
if usage_strip:
to_strip = title.rsplit(" ", 1)[0]
@@ -310,8 +321,7 @@ def render_rst(
yield ("!" if is_subgroup else "?") * len(title)
yield ""
- for line in (description or "").splitlines():
- yield line
+ yield from (description or "").splitlines()
yield ""
if usage is None:
@@ -340,7 +350,7 @@ def render_rst(
yield line or ""
-def setup(app):
+def setup(app: sphinx.application.Sphinx) -> Dict[str, Any]:
app.add_directive("autoprogramm", AutoprogrammDirective)
return {
"version": "0.2a0",
diff --git a/aiosmtpd/docs/conf.py b/aiosmtpd/docs/conf.py
index 6ee2d05..d3273f1 100644
--- a/aiosmtpd/docs/conf.py
+++ b/aiosmtpd/docs/conf.py
@@ -1,6 +1,7 @@
-# -*- coding: utf-8 -*-
-#
-# aiosmtpd documentation build configuration file, created by
+# Copyright 2014-2021 The aiosmtpd Developers
+# SPDX-License-Identifier: Apache-2.0
+
+# aiosmtpd documentation build configuration file, originally created by
# sphinx-quickstart on Fri Oct 16 12:18:52 2015.
#
# This file is execfile()d with the current directory set to its
@@ -331,5 +332,5 @@ texinfo_documents = [
# endregion
-def setup(app):
+def setup(app): # noqa: ANN001
app.add_css_file("aiosmtpd.css")
diff --git a/aiosmtpd/docs/proxyprotocol.rst b/aiosmtpd/docs/proxyprotocol.rst
index 30e01b7..eac41b0 100644
--- a/aiosmtpd/docs/proxyprotocol.rst
+++ b/aiosmtpd/docs/proxyprotocol.rst
@@ -203,7 +203,7 @@ Enums
Valid only for address family of :attr:`AF.INET` or :attr:`AF.INET6`
.. py:attribute:: rest
- :type: Union[bytes, bytearray]
+ :type: ByteString
The contents depend on the version of the PROXY header *and* (for version 2)
the address family.
@@ -374,7 +374,7 @@ Enums
.. py:classmethod:: from_raw(raw) -> Optional[ProxyTLV]
:param raw: The raw bytes containing the TLV Vectors
- :type raw: Union[bytes, bytearray]
+ :type raw: ByteString
:return: A new instance of ProxyTLV, or ``None`` if parsing failed
This triggers the parsing of raw bytes/bytearray into a ProxyTLV instance.
@@ -387,7 +387,7 @@ Enums
.. py:classmethod:: parse(chunk, partial_ok=True) -> Dict[str, Any]
:param chunk: The bytes to parse into TLV Vectors
- :type chunk: Union[bytes, bytearray]
+ :type chunk: ByteString
:param partial_ok: If ``True``, return partially-parsed TLV Vectors as is.
If ``False``, (re)raise ``MalformedTLV``
:type partial_ok: bool
diff --git a/aiosmtpd/docs/smtp.rst b/aiosmtpd/docs/smtp.rst
index 3305079..b647e32 100644
--- a/aiosmtpd/docs/smtp.rst
+++ b/aiosmtpd/docs/smtp.rst
@@ -499,7 +499,7 @@ aiosmtpd.smtp
:param challenge: The SMTP AUTH challenge to send to the client.
May be in plaintext, may be in base64. Do NOT prefix with "334 "!
- :type challenge: Union[str, bytes, bytearray]
+ :type challenge: AnyStr
:param encode_to_b64: If true, will perform base64-encoding before sending
the challenge to the client.
:type encode_to_b64: bool
diff --git a/aiosmtpd/handlers.py b/aiosmtpd/handlers.py
index ada1e91..b22821e 100644
--- a/aiosmtpd/handlers.py
+++ b/aiosmtpd/handlers.py
@@ -10,91 +10,114 @@ your own handling of messages. Implement only the methods you care about.
"""
import asyncio
+import io
import logging
import mailbox
+import os
import re
import smtplib
import sys
from abc import ABCMeta, abstractmethod
-from email import message_from_bytes, message_from_string
+from argparse import ArgumentParser
+from email.message import Message as Em_Message
+from email.parser import BytesParser, Parser
+from typing import AnyStr, Dict, List, Tuple, Type, TypeVar
from public import public
-EMPTYBYTES = b''
-COMMASPACE = ', '
-CRLF = b'\r\n'
-NLCRE = re.compile(br'\r\n|\r|\n')
-log = logging.getLogger('mail.debug')
+from aiosmtpd.smtp import SMTP as SMTPServer
+from aiosmtpd.smtp import Envelope as SMTPEnvelope
+from aiosmtpd.smtp import Session as SMTPSession
+T = TypeVar("T")
-def _format_peer(peer):
+EMPTYBYTES = b""
+COMMASPACE = ", "
+CRLF = b"\r\n"
+NLCRE = re.compile(br"\r\n|\r|\n")
+log = logging.getLogger("mail.debug")
+
+
+def _format_peer(peer: str) -> str:
# This is a separate function mostly so the test suite can craft a
# reproducible output.
- return 'X-Peer: {!r}'.format(peer)
+ return "X-Peer: {!r}".format(peer)
+
+
+def message_from_bytes(s, *args, **kws):
+ return BytesParser(*args, **kws).parsebytes(s)
+
+
+def message_from_string(s, *args, **kws):
+ return Parser(*args, **kws).parsestr(s)
@public
class Debugging:
- def __init__(self, stream=None):
+ def __init__(self, stream: io.TextIOBase = None):
self.stream = sys.stdout if stream is None else stream
@classmethod
- def from_cli(cls, parser, *args):
+ def from_cli(cls: Type[T], parser: ArgumentParser, *args) -> T:
error = False
stream = None
if len(args) == 0:
pass
elif len(args) > 1:
error = True
- elif args[0] == 'stdout':
+ elif args[0] == "stdout":
stream = sys.stdout
- elif args[0] == 'stderr':
+ elif args[0] == "stderr":
stream = sys.stderr
else:
error = True
if error:
- parser.error('Debugging usage: [stdout|stderr]')
+ parser.error("Debugging usage: [stdout|stderr]")
return cls(stream)
- def _print_message_content(self, peer, data):
+ def _print_message_content(self, peer: str, data: AnyStr) -> None:
in_headers = True
for line in data.splitlines():
# Dump the RFC 2822 headers first.
if in_headers and not line:
print(_format_peer(peer), file=self.stream)
in_headers = False
- if isinstance(data, bytes):
+ if isinstance(line, bytes):
# Avoid spurious 'str on bytes instance' warning.
- line = line.decode('utf-8', 'replace')
+ line = line.decode("utf-8", "replace")
print(line, file=self.stream)
- async def handle_DATA(self, server, session, envelope):
- print('---------- MESSAGE FOLLOWS ----------', file=self.stream)
+ async def handle_DATA(
+ self, server: SMTPServer, session: SMTPSession, envelope: SMTPEnvelope
+ ) -> str:
+ print("---------- MESSAGE FOLLOWS ----------", file=self.stream)
# Yes, actually test for truthiness since it's possible for either the
# keywords to be missing, or for their values to be empty lists.
add_separator = False
if envelope.mail_options:
- print('mail options:', envelope.mail_options, file=self.stream)
+ print("mail options:", envelope.mail_options, file=self.stream)
add_separator = True
# rcpt_options are not currently support by the SMTP class.
rcpt_options = envelope.rcpt_options
- if any(rcpt_options): # pragma: nocover
- print('rcpt options:', rcpt_options, file=self.stream)
+ if any(rcpt_options): # pragma: nocover
+ print("rcpt options:", rcpt_options, file=self.stream)
add_separator = True
if add_separator:
print(file=self.stream)
self._print_message_content(session.peer, envelope.content)
- print('------------ END MESSAGE ------------', file=self.stream)
- return '250 OK'
+ print("------------ END MESSAGE ------------", file=self.stream)
+ return "250 OK"
@public
class Proxy:
- def __init__(self, remote_hostname, remote_port):
+ def __init__(self, remote_hostname: str, remote_port: int):
self._hostname = remote_hostname
self._port = remote_port
- async def handle_DATA(self, server, session, envelope):
+ async def handle_DATA(
+ self, server: SMTPServer, session: SMTPSession, envelope: SMTPEnvelope
+ ) -> str:
if isinstance(envelope.content, str):
content = envelope.original_content
else:
@@ -107,15 +130,17 @@ class Proxy:
if NLCRE.match(line):
ending = line
break
- peer = session.peer[0].encode('ascii')
- lines.insert(_i, b'X-Peer: %s%s' % (peer, ending))
+ peer = session.peer[0].encode("ascii")
+ lines.insert(_i, b"X-Peer: " + peer + ending)
data = EMPTYBYTES.join(lines)
refused = self._deliver(envelope.mail_from, envelope.rcpt_tos, data)
# TBD: what to do with refused addresses?
- log.info('we got some refusals: %s', refused)
- return '250 OK'
+ log.info("we got some refusals: %s", refused)
+ return "250 OK"
- def _deliver(self, mail_from, rcpt_tos, data):
+ def _deliver(
+ self, mail_from: AnyStr, rcpt_tos: List[AnyStr], data: AnyStr
+ ) -> Dict[str, Tuple[int, bytes]]:
refused = {}
try:
s = smtplib.SMTP()
@@ -125,15 +150,15 @@ class Proxy:
finally:
s.quit()
except smtplib.SMTPRecipientsRefused as e:
- log.info('got SMTPRecipientsRefused')
+ log.info("got SMTPRecipientsRefused")
refused = e.recipients
except (OSError, smtplib.SMTPException) as e:
- log.exception('got %s', e.__class__)
+ log.exception("got %s", e.__class__)
# All recipients were refused. If the exception had an associated
# error code, use it. Otherwise, fake it with a non-triggering
# exception code.
- errcode = getattr(e, 'smtp_code', -1)
- errmsg = getattr(e, 'smtp_error', 'ignore')
+ errcode = getattr(e, "smtp_code", -1)
+ errmsg = getattr(e, "smtp_error", "ignore")
for r in rcpt_tos:
refused[r] = (errcode, errmsg)
return refused
@@ -142,75 +167,88 @@ class Proxy:
@public
class Sink:
@classmethod
- def from_cli(cls, parser, *args):
+ def from_cli(cls: Type[T], parser: ArgumentParser, *args) -> T:
if len(args) > 0:
- parser.error('Sink handler does not accept arguments')
+ parser.error("Sink handler does not accept arguments")
return cls()
@public
class Message(metaclass=ABCMeta):
- def __init__(self, message_class=None):
+ def __init__(self, message_class: Type[Em_Message] = None):
self.message_class = message_class
- async def handle_DATA(self, server, session, envelope):
- envelope = self.prepare_message(session, envelope)
- self.handle_message(envelope)
- return '250 OK'
+ async def handle_DATA(
+ self, server: SMTPServer, session: SMTPSession, envelope: SMTPEnvelope
+ ) -> str:
+ message = self.prepare_message(session, envelope)
+ self.handle_message(message)
+ return "250 OK"
- def prepare_message(self, session, envelope):
+ def prepare_message(
+ self, session: SMTPSession, envelope: SMTPEnvelope
+ ) -> Em_Message:
# If the server was created with decode_data True, then data will be a
# str, otherwise it will be bytes.
data = envelope.content
- if isinstance(data, bytes):
+ message: Em_Message
+ if isinstance(data, (bytes, bytearray)):
message = message_from_bytes(data, self.message_class)
- else:
- assert isinstance(data, str), (
- 'Expected str or bytes, got {}'.format(type(data)))
+ elif isinstance(data, str):
message = message_from_string(data, self.message_class)
- message['X-Peer'] = str(session.peer)
- message['X-MailFrom'] = envelope.mail_from
- message['X-RcptTo'] = COMMASPACE.join(envelope.rcpt_tos)
+ else:
+ raise TypeError(f"Expected str or bytes, got {type(data)}")
+ assert isinstance(message, Em_Message)
+ message["X-Peer"] = str(session.peer)
+ message["X-MailFrom"] = envelope.mail_from
+ message["X-RcptTo"] = COMMASPACE.join(envelope.rcpt_tos)
return message
@abstractmethod
- def handle_message(self, message):
+ def handle_message(self, message: Em_Message) -> None:
raise NotImplementedError
@public
class AsyncMessage(Message, metaclass=ABCMeta):
- def __init__(self, message_class=None, *, loop=None):
+ def __init__(
+ self,
+ message_class: Type[Em_Message] = None,
+ *,
+ loop: asyncio.AbstractEventLoop = None,
+ ):
super().__init__(message_class)
self.loop = loop or asyncio.get_event_loop()
- async def handle_DATA(self, server, session, envelope):
+ async def handle_DATA(
+ self, server: SMTPServer, session: SMTPSession, envelope: SMTPEnvelope
+ ) -> str:
message = self.prepare_message(session, envelope)
await self.handle_message(message)
- return '250 OK'
+ return "250 OK"
@abstractmethod
- async def handle_message(self, message):
+ async def handle_message(self, message: Em_Message) -> None:
raise NotImplementedError
@public
class Mailbox(Message):
- def __init__(self, mail_dir, message_class=None):
+ def __init__(self, mail_dir: os.PathLike, message_class: Type[Em_Message] = None):
self.mailbox = mailbox.Maildir(mail_dir)
self.mail_dir = mail_dir
super().__init__(message_class)
- def handle_message(self, message):
+ def handle_message(self, message: Em_Message) -> None:
self.mailbox.add(message)
- def reset(self):
+ def reset(self) -> None:
self.mailbox.clear()
@classmethod
- def from_cli(cls, parser, *args):
+ def from_cli(cls: Type[T], parser: ArgumentParser, *args) -> T:
if len(args) < 1:
- parser.error('The directory for the maildir is required')
+ parser.error("The directory for the maildir is required")
elif len(args) > 1:
- parser.error('Too many arguments for Mailbox handler')
+ parser.error("Too many arguments for Mailbox handler")
return cls(args[0])
diff --git a/aiosmtpd/lmtp.py b/aiosmtpd/lmtp.py
index 3f13af7..de68808 100644
--- a/aiosmtpd/lmtp.py
+++ b/aiosmtpd/lmtp.py
@@ -11,14 +11,14 @@ class LMTP(SMTP):
show_smtp_greeting: bool = False
@syntax('LHLO hostname')
- async def smtp_LHLO(self, arg):
+ async def smtp_LHLO(self, arg: str) -> None:
"""The LMTP greeting, used instead of HELO/EHLO."""
await super().smtp_EHLO(arg)
- async def smtp_HELO(self, arg):
+ async def smtp_HELO(self, arg: str) -> None:
"""HELO is not a valid LMTP command."""
await self.push('500 Error: command "HELO" not recognized')
- async def smtp_EHLO(self, arg):
+ async def smtp_EHLO(self, arg: str) -> None:
"""EHLO is not a valid LMTP command."""
await self.push('500 Error: command "EHLO" not recognized')
diff --git a/aiosmtpd/main.py b/aiosmtpd/main.py
index e978c60..2366ae4 100644
--- a/aiosmtpd/main.py
+++ b/aiosmtpd/main.py
@@ -7,11 +7,12 @@ import os
import signal
import ssl
import sys
-from argparse import ArgumentParser
+from argparse import ArgumentParser, Namespace
from contextlib import suppress
from functools import partial
from importlib import import_module
from pathlib import Path
+from typing import Optional, Sequence, Tuple
from public import public
@@ -167,7 +168,7 @@ def _parser() -> ArgumentParser:
return parser
-def parseargs(args=None):
+def parseargs(args: Optional[Sequence[str]] = None) -> Tuple[ArgumentParser, Namespace]:
parser = _parser()
parsed = parser.parse_args(args)
# Find the handler class.
@@ -214,7 +215,7 @@ def parseargs(args=None):
@public
-def main(args=None):
+def main(args: Optional[Sequence[str]] = None) -> None:
parser, args = parseargs(args=args)
if args.setuid: # pragma: on-win32
@@ -285,10 +286,8 @@ def main(args=None):
loop.add_signal_handler(signal.SIGINT, loop.stop)
log.debug("Starting asyncio loop")
- try:
+ with suppress(KeyboardInterrupt):
loop.run_forever()
- except KeyboardInterrupt:
- pass
server_loop.close()
log.debug("Completed asyncio loop")
loop.run_until_complete(server_loop.wait_closed())
diff --git a/aiosmtpd/proxy_protocol.py b/aiosmtpd/proxy_protocol.py
index 621098c..27d202a 100644
--- a/aiosmtpd/proxy_protocol.py
+++ b/aiosmtpd/proxy_protocol.py
@@ -1,6 +1,7 @@
# Copyright 2014-2021 The aiosmtpd Developers
# SPDX-License-Identifier: Apache-2.0
+import contextlib
import logging
import re
import struct
@@ -8,7 +9,7 @@ from collections import deque
from enum import IntEnum
from functools import partial
from ipaddress import IPv4Address, IPv6Address, ip_address
-from typing import Any, AnyStr, Dict, Optional, Tuple, Union
+from typing import Any, AnyStr, ByteString, Dict, Optional, Tuple, Union
import attr
from public import public
@@ -73,9 +74,10 @@ V2_PARSE_ADDR_FAMPRO = {
"""Family & Proto combinations that need address parsing"""
-__all__ = [
+__all__ = ["struct", "partial", "IPv4Address", "IPv6Address"]
+__all__.extend(
k for k in globals().keys() if k.startswith("V1_") or k.startswith("V2_")
-] + ["struct", "partial", "IPv4Address", "IPv6Address"]
+)
_NOT_FOUND = object()
@@ -144,10 +146,10 @@ class ProxyTLV(dict):
super().__init__(*args, **kwargs)
self.tlv_loc = _tlv_loc
- def __getattr__(self, item):
+ def __getattr__(self, item: str) -> Any:
return self.get(item)
- def __eq__(self, other):
+ def __eq__(self, other: Dict[str, Any]) -> bool:
return super().__eq__(other)
def same_attribs(self, _raises: bool = False, **kwargs) -> bool:
@@ -175,7 +177,7 @@ class ProxyTLV(dict):
@classmethod
def parse(
cls,
- data: Union[bytes, bytearray],
+ data: ByteString,
partial_ok: bool = True,
strict: bool = False,
) -> Tuple[Dict[str, Any], Dict[str, int]]:
@@ -189,7 +191,7 @@ class ProxyTLV(dict):
rslt: Dict[str, Any] = {}
tlv_loc: Dict[str, int] = {}
- def _pars(chunk: Union[bytes, bytearray], *, offset: int):
+ def _pars(chunk: ByteString, *, offset: int) -> None:
i = 0
while i < len(chunk):
typ = chunk[i]
@@ -228,7 +230,7 @@ class ProxyTLV(dict):
@classmethod
def from_raw(
- cls, raw: Union[bytes, bytearray], strict: bool = False
+ cls, raw: ByteString, strict: bool = False
) -> Optional["ProxyTLV"]:
"""
Parses raw bytes for TLV Vectors, decode them and giving them human-readable
@@ -275,7 +277,7 @@ class ProxyData:
dst_addr: Optional[EndpointAddress] = _anoinit(default=None)
src_port: Optional[int] = _anoinit(default=None)
dst_port: Optional[int] = _anoinit(default=None)
- rest: Union[bytes, bytearray] = _anoinit(default=b"")
+ rest: ByteString = _anoinit(default=b"")
"""
Rest of PROXY Protocol data following UNKNOWN (v1) or UNSPEC (v2), or containing
undecoded TLV (v2). If the latter, you can use the ProxyTLV class to parse the
@@ -302,12 +304,10 @@ class ProxyData:
return not (self.error or self.version is None or self.protocol is None)
@property
- def tlv(self):
+ def tlv(self) -> Optional[ProxyTLV]:
if self._tlv is None:
- try:
+ with contextlib.suppress(MalformedTLV):
self._tlv = ProxyTLV.from_raw(self.rest)
- except MalformedTLV:
- pass
return self._tlv
def with_error(self, error_msg: str, log_prefix: bool = True) -> "ProxyData":
@@ -340,7 +340,7 @@ class ProxyData:
return False
return True
- def __bool__(self):
+ def __bool__(self) -> bool:
return self.valid
@@ -353,7 +353,7 @@ RE_PORT_NOLEADZERO = re.compile(r"^[1-9]\d{0,4}|0$")
# Reference: https://github.com/haproxy/haproxy/blob/v2.3.0/doc/proxy-protocol.txt
-async def _get_v1(reader: AsyncReader, initial=b"") -> ProxyData:
+async def _get_v1(reader: AsyncReader, initial: ByteString = b"") -> ProxyData:
proxy_data = ProxyData(version=1)
proxy_data.whole_raw = bytearray(initial)
@@ -437,7 +437,7 @@ async def _get_v1(reader: AsyncReader, initial=b"") -> ProxyData:
return proxy_data
-async def _get_v2(reader: AsyncReader, initial=b"") -> ProxyData:
+async def _get_v2(reader: AsyncReader, initial: ByteString = b"") -> ProxyData:
proxy_data = ProxyData(version=2)
whole_raw = bytearray()
diff --git a/aiosmtpd/qa/test_0packaging.py b/aiosmtpd/qa/test_0packaging.py
index 2240762..9dbb115 100644
--- a/aiosmtpd/qa/test_0packaging.py
+++ b/aiosmtpd/qa/test_0packaging.py
@@ -2,8 +2,10 @@
# SPDX-License-Identifier: Apache-2.0
"""Test meta / packaging"""
+
import re
import subprocess
+from datetime import datetime
from itertools import tee
from pathlib import Path
@@ -15,6 +17,7 @@ from packaging import version
from aiosmtpd import __version__
RE_DUNDERVER = re.compile(r"__version__\s*?=\s*?(['\"])(?P<ver>[^'\"]+)\1\s*$")
+RE_VERHEADING = re.compile(r"(?P<ver>[0-9.]+)\s*\((?P<date>[^)]+)\)")
@pytest.fixture
@@ -23,14 +26,16 @@ def aiosmtpd_version() -> version.Version:
class TestVersion:
- def test_pep440(self, aiosmtpd_version):
+ def test_pep440(self, aiosmtpd_version: version.Version):
"""Ensure version number compliance to PEP-440"""
assert isinstance(
aiosmtpd_version, version.Version
), "Version number must comply with PEP-440"
# noinspection PyUnboundLocalVariable
- def test_ge_master(self, aiosmtpd_version, capsys):
+ def test_ge_master(
+ self, aiosmtpd_version: version.Version, capsys: pytest.CaptureFixture
+ ):
"""Ensure version is monotonically increasing"""
reference = "master:aiosmtpd/__init__.py"
cmd = f"git show {reference}".split()
@@ -50,10 +55,11 @@ class TestVersion:
assert aiosmtpd_version >= master_ver, "Version number cannot be < master's"
-class TestDocs:
- def test_NEWS_version(self, aiosmtpd_version):
- news_rst = next(Path("..").rglob("*/NEWS.rst"))
- with open(news_rst, "rt") as fin:
+class TestNews:
+ news_rst = list(Path("..").rglob("*/NEWS.rst"))[0]
+
+ def test_NEWS_version(self, aiosmtpd_version: version.Version):
+ with self.news_rst.open("rt") as fin:
# pairwise() from https://docs.python.org/3/library/itertools.html
a, b = tee(fin)
next(b, None)
@@ -73,3 +79,23 @@ class TestDocs:
f"NEWS.rst is not updated: "
f"{newsver.base_version} < {aiosmtpd_version.base_version}"
)
+
+ def test_release_date(self, aiosmtpd_version: version.Version):
+ if aiosmtpd_version.pre is not None:
+ pytest.skip("Not a release version")
+ with self.news_rst.open("rt") as fin:
+ for ln in fin:
+ ln = ln.strip()
+ m = RE_VERHEADING.match(ln)
+ if not m:
+ continue
+ ver = version.Version(m.group("ver"))
+ if ver != aiosmtpd_version:
+ continue
+ try:
+ datetime.strptime(m.group("date"), "%Y-%m-%d")
+ except ValueError:
+ pytest.fail("Release version not dated correctly")
+ break
+ else:
+ pytest.fail("Release version has no NEWS fragment")
diff --git a/aiosmtpd/qa/test_1testsuite.py b/aiosmtpd/qa/test_1testsuite.py
index e61a71d..db20c61 100644
--- a/aiosmtpd/qa/test_1testsuite.py
+++ b/aiosmtpd/qa/test_1testsuite.py
@@ -19,7 +19,7 @@ RE_ESC = re.compile(rb"(?P<digit1>\d)\.\d+\.\d+\s")
# noinspection PyUnresolvedReferences
@pytest.fixture(scope="module", autouse=True)
-def exit_on_fail(request):
+def exit_on_fail(request: pytest.FixtureRequest):
# Behavior of this will be undefined if tests are running in parallel.
# But since parallel running is not practically possible (the ports will conflict),
# then I don't think that will be a problem.
@@ -65,7 +65,9 @@ class TestStatusCodes:
f"{key}: First digit of Enhanced Status Code different from "
f"first digit of Standard Status Code"
)
- total_correct += 1
+ # Can't use enumerate(); total_correct does not increase in lockstep with
+ # the loop (there are several "continue"s above)
+ total_correct += 1 # noqa: SIM113
assert total_correct > 0
def test_commands(self):
diff --git a/aiosmtpd/smtp.py b/aiosmtpd/smtp.py
index 04a3497..b985b64 100644
--- a/aiosmtpd/smtp.py
+++ b/aiosmtpd/smtp.py
@@ -24,7 +24,9 @@ from typing import (
List,
NamedTuple,
Optional,
+ Sequence,
Tuple,
+ TypeVar,
Union,
)
from warnings import warn
@@ -39,7 +41,7 @@ from aiosmtpd.proxy_protocol import ProxyData, get_proxy
# region #### Custom Data Types #######################################################
class _Missing:
- def __repr__(self):
+ def __repr__(self) -> str:
return "MISSING"
@@ -59,6 +61,9 @@ AuthenticatorType = Callable[["SMTP", "Session", "Envelope", str, Any], "AuthRes
AuthMechanismType = Callable[["SMTP", List[str]], Awaitable[Any]]
_TriStateType = Union[None, _Missing, bytes]
+RT = TypeVar("RT") # "ReturnType"
+DecoratorType = Callable[[Callable[..., RT]], Callable[..., RT]]
+
# endregion
@@ -149,7 +154,7 @@ class LoginPassword(NamedTuple):
@public
class Session:
- def __init__(self, loop):
+ def __init__(self, loop: asyncio.AbstractEventLoop):
self.peer = None
self.ssl = None
self.host_name = None
@@ -172,7 +177,7 @@ class Session:
self.authenticated = None
@property
- def login_data(self):
+ def login_data(self) -> Any:
"""Legacy login_data, usually containing the username"""
log.warning(
"Session.login_data is deprecated and will be removed in version 2.0"
@@ -180,7 +185,7 @@ class Session:
return self._login_data
@login_data.setter
- def login_data(self, value):
+ def login_data(self, value: Any) -> None:
log.warning(
"Session.login_data is deprecated and will be removed in version 2.0"
)
@@ -189,7 +194,7 @@ class Session:
@public
class Envelope:
- def __init__(self):
+ def __init__(self) -> None:
self.mail_from = None
self.mail_options = []
self.smtp_utf8 = False
@@ -202,12 +207,14 @@ class Envelope:
# This is here to enable debugging output when the -E option is given to the
# unit test suite. In that case, this function is mocked to set the debug
# level on the loop (as if PYTHONASYNCIODEBUG=1 were set).
-def make_loop():
+def make_loop() -> asyncio.AbstractEventLoop:
return asyncio.get_event_loop()
@public
-def syntax(text, extended=None, when: Optional[str] = None):
+def syntax(
+ text: str, extended: str = None, when: Optional[str] = None
+) -> DecoratorType:
"""
A @decorator that provides helptext for (E)SMTP HELP.
Applies for smtp_* methods only!
@@ -217,7 +224,7 @@ def syntax(text, extended=None, when: Optional[str] = None):
:param when: The name of the attribute of SMTP class to check; if the value
of the attribute is false-y then HELP will not be available for the command
"""
- def decorator(f):
+ def decorator(f: Callable[..., RT]) -> Callable[..., RT]:
f.__smtp_syntax__ = text
f.__smtp_syntax_extended__ = extended
f.__smtp_syntax_when__ = when
@@ -226,7 +233,7 @@ def syntax(text, extended=None, when: Optional[str] = None):
@public
-def auth_mechanism(actual_name: str):
+def auth_mechanism(actual_name: str) -> DecoratorType:
"""
A @decorator to explicitly specifies the name of the AUTH mechanism implemented by
the function/method this decorates
@@ -234,9 +241,10 @@ def auth_mechanism(actual_name: str):
:param actual_name: Name of AUTH mechanism. Must consists of [A-Z0-9_-] only.
Will be converted to uppercase
"""
- def decorator(f):
+ def decorator(f: Callable[..., RT]) -> Callable[..., RT]:
f.__auth_mechanism_name__ = actual_name
return f
+
actual_name = actual_name.upper()
if not VALID_AUTHMECH.match(actual_name):
raise ValueError(f"Invalid AUTH mechanism name: {actual_name}")
@@ -249,7 +257,7 @@ def login_always_fail(
return False
-def is_int(o):
+def is_int(o: Any) -> bool:
return isinstance(o, int)
@@ -267,7 +275,7 @@ def sanitize(text: bytes) -> bytes:
@public
-def sanitized_log(func: Callable, msg: AnyStr, *args, **kwargs):
+def sanitized_log(func: Callable[..., None], msg: AnyStr, *args, **kwargs) -> None:
"""
Sanitize args before passing to a logging function.
"""
@@ -305,24 +313,24 @@ class SMTP(asyncio.StreamReaderProtocol):
def __init__(
self,
- handler,
+ handler: Any,
*,
- data_size_limit=DATA_SIZE_DEFAULT,
- enable_SMTPUTF8=False,
- decode_data=False,
- hostname=None,
- ident=None,
+ data_size_limit: int = DATA_SIZE_DEFAULT,
+ enable_SMTPUTF8: bool = False,
+ decode_data: bool = False,
+ hostname: str = None,
+ ident: str = None,
tls_context: Optional[ssl.SSLContext] = None,
- require_starttls=False,
- timeout=300,
- auth_required=False,
- auth_require_tls=True,
+ require_starttls: bool = False,
+ timeout: float = 300,
+ auth_required: bool = False,
+ auth_require_tls: bool = True,
auth_exclude_mechanism: Optional[Iterable[str]] = None,
auth_callback: AuthCallbackType = None,
command_call_limit: Union[int, Dict[str, int], None] = None,
authenticator: AuthenticatorType = None,
proxy_protocol_timeout: Optional[Union[int, float]] = None,
- loop=None
+ loop: asyncio.AbstractEventLoop = None
):
self.__ident__ = ident or __ident__
self.loop = loop if loop else make_loop()
@@ -343,7 +351,7 @@ class SMTP(asyncio.StreamReaderProtocol):
self.tls_context = tls_context
if tls_context:
if (tls_context.verify_mode
- not in {ssl.CERT_NONE, ssl.CERT_OPTIONAL}):
+ not in {ssl.CERT_NONE, ssl.CERT_OPTIONAL}): # noqa: DUO122
log.warning("tls_context.verify_mode not in {CERT_NONE, "
"CERT_OPTIONAL}; this might cause client "
"connection problems")
@@ -452,13 +460,13 @@ class SMTP(asyncio.StreamReaderProtocol):
else:
raise TypeError("command_call_limit must be int or Dict[str, int]")
- def _create_session(self):
+ def _create_session(self) -> Session:
return Session(self.loop)
- def _create_envelope(self):
+ def _create_envelope(self) -> Envelope:
return Envelope()
- async def _call_handler_hook(self, command, *args):
+ async def _call_handler_hook(self, command: str, *args) -> Any:
hook = self._handle_hooks.get(command)
if hook is None:
return MISSING
@@ -466,7 +474,7 @@ class SMTP(asyncio.StreamReaderProtocol):
return status
@property
- def max_command_size_limit(self):
+ def max_command_size_limit(self) -> int:
try:
return max(self.command_size_limits.values())
except ValueError:
@@ -484,7 +492,7 @@ class SMTP(asyncio.StreamReaderProtocol):
if closed.done() and not closed.cancelled():
closed.exception()
- def connection_made(self, transport):
+ def connection_made(self, transport: asyncio.transports.Transport) -> None:
# Reset state due to rfc3207 part 4.2.
self._set_rset_state()
self.session = self._create_session()
@@ -513,7 +521,7 @@ class SMTP(asyncio.StreamReaderProtocol):
self._handler_coroutine = self.loop.create_task(
self._handle_client())
- def connection_lost(self, error):
+ def connection_lost(self, error: Optional[Exception]) -> None:
log.info('%r connection lost', self.session.peer)
self._timeout_handle.cancel()
# If STARTTLS was issued, then our transport is the SSL protocol
@@ -527,7 +535,7 @@ class SMTP(asyncio.StreamReaderProtocol):
self._handler_coroutine.cancel()
self.transport = None
- def eof_received(self):
+ def eof_received(self) -> bool:
log.info('%r EOF received', self.session.peer)
self._handler_coroutine.cancel()
if self.session.ssl is not None:
@@ -537,7 +545,7 @@ class SMTP(asyncio.StreamReaderProtocol):
return False
return super().eof_received()
- def _reset_timeout(self, duration=None):
+ def _reset_timeout(self, duration: float = None) -> None:
if self._timeout_handle is not None:
self._timeout_handle.cancel()
self._timeout_handle = self.loop.call_later(
@@ -552,7 +560,9 @@ class SMTP(asyncio.StreamReaderProtocol):
# up state.
self.transport.close()
- def _client_connected_cb(self, reader, writer):
+ def _client_connected_cb(
+ self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+ ):
# This is redundant since we subclass StreamReaderProtocol, but I like
# the shorter names.
self._reader = reader
@@ -577,7 +587,7 @@ class SMTP(asyncio.StreamReaderProtocol):
log.debug("%r << %r", self.session.peer, response)
await self._writer.drain()
- async def handle_exception(self, error):
+ async def handle_exception(self, error: Exception) -> str:
if hasattr(self.event_handler, 'handle_exception'):
status = await self.event_handler.handle_exception(error)
return status
@@ -678,9 +688,11 @@ class SMTP(asyncio.StreamReaderProtocol):
await self.push('500 Error: strict ASCII mode')
# Should we await self.handle_exception()?
continue
- max_sz = (self.command_size_limits[command]
- if self.session.extended_smtp
- else self.command_size_limit)
+ max_sz = (
+ self.command_size_limits[command]
+ if self.session.extended_smtp
+ else self.command_size_limit
+ )
if len(line) > max_sz:
await self.push('500 Command line too long')
continue
@@ -720,7 +732,8 @@ class SMTP(asyncio.StreamReaderProtocol):
self.transport.close()
continue
await self.push(
- '500 Error: command "%s" not recognized' % command)
+ f'500 Error: command "{command}" not recognized'
+ )
continue
# Received a valid command, reset the timer.
@@ -785,7 +798,7 @@ class SMTP(asyncio.StreamReaderProtocol):
# SMTP and ESMTP commands
@syntax('HELO hostname')
- async def smtp_HELO(self, hostname):
+ async def smtp_HELO(self, hostname: str):
if not hostname:
await self.push('501 Syntax: HELO hostname')
return
@@ -798,7 +811,7 @@ class SMTP(asyncio.StreamReaderProtocol):
await self.push(status)
@syntax('EHLO hostname')
- async def smtp_EHLO(self, hostname):
+ async def smtp_EHLO(self, hostname: str):
if not hostname:
await self.push('501 Syntax: EHLO hostname')
return
@@ -806,9 +819,9 @@ class SMTP(asyncio.StreamReaderProtocol):
response = []
self._set_rset_state()
self.session.extended_smtp = True
- response.append('250-%s' % self.hostname)
+ response.append('250-' + self.hostname)
if self.data_size_limit:
- response.append('250-SIZE %s' % self.data_size_limit)
+ response.append(f'250-SIZE {self.data_size_limit}')
self.command_size_limits['MAIL'] += 26
if not self._decode_data:
response.append('250-8BITMIME')
@@ -848,12 +861,12 @@ class SMTP(asyncio.StreamReaderProtocol):
await self.push(r)
@syntax('NOOP [ignored]')
- async def smtp_NOOP(self, arg):
+ async def smtp_NOOP(self, arg: str):
status = await self._call_handler_hook('NOOP', arg)
await self.push('250 OK' if status is MISSING else status)
@syntax('QUIT')
- async def smtp_QUIT(self, arg):
+ async def smtp_QUIT(self, arg: str):
if arg:
await self.push('501 Syntax: QUIT')
else:
@@ -863,7 +876,7 @@ class SMTP(asyncio.StreamReaderProtocol):
self.transport.close()
@syntax('STARTTLS', when='tls_context')
- async def smtp_STARTTLS(self, arg):
+ async def smtp_STARTTLS(self, arg: str):
if arg:
await self.push('501 Syntax: STARTTLS')
return
@@ -1032,7 +1045,7 @@ class SMTP(asyncio.StreamReaderProtocol):
encode_to_b64=False,
)
- def _authenticate(self, mechanism, auth_data) -> AuthResult:
+ def _authenticate(self, mechanism: str, auth_data: Any) -> AuthResult:
if self._authenticator is not None:
# self.envelope is likely still empty, but we'll pass it anyways to
# make the invocation similar to the one in _call_handler_hook
@@ -1093,7 +1106,7 @@ class SMTP(asyncio.StreamReaderProtocol):
assert password is not None
return self._authenticate("PLAIN", LoginPassword(login, password))
- async def auth_LOGIN(self, _, args: List[str]):
+ async def auth_LOGIN(self, _, args: List[str]) -> AuthResult:
login: _TriStateType
if len(args) == 1:
# Client sent only "AUTH LOGIN"
@@ -1117,13 +1130,13 @@ class SMTP(asyncio.StreamReaderProtocol):
return self._authenticate("LOGIN", LoginPassword(login, password))
- def _strip_command_keyword(self, keyword, arg):
+ def _strip_command_keyword(self, keyword: str, arg: str) -> Optional[str]:
keylen = len(keyword)
if arg[:keylen].upper() == keyword:
return arg[keylen:].strip()
return None
- def _getaddr(self, arg) -> Tuple[Optional[str], Optional[str]]:
+ def _getaddr(self, arg: str) -> Tuple[Optional[str], Optional[str]]:
"""
Try to parse address given in SMTP command.
@@ -1145,7 +1158,9 @@ class SMTP(asyncio.StreamReaderProtocol):
return None, None
return address, rest
- def _getparams(self, params):
+ def _getparams(
+ self, params: Sequence[str]
+ ) -> Optional[Dict[str, Union[str, bool]]]:
# Return params as dictionary. Return None if not all parameters
# appear to be syntactically valid according to RFC 1869.
result = {}
@@ -1156,7 +1171,8 @@ class SMTP(asyncio.StreamReaderProtocol):
result[param] = value if eq else True
return result
- def _syntax_available(self, method):
+ # noinspection PyUnresolvedReferences
+ def _syntax_available(self, method: Callable) -> bool:
if not hasattr(method, '__smtp_syntax__'):
return False
if method.__smtp_syntax_when__:
@@ -1193,7 +1209,7 @@ class SMTP(asyncio.StreamReaderProtocol):
if arg:
address, params = self._getaddr(arg)
if address is None:
- await self.push('502 Could not VRFY %s' % arg)
+ await self.push('502 Could not VRFY ' + arg)
else:
status = await self._call_handler_hook('VRFY', address)
await self.push(
@@ -1314,7 +1330,7 @@ class SMTP(asyncio.StreamReaderProtocol):
await self.push(status)
@syntax('RSET')
- async def smtp_RSET(self, arg):
+ async def smtp_RSET(self, arg: str):
if arg:
await self.push('501 Syntax: RSET')
return
@@ -1458,5 +1474,5 @@ class SMTP(asyncio.StreamReaderProtocol):
await self.push('250 OK' if status is MISSING else status)
# Commands that have not been implemented.
- async def smtp_EXPN(self, arg):
+ async def smtp_EXPN(self, arg: str):
await self.push('502 EXPN not implemented')
diff --git a/aiosmtpd/testing/helpers.py b/aiosmtpd/testing/helpers.py
index 7fa62a2..2328704 100644
--- a/aiosmtpd/testing/helpers.py
+++ b/aiosmtpd/testing/helpers.py
@@ -12,7 +12,7 @@ import time
from smtplib import SMTP as SMTP_Client
from typing import List
-from aiosmtpd.smtp import Envelope
+from aiosmtpd.smtp import Envelope, Session, SMTP
ASYNCIO_CATCHUP_DELAY = float(os.environ.get("ASYNCIO_CATCHUP_DELAY", 0.1))
"""
@@ -52,12 +52,14 @@ class ReceivingHandler:
def __init__(self):
self.box = []
- async def handle_DATA(self, server, session, envelope):
+ async def handle_DATA(
+ self, server: SMTP, session: Session, envelope: Envelope
+ ) -> str:
self.box.append(envelope)
return "250 OK"
-def catchup_delay(delay=ASYNCIO_CATCHUP_DELAY):
+def catchup_delay(delay: float = ASYNCIO_CATCHUP_DELAY):
"""
Sleep for awhile to give asyncio's event loop time to catch up.
"""
@@ -65,7 +67,7 @@ def catchup_delay(delay=ASYNCIO_CATCHUP_DELAY):
def send_recv(
- sock: socket.socket, data: bytes, end: bytes = b"\r\n", timeout=0.1
+ sock: socket.socket, data: bytes, end: bytes = b"\r\n", timeout: float = 0.1
) -> bytes:
sock.send(data + end)
slist = [sock]
diff --git a/aiosmtpd/tests/conftest.py b/aiosmtpd/tests/conftest.py
index d0a6cd3..859d5ef 100644
--- a/aiosmtpd/tests/conftest.py
+++ b/aiosmtpd/tests/conftest.py
@@ -8,10 +8,11 @@ import ssl
from contextlib import suppress
from functools import wraps
from smtplib import SMTP as SMTPClient
-from typing import Generator, NamedTuple, Optional, Type
+from typing import Any, Callable, Generator, NamedTuple, Optional, Type, TypeVar
import pytest
from pkg_resources import resource_filename
+from pytest_mock import MockFixture
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Sink
@@ -50,6 +51,9 @@ class HostPort(NamedTuple):
port: int = 8025
+RT = TypeVar("RT") # "ReturnType"
+
+
# endregion
@@ -79,15 +83,13 @@ SERVER_KEY = resource_filename("aiosmtpd.tests.certs", "server.key")
# autouse=True and scope="session" automatically apply this fixture to ALL test cases
@pytest.fixture(autouse=True, scope="session")
-def cache_fqdn(session_mocker):
+def cache_fqdn(session_mocker: MockFixture):
"""
This fixture "caches" the socket.getfqdn() call. VERY necessary to prevent
situations where quick repeated getfqdn() causes extreme slowdown. Probably due to
the DNS server thinking it was an attack or something.
"""
session_mocker.patch("socket.getfqdn", return_value=Global.FQDN)
- #
- yield
# endregion
@@ -97,7 +99,7 @@ def cache_fqdn(session_mocker):
@pytest.fixture
-def get_controller(request):
+def get_controller(request: pytest.FixtureRequest) -> Callable[..., Controller]:
"""
Provides a function that will return an instance of a controller.
@@ -122,7 +124,7 @@ def get_controller(request):
markerdata = {}
def getter(
- handler,
+ handler: Any,
class_: Optional[Type[Controller]] = None,
**server_kwargs,
) -> Controller:
@@ -154,7 +156,7 @@ def get_controller(request):
@pytest.fixture
-def get_handler(request):
+def get_handler(request: pytest.FixtureRequest) -> Callable:
"""
Provides a function that will return an instance of
a :ref:`handler class <handlers>`.
@@ -179,7 +181,7 @@ def get_handler(request):
else:
markerdata = {}
- def getter(*args, **kwargs):
+ def getter(*args, **kwargs) -> Any:
if marker:
class_ = markerdata.pop("class_", default_class)
# *args overrides args_ in handler_data()
@@ -209,18 +211,22 @@ def temp_event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
@pytest.fixture
-def autostop_loop(temp_event_loop) -> Generator[asyncio.AbstractEventLoop, None, None]:
+def autostop_loop(
+ temp_event_loop: asyncio.AbstractEventLoop,
+) -> asyncio.AbstractEventLoop:
# Create a new event loop, and arrange for that loop to end almost
# immediately. This will allow the calls to main() in these tests to
# also exit almost immediately. Otherwise, the foreground test
# process will hang.
temp_event_loop.call_later(AUTOSTOP_DELAY, temp_event_loop.stop)
#
- yield temp_event_loop
+ return temp_event_loop
@pytest.fixture
-def plain_controller(get_handler, get_controller) -> Generator[Controller, None, None]:
+def plain_controller(
+ get_handler: Callable, get_controller: Callable
+) -> Generator[Controller, None, None]:
"""
Returns a Controller that, by default, gets invoked with no optional args.
Hence the moniker "plain".
@@ -246,7 +252,7 @@ def plain_controller(get_handler, get_controller) -> Generator[Controller, None,
@pytest.fixture
def nodecode_controller(
- get_handler, get_controller
+ get_handler: Callable, get_controller: Callable
) -> Generator[Controller, None, None]:
"""
Same as :fixture:`plain_controller`,
@@ -268,7 +274,7 @@ def nodecode_controller(
@pytest.fixture
def decoding_controller(
- get_handler, get_controller
+ get_handler: Callable, get_controller: Callable
) -> Generator[Controller, None, None]:
handler = get_handler()
controller = get_controller(handler, decode_data=True)
@@ -285,7 +291,7 @@ def decoding_controller(
@pytest.fixture
-def client(request) -> Generator[SMTPClient, None, None]:
+def client(request: pytest.FixtureRequest) -> Generator[SMTPClient, None, None]:
"""
Generic SMTP Client,
will connect to the ``host:port`` defined in ``Global.SrvAddr``
@@ -302,7 +308,7 @@ def client(request) -> Generator[SMTPClient, None, None]:
@pytest.fixture
-def ssl_context_server() -> Generator[ssl.SSLContext, None, None]:
+def ssl_context_server() -> ssl.SSLContext:
"""
Provides a server-side SSL Context
"""
@@ -310,11 +316,11 @@ def ssl_context_server() -> Generator[ssl.SSLContext, None, None]:
context.check_hostname = False
context.load_cert_chain(SERVER_CRT, SERVER_KEY)
#
- yield context
+ return context
@pytest.fixture
-def ssl_context_client() -> Generator[ssl.SSLContext, None, None]:
+def ssl_context_client() -> ssl.SSLContext:
"""
Provides a client-side SSL Context
"""
@@ -322,14 +328,14 @@ def ssl_context_client() -> Generator[ssl.SSLContext, None, None]:
context.check_hostname = False
context.load_verify_locations(SERVER_CRT)
#
- yield context
+ return context
# Please keep the scope as "module"; setting it as "function" (the default) somehow
# causes the 'hidden' exception to be detected when the loop starts over in the next
# test case, defeating the silencing.
@pytest.fixture(scope="module")
-def silence_event_loop_closed():
+def silence_event_loop_closed() -> bool:
"""
Mostly used to suppress "unhandled exception" error due to
``_ProactorBasePipeTransport`` raising an exception when doing ``__del__``
@@ -341,9 +347,9 @@ def silence_event_loop_closed():
return True
# From: https://github.com/aio-libs/aiohttp/issues/4324#issuecomment-733884349
- def silencer(func):
+ def silencer(func: Callable[..., RT]) -> Callable[..., RT]:
@wraps(func)
- def wrapper(self, *args, **kwargs):
+ def wrapper(self: Any, *args, **kwargs) -> RT:
try:
return func(self, *args, **kwargs)
except RuntimeError as e:
diff --git a/aiosmtpd/tests/test_handlers.py b/aiosmtpd/tests/test_handlers.py
index 51e06ce..35bd661 100644
--- a/aiosmtpd/tests/test_handlers.py
+++ b/aiosmtpd/tests/test_handlers.py
@@ -3,6 +3,7 @@
import logging
import sys
+from email.message import Message as Em_Message
from io import StringIO
from mailbox import Maildir
from operator import itemgetter
@@ -10,14 +11,16 @@ from pathlib import Path
from smtplib import SMTPDataError, SMTPRecipientsRefused
from textwrap import dedent
from types import SimpleNamespace
-from typing import AnyStr, Generator, Type, TypeVar, Union
+from typing import AnyStr, Callable, Generator, Type, TypeVar, Union
import pytest
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import AsyncMessage, Debugging, Mailbox, Proxy, Sink
+from aiosmtpd.handlers import Message as AbstractMessageHandler
from aiosmtpd.smtp import SMTP as Server
from aiosmtpd.smtp import Session as ServerSession
+from aiosmtpd.smtp import Envelope
from aiosmtpd.testing.statuscodes import SMTP_STATUS_CODES as S
from aiosmtpd.testing.statuscodes import StatusCode
@@ -54,7 +57,7 @@ class FakeParser:
message: AnyStr = None
- def error(self, message):
+ def error(self, message: AnyStr):
self.message = message
raise SystemExit
@@ -63,16 +66,23 @@ class DataHandler:
content: AnyStr = None
original_content: bytes = None
- async def handle_DATA(self, server, session, envelope):
+ async def handle_DATA(
+ self, server: Server, session: ServerSession, envelope: Envelope
+ ) -> str:
self.content = envelope.content
self.original_content = envelope.original_content
return S.S250_OK.to_str()
+class MessageHandler(AbstractMessageHandler):
+ def handle_message(self, message: Em_Message) -> None:
+ pass
+
+
class AsyncMessageHandler(AsyncMessage):
- handled_message = None
+ handled_message: Em_Message = None
- async def handle_message(self, message):
+ async def handle_message(self, message: Em_Message) -> None:
self.handled_message = message
@@ -209,14 +219,13 @@ def debugging_controller(get_controller) -> Generator[Controller, None, None]:
@pytest.fixture
-def temp_maildir(tmp_path: Path) -> Generator[Path, None, None]:
- maildir_path = tmp_path / "maildir"
- yield maildir_path
+def temp_maildir(tmp_path: Path) -> Path:
+ return tmp_path / "maildir"
@pytest.fixture
def mailbox_controller(
- temp_maildir, get_controller
+ temp_maildir, get_controller
) -> Generator[Controller, None, None]:
handler = Mailbox(temp_maildir)
controller = get_controller(handler)
@@ -229,7 +238,7 @@ def mailbox_controller(
@pytest.fixture
-def with_fake_parser():
+def with_fake_parser() -> Callable:
"""
Gets a function that will instantiate a handler_class using the class's
from_cli() @classmethod, using FakeParser as the parser.
@@ -250,7 +259,7 @@ def with_fake_parser():
handler = SimpleNamespace(fparser=parser, exception=type(e))
return handler
- yield handler_initer
+ return handler_initer
@pytest.fixture
@@ -435,6 +444,43 @@ class TestDebugging:
class TestMessage:
+ @pytest.mark.parametrize(
+ "content",
+ [
+ b"",
+ bytearray(),
+ "",
+ ],
+ ids=["bytes", "bytearray", "str"]
+ )
+ def test_prepare_message(self, temp_event_loop, content):
+ sess_ = ServerSession(temp_event_loop)
+ enve_ = Envelope()
+ handler = MessageHandler()
+ enve_.content = content
+ msg = handler.prepare_message(sess_, enve_)
+ assert isinstance(msg, Em_Message)
+ assert msg.keys() == ['X-Peer', 'X-MailFrom', 'X-RcptTo']
+ assert msg.get_payload() == ""
+
+ @pytest.mark.parametrize(
+ ("content", "expectre"),
+ [
+ (None, r"Expected str or bytes, got <class 'NoneType'>"),
+ ([], r"Expected str or bytes, got <class 'list'>"),
+ ({}, r"Expected str or bytes, got <class 'dict'>"),
+ ((), r"Expected str or bytes, got <class 'tuple'>"),
+ ],
+ ids=("None", "List", "Dict", "Tuple")
+ )
+ def test_prepare_message_err(self, temp_event_loop, content, expectre):
+ sess_ = ServerSession(temp_event_loop)
+ enve_ = Envelope()
+ handler = MessageHandler()
+ enve_.content = content
+ with pytest.raises(TypeError, match=expectre):
+ _ = handler.prepare_message(sess_, enve_)
+
@handler_data(class_=DataHandler)
def test_message(self, plain_controller, client):
handler = plain_controller.handler
@@ -585,11 +631,8 @@ class TestMailbox:
# Check the messages in the mailbox.
mailbox = Maildir(temp_maildir)
messages = sorted(mailbox, key=itemgetter("message-id"))
- assert list(message["message-id"] for message in messages) == [
- "<ant>",
- "<bee>",
- "<cat>",
- ]
+ expect = ["<ant>", "<bee>", "<cat>"]
+ assert [message["message-id"] for message in messages] == expect
def test_mailbox_reset(self, temp_maildir, mailbox_controller, client):
client.sendmail(
@@ -766,7 +809,6 @@ class TestProxyMocked:
def patch_smtp_oserror(self, mocker):
mock = mocker.patch("aiosmtpd.handlers.smtplib.SMTP")
mock().sendmail.side_effect = OSError
- yield
def test_oserror(
self, caplog, patch_smtp_oserror, proxy_decoding_controller, client
@@ -804,13 +846,13 @@ class TestHooks:
def test_hook_EHLO_deprecated_warning(self):
with pytest.warns(
- DeprecationWarning,
- match=(
- # Is a regex; escape regex special chars if necessary
- r"Use the 5-argument handle_EHLO\(\) hook instead of the "
- r"4-argument handle_EHLO\(\) hook; support for the 4-argument "
- r"handle_EHLO\(\) hook will be removed in version 2.0"
- )
+ DeprecationWarning,
+ match=(
+ # Is a regex; escape regex special chars if necessary
+ r"Use the 5-argument handle_EHLO\(\) hook instead of the "
+ r"4-argument handle_EHLO\(\) hook; support for the 4-argument "
+ r"handle_EHLO\(\) hook will be removed in version 2.0"
+ ),
):
_ = Server(EHLOHandlerDeprecated())
diff --git a/aiosmtpd/tests/test_main.py b/aiosmtpd/tests/test_main.py
index 36992f3..e6b3868 100644
--- a/aiosmtpd/tests/test_main.py
+++ b/aiosmtpd/tests/test_main.py
@@ -7,11 +7,13 @@ import multiprocessing as MP
import os
import time
from contextlib import contextmanager
+from multiprocessing.synchronize import Event as MP_Event
from smtplib import SMTP as SMTPClient
from smtplib import SMTP_SSL
from typing import Generator
import pytest
+from pytest_mock import MockFixture
from aiosmtpd import __version__
from aiosmtpd.handlers import Debugging
@@ -33,7 +35,7 @@ MAIL_LOG = logging.getLogger("mail.log")
class FromCliHandler:
- def __init__(self, called):
+ def __init__(self, called: bool):
self.called = called
@classmethod
@@ -63,14 +65,12 @@ def nobody_uid() -> Generator[int, None, None]:
@pytest.fixture
-def setuid(mocker):
+def setuid(mocker: MockFixture):
if not HAS_SETUID:
pytest.skip("setuid is unavailable")
mocker.patch("aiosmtpd.main.pwd", None)
mocker.patch("os.setuid", side_effect=PermissionError)
mocker.patch("aiosmtpd.main.partial", side_effect=RuntimeError)
- #
- yield
# endregion
@@ -78,7 +78,7 @@ def setuid(mocker):
# region ##### Helper Funcs ###########################################################
-def watch_for_tls(ready_flag, retq: MP.Queue):
+def watch_for_tls(ready_flag: MP_Event, retq: MP.Queue):
has_tls = False
req_tls = False
ready_flag.set()
@@ -100,7 +100,7 @@ def watch_for_tls(ready_flag, retq: MP.Queue):
retq.put(req_tls)
-def watch_for_smtps(ready_flag, retq: MP.Queue):
+def watch_for_smtps(ready_flag: MP_Event, retq: MP.Queue):
has_smtps = False
ready_flag.set()
start = time.monotonic()
@@ -276,7 +276,7 @@ class TestParseArgs:
)
@pytest.mark.parametrize(
- "args, exp_host, exp_port",
+ ("args", "exp_host", "exp_port"),
[
((), "localhost", 8025),
(("-l", "foo:25"), "foo", 25),
@@ -333,7 +333,7 @@ class TestParseArgs:
assert args.requiretls is False
@pytest.mark.parametrize(
- "certfile, keyfile, expect",
+ ("certfile", "keyfile", "expect"),
[
("x", "x", "Cert file x not found"),
(SERVER_CRT, "x", "Key file x not found"),
diff --git a/aiosmtpd/tests/test_proxyprotocol.py b/aiosmtpd/tests/test_proxyprotocol.py
index bf7f939..ad9dc9a 100644
--- a/aiosmtpd/tests/test_proxyprotocol.py
+++ b/aiosmtpd/tests/test_proxyprotocol.py
@@ -10,12 +10,12 @@ import socket
import struct
import time
from base64 import b64decode
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
from functools import partial
from ipaddress import IPv4Address, IPv6Address
from smtplib import SMTP as SMTPClient
from smtplib import SMTPServerDisconnected
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
import pytest
from pytest_mock import MockFixture
@@ -35,6 +35,7 @@ from aiosmtpd.proxy_protocol import (
)
from aiosmtpd.smtp import SMTP as SMTPServer
from aiosmtpd.smtp import Session as SMTPSession
+from aiosmtpd.smtp import Envelope as SMTPEnvelope
from aiosmtpd.tests.conftest import Global, controller_data, handler_data
DEFAULT_AUTOCANCEL = 0.1
@@ -94,13 +95,19 @@ HANDSHAKES = {
class ProxyPeekerHandler(Sink):
- def __init__(self, retval=True):
+ def __init__(self, retval: bool = True):
self.called = False
self.sessions: List[SMTPSession] = []
self.proxy_datas: List[ProxyData] = []
self.retval = retval
- async def handle_PROXY(self, server, session, envelope, proxy_data):
+ async def handle_PROXY(
+ self,
+ server: SMTPServer,
+ session: SMTPSession,
+ envelope: SMTPEnvelope,
+ proxy_data: ProxyData,
+ ) -> bool:
self.called = True
self.sessions.append(session)
self.proxy_datas.append(proxy_data)
@@ -113,7 +120,9 @@ def does_not_raise():
@pytest.fixture
-def setup_proxy_protocol(mocker: MockFixture, temp_event_loop):
+def setup_proxy_protocol(
+ mocker: MockFixture, temp_event_loop: asyncio.AbstractEventLoop
+) -> Callable:
proxy_timeout = 1.0
responses = []
transport = mocker.Mock()
@@ -129,16 +138,14 @@ def setup_proxy_protocol(mocker: MockFixture, temp_event_loop):
def runner(stop_after: float = DEFAULT_AUTOCANCEL):
loop.call_later(stop_after, protocol._handler_coroutine.cancel)
- try:
+ with suppress(asyncio.CancelledError):
loop.run_until_complete(protocol._handler_coroutine)
- except asyncio.CancelledError:
- pass
test_obj.protocol = protocol
test_obj.runner = runner
test_obj.transport = transport
- yield getter
+ return getter
class _TestProxyProtocolCommon:
@@ -303,7 +310,7 @@ class TestProxyTLV:
(None, "wrongname"),
],
)
- def test_backmap(self, typename, typeint):
+ def test_backmap(self, typename: str, typeint: int):
assert ProxyTLV.name_to_num(typename) == typeint
def test_parse_partial(self):
@@ -384,14 +391,23 @@ class TestModule:
return emit
@parametrize("handshake", HANDSHAKES.values(), ids=HANDSHAKES.keys())
- def test_get(self, caplog, temp_event_loop, handshake):
+ def test_get(
+ self,
+ caplog: pytest.LogCaptureFixture,
+ temp_event_loop: asyncio.AbstractEventLoop,
+ handshake: bytes,
+ ):
caplog.set_level(logging.DEBUG)
mock_reader = self.MockAsyncReader(handshake)
reslt = temp_event_loop.run_until_complete(get_proxy(mock_reader))
assert isinstance(reslt, ProxyData)
assert reslt.valid
- def test_get_cut_v1(self, caplog, temp_event_loop):
+ def test_get_cut_v1(
+ self,
+ caplog: pytest.LogCaptureFixture,
+ temp_event_loop: asyncio.AbstractEventLoop,
+ ):
caplog.set_level(logging.DEBUG)
mock_reader = self.MockAsyncReader(GOOD_V1_HANDSHAKE[0:20])
reslt = temp_event_loop.run_until_complete(get_proxy(mock_reader))
@@ -401,7 +417,11 @@ class TestModule:
expect = ("mail.debug", 30, "PROXY error: PROXYv1 malformed")
assert expect in caplog.record_tuples
- def test_get_cut_v2(self, caplog, temp_event_loop):
+ def test_get_cut_v2(
+ self,
+ caplog: pytest.LogCaptureFixture,
+ temp_event_loop: asyncio.AbstractEventLoop,
+ ):
caplog.set_level(logging.DEBUG)
mock_reader = self.MockAsyncReader(TEST_V2_DATA1_EXACT[0:20])
reslt = temp_event_loop.run_until_complete(get_proxy(mock_reader))
@@ -412,7 +432,11 @@ class TestModule:
expect = ("mail.debug", 30, expect_msg)
assert expect in caplog.record_tuples
- def test_get_invalid_sig(self, caplog, temp_event_loop):
+ def test_get_invalid_sig(
+ self,
+ caplog: pytest.LogCaptureFixture,
+ temp_event_loop: asyncio.AbstractEventLoop,
+ ):
caplog.set_level(logging.DEBUG)
mock_reader = self.MockAsyncReader(b"PROXI TCP4 1.2.3.4 5.6.7.8 9 10\r\n")
reslt = temp_event_loop.run_until_complete(get_proxy(mock_reader))
@@ -451,7 +475,7 @@ class TestGetV1(_TestProxyProtocolCommon):
assert self.transport.close.called
@parametrize("patt", PUBLIC_V1_PATTERNS.values(), ids=PUBLIC_V1_PATTERNS.keys())
- def test_valid_patterns(self, setup_proxy_protocol, patt: bytes):
+ def test_valid_patterns(self, setup_proxy_protocol: Callable, patt: bytes):
if not patt.endswith(b"\r\n"):
patt += b"\r\n"
setup_proxy_protocol(self)
@@ -1004,7 +1028,7 @@ class TestWithController:
# Try resending the handshake. Should also fail (because connection has
# been closed by the server.
# noinspection PyTypeChecker
- with pytest.raises(OSError) as exc_info:
+ with pytest.raises(OSError) as exc_info: # noqa: PT011
sock.send(handshake)
resp = sock.recv(4096)
if resp == b"":
@@ -1041,7 +1065,7 @@ class TestWithController:
# Try resending the handshake. Should also fail (because connection has
# been closed by the server.
# noinspection PyTypeChecker
- with pytest.raises(OSError) as exc_info:
+ with pytest.raises(OSError) as exc_info: # noqa: PT011
sock.send(handshake)
resp = sock.recv(4096)
if resp == b"":
@@ -1094,8 +1118,7 @@ class TestHandlerAcceptReject:
sock.sendall(handshake)
resp = sock.recv(4096)
assert oper(resp, b"")
- with expect:
- with SMTPClient() as client:
- client.sock = sock
- code, mesg = client.ehlo("example.org")
- assert code == 250
+ with expect, SMTPClient() as client:
+ client.sock = sock
+ code, mesg = client.ehlo("example.org")
+ assert code == 250
diff --git a/aiosmtpd/tests/test_server.py b/aiosmtpd/tests/test_server.py
index 41225dc..5e27070 100644
--- a/aiosmtpd/tests/test_server.py
+++ b/aiosmtpd/tests/test_server.py
@@ -10,6 +10,7 @@ import socket
import time
from contextlib import ExitStack
from functools import partial
+from threading import Event
from pathlib import Path
from smtplib import SMTP as SMTPClient, SMTPServerDisconnected
from tempfile import mkdtemp
@@ -40,7 +41,7 @@ class SlowStartController(Controller):
kwargs.setdefault("ready_timeout", 0.5)
super().__init__(*args, **kwargs)
- def _run(self, ready_event):
+ def _run(self, ready_event: Event):
time.sleep(self.ready_timeout * 1.5)
super()._run(ready_event)
@@ -88,7 +89,7 @@ def safe_socket_dir() -> Generator[Path, None, None]:
#
yield tmpdir
#
- plist = [p for p in tmpdir.rglob("*")]
+ plist = list(tmpdir.rglob("*"))
for p in reversed(plist):
if p.is_dir():
p.rmdir()
@@ -97,7 +98,7 @@ def safe_socket_dir() -> Generator[Path, None, None]:
tmpdir.rmdir()
-def assert_smtp_socket(controller: UnixSocketMixin):
+def assert_smtp_socket(controller: UnixSocketMixin) -> bool:
assert Path(controller.unix_socket).exists()
sockfile = controller.unix_socket
ssl_context = controller.ssl_context
@@ -134,6 +135,7 @@ def assert_smtp_socket(controller: UnixSocketMixin):
catchup_delay()
resp = sock.recv(1024)
assert resp.startswith(b"221")
+ return True
class TestServer:
@@ -207,8 +209,9 @@ class TestController:
contr2 = Controller(
Sink(), hostname=Global.SrvAddr.host, port=Global.SrvAddr.port
)
+ expectedre = r"error while attempting to bind on address"
try:
- with pytest.raises(socket.error):
+ with pytest.raises(socket.error, match=expectedre):
contr2.start()
finally:
contr2.stop()
@@ -526,6 +529,7 @@ class TestUnthreaded:
assert temp_event_loop.is_closed() is False
+@pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
class TestFactory:
def test_normal_situation(self):
cont = Controller(Sink())
@@ -537,8 +541,7 @@ class TestFactory:
finally:
cont.stop()
- @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
- def test_unknown_args_direct(self, silence_event_loop_closed):
+ def test_unknown_args_direct(self, silence_event_loop_closed: bool):
unknown = "this_is_an_unknown_kwarg"
cont = Controller(Sink(), ready_timeout=0.3, **{unknown: True})
expectedre = r"__init__.. got an unexpected keyword argument '" + unknown + r"'"
@@ -553,8 +556,7 @@ class TestFactory:
@pytest.mark.filterwarnings(
"ignore:server_kwargs will be removed:DeprecationWarning"
)
- @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
- def test_unknown_args_inkwargs(self, silence_event_loop_closed):
+ def test_unknown_args_inkwargs(self, silence_event_loop_closed: bool):
unknown = "this_is_an_unknown_kwarg"
cont = Controller(Sink(), ready_timeout=0.3, server_kwargs={unknown: True})
expectedre = r"__init__.. got an unexpected keyword argument '" + unknown + r"'"
@@ -565,8 +567,7 @@ class TestFactory:
finally:
cont.stop()
- @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
- def test_factory_none(self, mocker: MockFixture, silence_event_loop_closed):
+ def test_factory_none(self, mocker: MockFixture, silence_event_loop_closed: bool):
# Hypothetical situation where factory() did not raise an Exception
# but returned None instead
mocker.patch("aiosmtpd.controller.SMTP", return_value=None)
@@ -579,8 +580,9 @@ class TestFactory:
finally:
cont.stop()
- @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
- def test_noexc_smtpd_missing(self, mocker, silence_event_loop_closed):
+ def test_noexc_smtpd_missing(
+ self, mocker: MockFixture, silence_event_loop_closed: bool
+ ):
# Hypothetical situation where factory() failed but no
# Exception was generated.
cont = Controller(Sink())
diff --git a/aiosmtpd/tests/test_smtp.py b/aiosmtpd/tests/test_smtp.py
index 6fd8bfb..0fb3a15 100644
--- a/aiosmtpd/tests/test_smtp.py
+++ b/aiosmtpd/tests/test_smtp.py
@@ -9,6 +9,7 @@ import logging
import socket
import time
import warnings
+from asyncio.transports import Transport
from base64 import b64encode
from contextlib import suppress
from smtplib import (
@@ -19,7 +20,7 @@ from smtplib import (
SMTPServerDisconnected,
)
from textwrap import dedent
-from typing import Any, AnyStr, Callable, Generator, List, Tuple
+from typing import cast, Any, AnyStr, Callable, Generator, List, Tuple
import pytest
from pytest_mock import MockFixture
@@ -62,10 +63,7 @@ MAIL_LOG.setLevel(logging.DEBUG)
def auth_callback(mechanism, login, password) -> bool:
- if login and login.decode() == "goodlogin":
- return True
- else:
- return False
+ return login and login.decode() == "goodlogin"
def assert_nopassleak(passwd: str, record_tuples: List[Tuple[str, int, str]]):
@@ -87,7 +85,7 @@ class UndescribableError(Exception):
class ErrorSMTP(Server):
exception_type = ValueError
- async def smtp_HELO(self, hostname):
+ async def smtp_HELO(self, hostname: str):
raise self.exception_type("test")
@@ -136,8 +134,13 @@ class PeekerHandler:
return AuthResult(success=True, auth_data=login_data)
async def handle_MAIL(
- self, server, session: SMTPSession, envelope, address, mail_options
- ):
+ self,
+ server: Server,
+ session: SMTPSession,
+ envelope: SMTPEnvelope,
+ address: str,
+ mail_options: dict,
+ ) -> str:
self.sess = session
return S.S250_OK.to_str()
@@ -157,7 +160,7 @@ class PeekerHandler:
async def auth_DONT(self, server, args):
return MISSING
- async def auth_WITH_UNDERSCORE(self, server: Server, args):
+ async def auth_WITH_UNDERSCORE(self, server: Server, args) -> str:
"""
Be careful when using this AUTH mechanism; log_client_response is set to
True, and this will raise some severe warnings.
@@ -180,7 +183,9 @@ class StoreEnvelopeOnVRFYHandler:
envelope = None
- async def handle_VRFY(self, server, session, envelope, addr):
+ async def handle_VRFY(
+ self, server: Server, session: SMTPSession, envelope: SMTPEnvelope, addr: str
+ ) -> str:
self.envelope = envelope
return S.S250_OK.to_str()
@@ -189,10 +194,10 @@ class ErroringHandler:
error = None
custom_response = False
- async def handle_DATA(self, server, session, envelope):
+ async def handle_DATA(self, server, session, envelope) -> str:
return "499 Could not accept the message"
- async def handle_exception(self, error):
+ async def handle_exception(self, error) -> str:
self.error = error
if not self.custom_response:
return "500 ErroringHandler handling error"
@@ -215,7 +220,7 @@ class ErroringHandlerConnectionLost:
class ErroringErrorHandler:
error = None
- async def handle_exception(self, error):
+ async def handle_exception(self, error: Exception):
self.error = error
raise ValueError("ErroringErrorHandler test")
@@ -223,13 +228,19 @@ class ErroringErrorHandler:
class UndescribableErrorHandler:
error = None
- async def handle_exception(self, error):
+ async def handle_exception(self, error: Exception):
self.error = error
raise UndescribableError()
class SleepingHeloHandler:
- async def handle_HELO(self, server, session, envelope, hostname):
+ async def handle_HELO(
+ self,
+ server: Server,
+ session: SMTPSession,
+ envelope: SMTPEnvelope,
+ hostname: str,
+ ) -> str:
await asyncio.sleep(0.01)
session.host_name = hostname
return "250 {}".format(server.hostname)
@@ -267,8 +278,7 @@ class CustomIdentController(Controller):
ident: bytes = b"Identifying SMTP v2112"
def factory(self):
- server = Server(self.handler, ident=self.ident.decode())
- return server
+ return Server(self.handler, ident=self.ident.decode())
# endregion
@@ -278,18 +288,19 @@ class CustomIdentController(Controller):
@pytest.fixture
-def transport_resp(mocker: MockFixture):
+def transport_resp(mocker: MockFixture) -> Tuple[Transport, list]:
responses = []
mocked = mocker.Mock()
mocked.write = responses.append
#
- yield mocked, responses
+ return cast(Transport, mocked), responses
@pytest.fixture
def get_protocol(
- temp_event_loop, transport_resp
-) -> Generator[Callable[..., Server], None, None]:
+ temp_event_loop: asyncio.AbstractEventLoop,
+ transport_resp: Any,
+) -> Callable[..., Server]:
transport, _ = transport_resp
def getter(*args, **kwargs) -> Server:
@@ -297,14 +308,16 @@ def get_protocol(
proto.connection_made(transport)
return proto
- yield getter
+ return getter
# region #### Fixtures: Controllers ##################################################
@pytest.fixture
-def auth_peeker_controller(get_controller) -> Generator[Controller, None, None]:
+def auth_peeker_controller(
+ get_controller: Callable[..., Controller]
+) -> Generator[Controller, None, None]:
handler = PeekerHandler()
controller = get_controller(
handler,
@@ -324,7 +337,7 @@ def auth_peeker_controller(get_controller) -> Generator[Controller, None, None]:
@pytest.fixture
def authenticator_peeker_controller(
- get_controller,
+ get_controller: Callable[..., Controller]
) -> Generator[Controller, None, None]:
handler = PeekerHandler()
controller = get_controller(
@@ -345,7 +358,8 @@ def authenticator_peeker_controller(
@pytest.fixture
def decoding_authnotls_controller(
- get_handler, get_controller
+ get_handler: Callable,
+ get_controller: Callable[..., Controller]
) -> Generator[Controller, None, None]:
handler = get_handler()
controller = get_controller(
@@ -368,7 +382,7 @@ def decoding_authnotls_controller(
@pytest.fixture
-def error_controller(get_handler) -> Generator[ErrorController, None, None]:
+def error_controller(get_handler: Callable) -> Generator[ErrorController, None, None]:
handler = get_handler()
controller = ErrorController(handler)
controller.start()
@@ -417,10 +431,8 @@ class TestProtocol:
]
)
)
- try:
+ with suppress(asyncio.CancelledError):
temp_event_loop.run_until_complete(protocol._handler_coroutine)
- except asyncio.CancelledError:
- pass
_, responses = transport_resp
assert responses[5] == S.S250_OK.to_bytes() + b"\r\n"
assert len(handler.box) == 1
@@ -441,10 +453,8 @@ class TestProtocol:
]
)
)
- try:
+ with suppress(asyncio.CancelledError):
temp_event_loop.run_until_complete(protocol._handler_coroutine)
- except asyncio.CancelledError:
- pass
_, responses = transport_resp
assert responses[5] == S.S250_OK.to_bytes() + b"\r\n"
assert len(handler.box) == 1
@@ -986,19 +996,19 @@ class TestAuthMechanisms(_CommonMethods):
@pytest.fixture
def do_auth_plain1(
self, client
- ) -> Generator[Callable[[str], Tuple[int, bytes]], None, None]:
+ ) -> Callable[[str], Tuple[int, bytes]]:
self._ehlo(client)
def do(param: str) -> Tuple[int, bytes]:
return client.docmd("AUTH PLAIN " + param)
do.client = client
- yield do
+ return do
@pytest.fixture
def do_auth_login3(
self, client
- ) -> Generator[Callable[[str], Tuple[int, bytes]], None, None]:
+ ) -> Callable[[str], Tuple[int, bytes]]:
self._ehlo(client)
resp = client.docmd("AUTH LOGIN")
assert resp == S.S334_AUTH_USERNAME
@@ -1007,7 +1017,7 @@ class TestAuthMechanisms(_CommonMethods):
return client.docmd(param)
do.client = client
- yield do
+ return do
def test_ehlo(self, client):
code, mesg = client.ehlo("example.com")
@@ -1119,11 +1129,11 @@ class TestAuthMechanisms(_CommonMethods):
assert_nopassleak(PW, caplog.record_tuples)
@pytest.fixture
- def client_auth_plain2(self, client) -> Generator[SMTPClient, None, None]:
+ def client_auth_plain2(self, client) -> SMTPClient:
self._ehlo(client)
resp = client.docmd("AUTH PLAIN")
assert resp == S.S334_AUTH_EMPTYPROMPT
- yield client
+ return client
def test_plain2_good_credentials(
self, caplog, auth_peeker_controller, client_auth_plain2
@@ -1965,7 +1975,8 @@ class TestAuthArgs:
],
)
def test_authmechname_decorator_badname(self, name):
- with pytest.raises(ValueError):
+ expectre = r"Invalid AUTH mechanism name"
+ with pytest.raises(ValueError, match=expectre):
auth_mechanism(name)
diff --git a/aiosmtpd/tests/test_starttls.py b/aiosmtpd/tests/test_starttls.py
index 6bb2cbd..5e0a180 100644
--- a/aiosmtpd/tests/test_starttls.py
+++ b/aiosmtpd/tests/test_starttls.py
@@ -12,6 +12,7 @@ import pytest
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Sink
from aiosmtpd.smtp import SMTP as Server
+from aiosmtpd.smtp import Envelope
from aiosmtpd.smtp import Session as Sess_
from aiosmtpd.smtp import TLSSetupException
from aiosmtpd.testing.helpers import ReceivingHandler, catchup_delay
@@ -31,14 +32,18 @@ class EOFingHandler:
ssl_existed = None
result = None
- async def handle_NOOP(self, server: Server, session: Sess_, envelope, arg):
+ async def handle_NOOP(
+ self, server: Server, session: Sess_, envelope: Envelope, arg: str
+ ) -> str:
self.ssl_existed = session.ssl is not None
self.result = server.eof_received()
return "250 OK"
class HandshakeFailingHandler:
- def handle_STARTTLS(self, server, session, envelope):
+ def handle_STARTTLS(
+ self, server: Server, session: Sess_, envelope: Envelope
+ ) -> bool:
return False
@@ -198,7 +203,7 @@ class TestStartTLS:
class ExceptionCaptureHandler:
error = None
- async def handle_exception(self, error):
+ async def handle_exception(self, error: Exception) -> str:
self.error = error
return "500 ExceptionCaptureHandler handling error"
@@ -354,7 +359,7 @@ class TestRequireTLSAUTH:
class TestTLSContext:
def test_verify_mode_nochange(self, ssl_context_server):
context = ssl_context_server
- for mode in (ssl.CERT_NONE, ssl.CERT_OPTIONAL):
+ for mode in (ssl.CERT_NONE, ssl.CERT_OPTIONAL): # noqa: DUO122
context.verify_mode = mode
_ = Server(Sink(), tls_context=context)
assert context.verify_mode == mode
@@ -370,10 +375,10 @@ class TestTLSContext:
def test_nocertreq_chkhost_warn(self, caplog, ssl_context_server):
context = ssl_context_server
- context.verify_mode = ssl.CERT_OPTIONAL
+ context.verify_mode = ssl.CERT_OPTIONAL # noqa: DUO122
context.check_hostname = True
_ = Server(Sink(), tls_context=context)
- assert context.verify_mode == ssl.CERT_OPTIONAL
+ assert context.verify_mode == ssl.CERT_OPTIONAL # noqa: DUO122
logmsg = caplog.record_tuples[0][-1]
assert "tls_context.check_hostname == True" in logmsg
assert "might cause client connection problems" in logmsg
diff --git a/housekeep.py b/housekeep.py
index 88dddd5..92b8cb6 100644
--- a/housekeep.py
+++ b/housekeep.py
@@ -69,7 +69,8 @@ TERM_WIDTH, TERM_HEIGHT = shutil.get_terminal_size()
def deldir(targ: Path, verbose: bool = True):
if not targ.exists():
return
- for i, pp in enumerate(reversed(sorted(targ.rglob("*"))), start=1):
+ rev_items = sorted(targ.rglob("*"), reverse=True)
+ for i, pp in enumerate(rev_items, start=1):
if pp.is_symlink():
pp.unlink()
elif pp.is_file():
diff --git a/setup.cfg b/setup.cfg
index 7cfbf7f..6638b75 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -66,4 +66,54 @@ source-dir = aiosmtpd/docs
[flake8]
jobs = 1
max-line-length = 88
-ignore = E123, E133, W503, W504, W293, E203
+# "E,F,W,C90" are flake8 defaults
+# For others, take a gander at tox.ini to see which prefix provided by who
+select = E,F,W,C90,C4,MOD,JS,PIE,PT,SIM,ECE,C801,DUO,TAE,ANN,YTT,N400
+ignore =
+ # black conflicts with E123 & E133
+ E123
+ E133
+ # W503 conflicts with PEP8...
+ W503
+ # W293 is a bit too noisy. Many files have been edited using editors that do not remove spaces from blank lines.
+ W293
+ # Sometimes spaces around colons improve readability
+ E203
+ # Sometimes we prefer the func()-based creation, not literal, for readability
+ C408
+ # Sometimes we need to catch Exception broadly
+ PIE786
+ # We don't really care about pytest.fixture vs pytest.fixture()
+ PT001
+ # Good idea, but too many changes. Remove this in the future, and create separate PR
+ PT004
+ # Sometimes exception needs to be explicitly raised in special circumstances, needing additional lines of code
+ PT012
+ # I still can't grok the need to annotate "self" or "cls" ...
+ ANN101
+ ANN102
+ # I don't think forcing annotation for *args and **kwargs is a wise idea...
+ ANN002
+ ANN003
+ # We have too many "if..elif..else: raise" structures that does not convert well to "error-first" design
+ SIM106
+per-file-ignores =
+ aiosmtpd/tests/test_*:ANN001
+ aiosmtpd/tests/test_proxyprotocol.py:DUO102
+ aiosmtpd/docs/_exts/autoprogramm.py:C801
+# flake8-coding
+no-accept-encodings = True
+# flake8-copyright
+copyright-check = True
+# The number below was determined empirically by bisecting from 100 until no copyright-unnecessary files appear
+copyright-min-file-size = 44
+copyright-author = The aiosmtpd Developers
+# flake8-annotations-complexity
+max-annotations-complexity = 4
+# flake8-annotations-coverage
+min-coverage-percents = 12
+# flake8-annotations
+mypy-init-return = True
+suppress-none-returning = True
+suppress-dummy-args = True
+allow-untyped-defs = True
diff --git a/tox.ini b/tox.ini
index 17d246e..eb2f4f6 100644
--- a/tox.ini
+++ b/tox.ini
@@ -10,11 +10,14 @@ envdir =
py37: {toxworkdir}/3.7
py38: {toxworkdir}/3.8
py39: {toxworkdir}/3.9
+ py310: {toxworkdir}/3.10
pypy3: {toxworkdir}/pypy3
py: {toxworkdir}/py
commands =
python housekeep.py prep
- !diffcov: bandit -c bandit.yml -r aiosmtpd
+ # Bandit is not needed on diffcov, and seems to be incompatible with 310
+ # So, run only if "not (310 or diffcov)" ==> "(not 310) and (not diffcov)"
+ !py310-!diffcov: bandit -c bandit.yml -r aiosmtpd
nocov: pytest --verbose -p no:cov --tb=short {posargs}
cov: pytest --cov --cov-report=xml --cov-report=html --cov-report=term --tb=short {posargs}
diffcov: diff-cover _dump/coverage-{env:INTERP}.xml --html-report htmlcov/diffcov-{env:INTERP}.html
@@ -24,6 +27,7 @@ commands =
#sitepackages = True
usedevelop = True
deps =
+ # do NOT make these conditional, that way we can reuse same envdir for nocov+cov+diffcov
bandit
colorama
coverage[toml]
@@ -43,6 +47,7 @@ setenv =
py37: INTERP=py37
py38: INTERP=py38
py39: INTERP=py39
+ py310: INTERP=py310
pypy3: INTERP=pypy3
py: INTERP=py
passenv =
@@ -51,18 +56,62 @@ passenv =
CI
GITHUB*
+[flake8_plugins]
+# This is a pseudo-section that feeds into [testenv:qa] and GA
+# Snippets of letters above these plugins are tests that need to be "select"-ed in flake8 config (in
+# setup.cfg) to activate the respective plugins. If no snippet is given, that means the plugin is
+# always active.
+deps =
+ flake8-bugbear
+ flake8-builtins
+ flake8-coding
+ # C4
+ flake8-comprehensions
+ # JS
+ flake8-multiline-containers
+ # PIE
+ flake8-pie
+ # MOD
+ flake8-printf-formatting
+ # PT
+ flake8-pytest-style
+ # SIM
+ flake8-simplify
+ # Cognitive Complexity looks like a good idea, but to fix the complaints... it will be an epic effort.
+ # So we disable it for now and reenable when we're ready, probably just before 2.0
+ # # CCR
+ # flake8-cognitive-complexity
+ # ECE
+ flake8-expression-complexity
+ # C801
+ flake8-copyright
+ # DUO
+ dlint
+ # TAE
+ flake8-annotations-complexity
+ # TAE
+ flake8-annotations-coverage
+ # ANN
+ flake8-annotations
+ # YTT
+ flake8-2020
+ # N400
+ flake8-broken-line
+
[testenv:qa]
basepython = python3
envdir = {toxworkdir}/qa
commands =
python housekeep.py prep
+ # The next line lists enabled plugins
+ python -m flake8 --version
python -m flake8 aiosmtpd setup.py housekeep.py release.py
check-manifest -v
pytest -v --tb=short aiosmtpd/qa
deps =
colorama
flake8
- flake8-bugbear
+ {[flake8_plugins]deps}
pytest
check-manifest
@@ -79,11 +128,13 @@ deps:
# - .github/workflows/unit-testing-and-coverage.yml
# - aiosmtpd/docs/RTD-requirements.txt
colorama
- pytest
sphinx
sphinx-autofixture
sphinx_rtd_theme
pickle5 ; python_version < '3.8'
+ # The below used as deps, need to be installed so autofixture work properly
+ pytest
+ pytest-mock
[testenv:static]
basepython = python3
--
2.32.0