New attempt at getting secure SMTP working

tests/smtp/aiosmtpd_fake.py

@@ -0,0 +1,133 @@
+import os
+import time
+import ssl
+
+from aiosmtpd.smtp import SMTP as Server, syntax
+from aiosmtpd.controller import Controller
+
+SMTP_PORT = int(os.environ.get(u'SMTP_PORT', u'8080'))
+SECURE_SMTP_PORT = int(os.environ.get(u'SECURE_SMTP_PORT', u'8081'))
+
+
+class MyController(Controller):
+    def factory(self):
+        return MyServer(self.handler)
+
+class MyServer(Server):
+    pass
+
+class MyHandler:
+    async def handle_HELO(self, server, session, envelope, hostname):
+        print(u'HELO entered')
+        return None
+
+    async def handle_EHLO(self, server, session, envelope, hostname):
+        print(u'EHLO entered')
+        return None
+
+    async def handle_NOOP(self, server, session, envelope, arg):
+        print(u'NOOP entered')
+        return None
+
+    async def handle_QUIT(self, server, session, envelope):
+        print(u'QUIT entered')
+        return None
+
+    async def handle_VRFY(self, server, session, envelope, address):
+        print(u'VRFY entered')
+        return None
+
+    async def handle_MAIL(self, server, session, envelope, address, mail_options):
+        print(u'MAIL entered')
+        return None
+
+    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
+        print(u'RCPT entered')
+        return None
+
+    async def handle_RSET(self, server, session, envelope):
+        print(u'RSET entered')
+        return None
+
+    async def handle_DATA(self, server, session, envelope):
+        print(u'DATA entered')
+        return None
+
+    async def handle_STARTTLS(self, server, session, envelope):
+        print(u'STARTTLS entered')
+        return True
+
+    async def handle_exception(self, error):
+        print(u'exception entered')
+        return u'542 Internal server error'
+
+_controller = None
+def smtp_server_start():
+    global _controller
+    if _controller is None:
+        _controller = MyController(MyHandler(), hostname=u'localhost', port=SMTP_PORT)
+        _controller.start()
+        time.sleep(1.0)
+
+class MySecureController(Controller):
+    def factory(self):
+        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        return MyServer(self.handler, tls_context=context, require_starttls=True)
+
+class MySecureServer(Server):
+    async def smtp_AUTH(self, *args, **kwargs):
+        print(u'AUTH entered: {}\n{}'.format(args, kwargs))
+        await self.push(u'250 OK')
+
+class MySecureHandler:
+    async def handle_HELO(self, server, session, envelope, hostname):
+        print(u'HELO entered')
+        return None
+
+    async def handle_EHLO(self, server, session, envelope, hostname):
+        print(u'EHLO entered')
+        return None
+
+    async def handle_NOOP(self, server, session, envelope, arg):
+        print(u'NOOP entered')
+        return None
+
+    async def handle_QUIT(self, server, session, envelope):
+        print(u'QUIT entered')
+        return None
+
+    async def handle_VRFY(self, server, session, envelope, address):
+        print(u'VRFY entered')
+        return None
+
+    async def handle_MAIL(self, server, session, envelope, address, mail_options):
+        print(u'MAIL entered')
+        return None
+
+    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
+        print(u'RCPT entered')
+        return None
+
+    async def handle_RSET(self, server, session, envelope):
+        print(u'RSET entered')
+        return None
+
+    async def handle_DATA(self, server, session, envelope):
+        print(u'DATA entered')
+        return None
+
+    async def handle_STARTTLS(self, server, session, envelope):
+        print(u'STARTTLS entered')
+        return True
+
+    async def handle_exception(self, error):
+        print(u'exception entered')
+        return u'542 Internal server error'
+
+_secure_controller = None
+def secure_smtp_server_start():
+    global _secure_controller
+    if _secure_controller is None:
+        _secure_controller = MySecureController(MySecureHandler(), hostname=u'localhost', port=u'')
+        _secure_controller.start()
+        time.sleep(1.0)

