summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_asyncio/test_protocols.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_asyncio/test_protocols.py')
-rw-r--r--Lib/test/test_asyncio/test_protocols.py57
1 files changed, 57 insertions, 0 deletions
diff --git a/Lib/test/test_asyncio/test_protocols.py b/Lib/test/test_asyncio/test_protocols.py
new file mode 100644
index 0000000..438111c
--- /dev/null
+++ b/Lib/test/test_asyncio/test_protocols.py
@@ -0,0 +1,57 @@
+import unittest
+from unittest import mock
+
+import asyncio
+
+
+class ProtocolsAbsTests(unittest.TestCase):
+
+ def test_base_protocol(self):
+ f = mock.Mock()
+ p = asyncio.BaseProtocol()
+ self.assertIsNone(p.connection_made(f))
+ self.assertIsNone(p.connection_lost(f))
+ self.assertIsNone(p.pause_writing())
+ self.assertIsNone(p.resume_writing())
+ self.assertFalse(hasattr(p, '__dict__'))
+
+ def test_protocol(self):
+ f = mock.Mock()
+ p = asyncio.Protocol()
+ self.assertIsNone(p.connection_made(f))
+ self.assertIsNone(p.connection_lost(f))
+ self.assertIsNone(p.data_received(f))
+ self.assertIsNone(p.eof_received())
+ self.assertIsNone(p.pause_writing())
+ self.assertIsNone(p.resume_writing())
+ self.assertFalse(hasattr(p, '__dict__'))
+
+ def test_buffered_protocol(self):
+ f = mock.Mock()
+ p = asyncio.BufferedProtocol()
+ self.assertIsNone(p.connection_made(f))
+ self.assertIsNone(p.connection_lost(f))
+ self.assertIsNone(p.get_buffer(100))
+ self.assertIsNone(p.buffer_updated(150))
+ self.assertIsNone(p.pause_writing())
+ self.assertIsNone(p.resume_writing())
+ self.assertFalse(hasattr(p, '__dict__'))
+
+ def test_datagram_protocol(self):
+ f = mock.Mock()
+ dp = asyncio.DatagramProtocol()
+ self.assertIsNone(dp.connection_made(f))
+ self.assertIsNone(dp.connection_lost(f))
+ self.assertIsNone(dp.error_received(f))
+ self.assertIsNone(dp.datagram_received(f, f))
+ self.assertFalse(hasattr(dp, '__dict__'))
+
+ def test_subprocess_protocol(self):
+ f = mock.Mock()
+ sp = asyncio.SubprocessProtocol()
+ self.assertIsNone(sp.connection_made(f))
+ self.assertIsNone(sp.connection_lost(f))
+ self.assertIsNone(sp.pipe_data_received(1, f))
+ self.assertIsNone(sp.pipe_connection_lost(1, f))
+ self.assertIsNone(sp.process_exited())
+ self.assertFalse(hasattr(sp, '__dict__'))