mirror of
https://github.com/python/cpython.git
synced 2025-07-19 01:05:26 +00:00
bpo-32154: Remove asyncio.windows_utils.socketpair (#4609)
This commit is contained in:
parent
4d193bcc25
commit
ac577d7d0b
9 changed files with 21 additions and 123 deletions
|
@ -19,56 +19,6 @@ except ImportError:
|
|||
from asyncio import test_support as support
|
||||
|
||||
|
||||
class WinsocketpairTests(unittest.TestCase):
|
||||
|
||||
def check_winsocketpair(self, ssock, csock):
|
||||
csock.send(b'xxx')
|
||||
self.assertEqual(b'xxx', ssock.recv(1024))
|
||||
csock.close()
|
||||
ssock.close()
|
||||
|
||||
def test_winsocketpair(self):
|
||||
ssock, csock = windows_utils.socketpair()
|
||||
self.check_winsocketpair(ssock, csock)
|
||||
|
||||
@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
|
||||
def test_winsocketpair_ipv6(self):
|
||||
ssock, csock = windows_utils.socketpair(family=socket.AF_INET6)
|
||||
self.check_winsocketpair(ssock, csock)
|
||||
|
||||
@unittest.skipIf(hasattr(socket, 'socketpair'),
|
||||
'socket.socketpair is available')
|
||||
@mock.patch('asyncio.windows_utils.socket')
|
||||
def test_winsocketpair_exc(self, m_socket):
|
||||
m_socket.AF_INET = socket.AF_INET
|
||||
m_socket.SOCK_STREAM = socket.SOCK_STREAM
|
||||
m_socket.socket.return_value.getsockname.return_value = ('', 12345)
|
||||
m_socket.socket.return_value.accept.return_value = object(), object()
|
||||
m_socket.socket.return_value.connect.side_effect = OSError()
|
||||
|
||||
self.assertRaises(OSError, windows_utils.socketpair)
|
||||
|
||||
def test_winsocketpair_invalid_args(self):
|
||||
self.assertRaises(ValueError,
|
||||
windows_utils.socketpair, family=socket.AF_UNSPEC)
|
||||
self.assertRaises(ValueError,
|
||||
windows_utils.socketpair, type=socket.SOCK_DGRAM)
|
||||
self.assertRaises(ValueError,
|
||||
windows_utils.socketpair, proto=1)
|
||||
|
||||
@unittest.skipIf(hasattr(socket, 'socketpair'),
|
||||
'socket.socketpair is available')
|
||||
@mock.patch('asyncio.windows_utils.socket')
|
||||
def test_winsocketpair_close(self, m_socket):
|
||||
m_socket.AF_INET = socket.AF_INET
|
||||
m_socket.SOCK_STREAM = socket.SOCK_STREAM
|
||||
sock = mock.Mock()
|
||||
m_socket.socket.return_value = sock
|
||||
sock.bind.side_effect = OSError
|
||||
self.assertRaises(OSError, windows_utils.socketpair)
|
||||
self.assertTrue(sock.close.called)
|
||||
|
||||
|
||||
class PipeTests(unittest.TestCase):
|
||||
|
||||
def test_pipe_overlapped(self):
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue