https://github.com/python/cpython/commit/ffcb4c0165827d0a48ea973cc88bc134c74... commit: ffcb4c0165827d0a48ea973cc88bc134c74879fb branch: master author: Andrew Svetlov <andrew.svetlov@gmail.com> committer: GitHub <noreply@github.com> date: 2017-12-30T18:52:56+02:00 summary: bpo-32418: Postfix, raise NotImplementdError and close resources in tests (#5052) files: M Lib/asyncio/events.py M Lib/test/test_asyncio/test_events.py diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py index 731a0c5e802..af4545b2cbe 100644 --- a/Lib/asyncio/events.py +++ b/Lib/asyncio/events.py @@ -149,15 +149,15 @@ class AbstractServer: def close(self): """Stop serving. This leaves existing connections open.""" - return NotImplemented + raise NotImplementedError async def wait_closed(self): """Coroutine to wait until service is closed.""" - return NotImplemented + raise NotImplementedError def get_loop(self): """ Get the event loop the Server object is attached to.""" - return NotImplemented + raise NotImplementedError class AbstractEventLoop: diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py index e8320564c74..f63fd3c723a 100644 --- a/Lib/test/test_asyncio/test_events.py +++ b/Lib/test/test_asyncio/test_events.py @@ -2826,19 +2826,36 @@ class TestCGetEventLoop(GetEventLoopTestsMixin, unittest.TestCase): get_running_loop_impl = events._c_get_running_loop get_event_loop_impl = events._c_get_event_loop + class TestServer(unittest.TestCase): def test_get_loop(self): loop = asyncio.new_event_loop() + self.addCleanup(loop.close) proto = MyProto(loop) server = loop.run_until_complete(loop.create_server(lambda: proto, '0.0.0.0', 0)) self.assertEqual(server.get_loop(), loop) - loop.close() + server.close() + loop.run_until_complete(server.wait_closed()) + class TestAbstractServer(unittest.TestCase): + def test_close(self): + with self.assertRaises(NotImplementedError): + events.AbstractServer().close() + + def test_wait_closed(self): + loop = asyncio.new_event_loop() + self.addCleanup(loop.close) + + with self.assertRaises(NotImplementedError): + loop.run_until_complete(events.AbstractServer().wait_closed()) + def test_get_loop(self): - self.assertEqual(events.AbstractServer().get_loop(), NotImplemented) + with self.assertRaises(NotImplementedError): + events.AbstractServer().get_loop() + if __name__ == '__main__': unittest.main()