8000 bpo-32418: Postfix, raise NotImplementdError and close resources in tests by asvetlov · Pull Request #5052 · python/cpython · GitHub
[go: up one dir, main page]
Skip to content

bpo-32418: Postfix, raise NotImplementdError and close resources in tests #5052

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Dec 30, 2017
Merged

bpo-32418: Postfix, raise NotImplementdError and close resources in tests #5052

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions Lib/asyncio/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
21 changes: 19 additions & 2 deletions Lib/test/test_asyncio/test_events.py
8000
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
0