diff --git a/README.md b/README.md index 23a42557..40460cae 100644 --- a/README.md +++ b/README.md @@ -568,13 +568,17 @@ handshake() method, doing some sort of server handshake on the connection argument. If the handshake method returns True, the RequestHandler will be triggered. See the tests/httpsserver.py example. -10 Using tlslite-ng with asyncore +10 Using tlslite-ng with asyncore (or asyncio - Python 3.12+) ================================= tlslite-ng can be used with subclasses of asyncore.dispatcher. See the comments in TLSAsyncDispatcherMixIn.py for details. This is still experimental, and may not work with all asyncore.dispatcher subclasses. +as said above, asyncore is deprecated in Python 3.12, and asyncio should be used. +Implementation is similar to TLSAsyncDispatcherMixIn.py, but instead, use the class +TLSAsyncioDispatcherMixIn.py. + 11 History =========== diff --git a/docs/tlslite.integration.tlsasynciodispatchermixin b/docs/tlslite.integration.tlsasynciodispatchermixin new file mode 100644 index 00000000..5c944f1d --- /dev/null +++ b/docs/tlslite.integration.tlsasynciodispatchermixin @@ -0,0 +1,8 @@ +tlslite.integration.tlsasynciodispatchermixin module +================================================== + +.. automodule:: tlslite.integration.tlsasynciodispatchermixin + :members: + :special-members: __init__ + :undoc-members: + :show-inheritance: diff --git a/unit_tests/test_tlslite_integration_tlsasynciodispatchermixin.py b/unit_tests/test_tlslite_integration_tlsasynciodispatchermixin.py new file mode 100644 index 00000000..af053c42 --- /dev/null +++ b/unit_tests/test_tlslite_integration_tlsasynciodispatchermixin.py @@ -0,0 +1,74 @@ +import asyncio +import unittest +from unittest.mock import Mock +from tlslite.integration.tlsasynciodispatchermixin import TLSAsyncioDispatcherMixIn + + +class MockProtocol(asyncio.Protocol): + + def connection_lost(self, exc): + self.in_write_event() + + def in_write_event(self): + pass + + def readable(self): + return True + + def writable(self): + return True + + +class TestTLSAsyncioDispatcherMixIn(unittest.TestCase): + def setUp(self): + self.protocol = TLSAsyncioDispatcherMixIn() + self.protocol.__class__ = type('TestProtocol', (MockProtocol, TLSAsyncioDispatcherMixIn), {}) + self.protocol.transport = Mock() + self.protocol.tls_connection = Mock() + + def test_readable(self): + self.protocol.wants_read_event = Mock(return_value=None) + self.protocol._get_sibling_class = Mock(return_value=Mock(readable=Mock(return_value=True))) + self.assertTrue(self.protocol.readable()) + + def test_writable(self): + self.protocol.wants_write_event = Mock(return_value=None) + self.protocol._get_sibling_class = Mock(return_value=Mock(writable=Mock(return_value=True))) + self.assertTrue(self.protocol.writable()) + + def test_data_received(self): + self.protocol.transport = Mock() + self.protocol.sibling_class.data_received = Mock() + self.protocol.data_received(b'test') + self.protocol.sibling_class.data_received.assert_called_once_with(self.protocol, b'test') + + def test_connection_lost(self): + self.protocol.in_write_event = Mock() + self.protocol.connection_lost(None) + self.protocol.in_write_event.assert_called_once() + + def test_connection_made(self): + self.protocol.transport = Mock() + self.protocol.sibling_class.connection_made = Mock() + self.protocol.connection_made(self.protocol.transport) + self.protocol.sibling_class.connection_made.assert_called_once_with(self.protocol.transport) + + def test_out_close_event(self): + self.protocol.out_close_event() + self.protocol.transport.close.assert_called_once() + + def test_recv(self): + self.protocol.read_buffer = b"test" + self.assertEqual(self.protocol.recv(), b"test") + self.assertIsNone(self.protocol.read_buffer) + + def test_send(self): + write_buffer = b"test" + self.protocol.set_write_op = Mock() + self.assertEqual(self.protocol.send(write_buffer), len(write_buffer)) + self.protocol.set_write_op.assert_called_once_with(write_buffer) + + def test_close(self): + self.protocol.set_close_op = Mock() + self.protocol.close() + self.protocol.set_close_op.assert_called_once() \ No newline at end of file