Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

gh-113879: Fix ResourceWarning in test_asyncio.test_server (GH-113881)

This commit is contained in:
Serhiy Storchaka 2024-01-10 12:38:36 +02:00 committed by GitHub
parent 1b7e0024a1
commit ab0ad62038
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
Lib/test/test_asyncio/test_server.py
View File

@ -200,13 +200,13 @@ class UnixServerCleanupTests(unittest.IsolatedAsyncioTestCase):
async def serve(*args):
pass
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(addr)
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.bind(addr)
srv = await asyncio.start_unix_server(serve, sock=sock)
srv = await asyncio.start_unix_server(serve, sock=sock)
srv.close()
self.assertFalse(os.path.exists(addr))
srv.close()
self.assertFalse(os.path.exists(addr))
@socket_helper.skip_unless_bind_unix_socket
async def test_unix_server_cleanup_gone(self):
@ -215,14 +215,14 @@ class UnixServerCleanupTests(unittest.IsolatedAsyncioTestCase):
async def serve(*args):
pass
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(addr)
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.bind(addr)
srv = await asyncio.start_unix_server(serve, sock=sock)
srv = await asyncio.start_unix_server(serve, sock=sock)
os.unlink(addr)
os.unlink(addr)
srv.close()
srv.close()
@socket_helper.skip_unless_bind_unix_socket
async def test_unix_server_cleanup_replaced(self):
@ -234,11 +234,11 @@ class UnixServerCleanupTests(unittest.IsolatedAsyncioTestCase):
srv = await asyncio.start_unix_server(serve, addr)
os.unlink(addr)
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(addr)
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.bind(addr)
srv.close()
self.assertTrue(os.path.exists(addr))
srv.close()
self.assertTrue(os.path.exists(addr))
@socket_helper.skip_unless_bind_unix_socket
async def test_unix_server_cleanup_prevented(self):