tests/smtpd_fake.py → tests/smtp/smtpd_fake.py

@@ -21,8 +21,12 @@ def _smtp_server_func():
 
 
 _smtp_server_thead = None
-def smtp_server_thread():
+def smtp_server_start():
     global _smtp_server_thead
     if _smtp_server_thead is None:
         _smtp_server_thead = _thread.start_new_thread(_smtp_server_func, ())
         time.sleep(1.0)
+
+
+def secure_smtp_server_start():
+    raise NotImplementedError(u'not available in this environment')

tests/test_quick_email.py

@@ -1,34 +1,36 @@
 import unittest
 
 from quick_email import send_email
-import smtpd_fake
 
 try:
-    import aiosmtpd
-    del aiosmtpd
+    import tests.smtp.aiosmtpd_fake as smtp
 
     _can_run_ssl_tests = True
     _can_run_auth_tests = True
-except ImportError:
+except (ImportError, SyntaxError):
+    import tests.smtp.smtpd_fake as smtp
     _can_run_ssl_tests = False
     _can_run_auth_tests = False
 
 class TestQuickEmail(unittest.TestCase):
     def test_send_email(self):
-        smtpd_fake.smtp_server_thread()
-        send_email(u'localhost', smtpd_fake.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', send_cc=None, send_bcc=None, plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', attachment_list=None, inline_attachment_dict=None)
+        smtp.smtp_server_start()
+        send_email(u'localhost', smtp.SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
 
     @unittest.skipUnless(_can_run_ssl_tests, 'SSL tests unsupported')
     def test_send_email_ssl(self):
-        pass
+        smtp.secure_smtp_server_start()
+        send_email(u'localhost', smtp.SECURE_SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', ssl=True)
 
     @unittest.skipUnless(_can_run_auth_tests, 'Auth tests unsupported')
     def test_send_email_auth(self):
-        pass
+        smtp.secure_smtp_server_start()
+        send_email(u'localhost', smtp.SECURE_SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', username=u'testuser', password=u'password')
 
     @unittest.skipUnless(_can_run_ssl_tests and _can_run_auth_tests, 'SSL or Auth tests unsupported')
     def test_send_email_ssl_auth(self):
-        pass
+        smtp.secure_smtp_server_start()
+        send_email(u'localhost', smtp.SECURE_SMTP_PORT, u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>', username=u'testuser', password=u'password', ssl=True)
 
 if __name__ == u'__main__':
     unittest.main()

tests/test_sender.py

@@ -1,19 +1,28 @@
 import unittest
 
 from quick_email import builder, sender
-import smtpd_fake
+
+try:
+    import tests.smtp.aiosmtpd_fake as smtp
+
+    _can_run_ssl_tests = True
+    _can_run_auth_tests = True
+except (ImportError, SyntaxError):
+    import tests.smtp.smtpd_fake as smtp
+    _can_run_ssl_tests = False
+    _can_run_auth_tests = False
 
 
 class TestSender(unittest.TestCase):
     def test_minimal(self):
-        smtpd_fake.smtp_server_thread()
+        smtp.smtp_server_start()
         msg = builder.build_msg(u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
-        sender.send_msg(msg, u'localhost', smtpd_fake.SMTP_PORT)
+        sender.send_msg(msg, u'localhost', smtp.SMTP_PORT)
 
     def test_multiple_recipients(self):
-        smtpd_fake.smtp_server_thread()
+        smtp.smtp_server_start()
         msg = builder.build_msg(u'Example <example@example.com>', u'The Subject', send_to=u'Test <test@test.com>', send_cc=[u'Example <example@example.com>', u'Example2 <example2@example.com>'], send_bcc=u'Example3 <example3@example.com>', plain_text=u'Some Text', html_text=u'<b>Some Bold Text</b>')
-        sender.send_msg(msg, u'localhost', smtpd_fake.SMTP_PORT)
+        sender.send_msg(msg, u'localhost', smtp.SMTP_PORT)
 
 
 if __name__ == u'__main__':