diff options
| author | Kumar Aditya <59607654+kumaraditya303@users.noreply.github.com> | 2022-10-27 03:07:31 (GMT) |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2022-10-27 03:07:31 (GMT) |
| commit | 96ae80f1d004f2df4a707919853f0745c9c352d1 (patch) | |
| tree | cd74b029b99a2ad9a5c56b8210ff0903ba942a31 /Lib/test/test_asyncio | |
| parent | 8a755423eba8e19704d96905730cf7f50083eb23 (diff) | |
| download | cpython-96ae80f1d004f2df4a707919853f0745c9c352d1.zip cpython-96ae80f1d004f2df4a707919853f0745c9c352d1.tar.gz cpython-96ae80f1d004f2df4a707919853f0745c9c352d1.tar.bz2 | |
gh-98703: Add tests for closing `_ProactorSocketTransport` with proactor event loop (GH-98730)
Diffstat (limited to 'Lib/test/test_asyncio')
| -rw-r--r-- | Lib/test/test_asyncio/test_proactor_events.py | 21 |
1 files changed, 21 insertions, 0 deletions
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py index 7fd8b26..ae30185 100644 --- a/Lib/test/test_asyncio/test_proactor_events.py +++ b/Lib/test/test_asyncio/test_proactor_events.py @@ -297,6 +297,27 @@ class ProactorSocketTransportTests(test_utils.TestCase): # and waiters will never be notified leading to hang. self.assertTrue(self.protocol.connection_lost.called) + def test_force_close_protocol_connection_lost_once(self): + tr = self.socket_transport() + self.assertFalse(self.protocol.connection_lost.called) + tr._closing = True + # Calling _force_close twice should not call + # protocol.connection_lost twice + tr._force_close(None) + tr._force_close(None) + test_utils.run_briefly(self.loop) + self.assertEqual(1, self.protocol.connection_lost.call_count) + + def test_close_protocol_connection_lost_once(self): + tr = self.socket_transport() + self.assertFalse(self.protocol.connection_lost.called) + # Calling close twice should not call + # protocol.connection_lost twice + tr.close() + tr.close() + test_utils.run_briefly(self.loop) + self.assertEqual(1, self.protocol.connection_lost.call_count) + def test_fatal_error_2(self): tr = self.socket_transport() tr._buffer = [b'data'] |