Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(164652)

Delta Between Two Patch Sets: Lib/test/test_socket.py

Issue 16510: Using appropriate checks in tests
Left Patch Set: Created 6 years ago
Right Patch Set: Created 5 years, 10 months ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
Left: Side by side diff | Download
Right: Side by side diff | Download
« no previous file with change/comment | « Lib/test/test_slice.py ('k') | Lib/test/test_ssl.py » ('j') | no next file with change/comment »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
LEFTRIGHT
1 #!/usr/bin/env python3
2
3 import unittest 1 import unittest
4 from test import support 2 from test import support
5 3
6 import errno 4 import errno
7 import io 5 import io
8 import socket 6 import socket
9 import select 7 import select
10 import tempfile 8 import tempfile
11 import _testcapi
12 import time 9 import time
13 import traceback 10 import traceback
14 import queue 11 import queue
15 import sys 12 import sys
16 import os 13 import os
17 import array 14 import array
18 import platform 15 import platform
19 import contextlib 16 import contextlib
20 from weakref import proxy 17 from weakref import proxy
21 import signal 18 import signal
(...skipping 716 matching lines...) Expand 10 before | Expand all | Expand 10 after
738 socket.SOL_SOCKET 735 socket.SOL_SOCKET
739 socket.SO_REUSEADDR 736 socket.SO_REUSEADDR
740 737
741 def testHostnameRes(self): 738 def testHostnameRes(self):
742 # Testing hostname resolution mechanisms 739 # Testing hostname resolution mechanisms
743 hostname = socket.gethostname() 740 hostname = socket.gethostname()
744 try: 741 try:
745 ip = socket.gethostbyname(hostname) 742 ip = socket.gethostbyname(hostname)
746 except OSError: 743 except OSError:
747 # Probably name lookup wasn't set up right; skip this test 744 # Probably name lookup wasn't set up right; skip this test
748 return 745 self.skipTest('name lookup failure')
749 self.assertGreaterEqual(ip.find('.'), 0, "Error resolving host to ip.") 746 self.assertIn('.', ip, "Error resolving host to ip.")
750 try: 747 try:
751 hname, aliases, ipaddrs = socket.gethostbyaddr(ip) 748 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
752 except OSError: 749 except OSError:
753 # Probably a similar problem as above; skip this test 750 # Probably a similar problem as above; skip this test
754 return 751 self.skipTest('name lookup failure')
755 all_host_names = [hostname, hname] + aliases 752 all_host_names = [hostname, hname] + aliases
756 fqhn = socket.getfqdn(ip) 753 fqhn = socket.getfqdn(ip)
757 if not fqhn in all_host_names: 754 if not fqhn in all_host_names:
758 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names))) 755 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names)))
759 756
760 def test_host_resolution(self): 757 def test_host_resolution(self):
761 for addr in ['0.1.1.~1', '1+.1.1.1', '::1q', '::1::2', 758 for addr in ['0.1.1.~1', '1+.1.1.1', '::1q', '::1::2',
762 '1:1:1:1:1:1:1:1:1']: 759 '1:1:1:1:1:1:1:1:1']:
763 self.assertRaises(OSError, socket.gethostbyname, addr) 760 self.assertRaises(OSError, socket.gethostbyname, addr)
764 self.assertRaises(OSError, socket.gethostbyaddr, addr) 761 self.assertRaises(OSError, socket.gethostbyaddr, addr)
(...skipping 205 matching lines...) Expand 10 before | Expand all | Expand 10 after
970 assertInvalid(g, 'a.0.0.0') 967 assertInvalid(g, 'a.0.0.0')
971 assertInvalid(g, '1.2.3.4.5') 968 assertInvalid(g, '1.2.3.4.5')
972 assertInvalid(g, '::1') 969 assertInvalid(g, '::1')
973 970
974 @unittest.skipUnless(hasattr(socket, 'inet_pton'), 971 @unittest.skipUnless(hasattr(socket, 'inet_pton'),
975 'test needs socket.inet_pton()') 972 'test needs socket.inet_pton()')
976 def testIPv6toString(self): 973 def testIPv6toString(self):
977 try: 974 try:
978 from socket import inet_pton, AF_INET6, has_ipv6 975 from socket import inet_pton, AF_INET6, has_ipv6
979 if not has_ipv6: 976 if not has_ipv6:
980 return 977 self.skipTest('IPv6 not available')
981 except ImportError: 978 except ImportError:
982 return 979 self.skipTest('could not import needed symbols from socket')
983 980
984 if sys.platform == "win32": 981 if sys.platform == "win32":
985 try: 982 try:
986 inet_pton(AF_INET6, '::') 983 inet_pton(AF_INET6, '::')
987 except OSError as e: 984 except OSError as e:
988 if e.winerror == 10022: 985 if e.winerror == 10022:
989 return # IPv6 might not be installed on this PC 986 self.skipTest('IPv6 might not be supported')
990 987
991 f = lambda a: inet_pton(AF_INET6, a) 988 f = lambda a: inet_pton(AF_INET6, a)
992 assertInvalid = lambda a: self.assertRaises( 989 assertInvalid = lambda a: self.assertRaises(
993 (OSError, ValueError), f, a 990 (OSError, ValueError), f, a
994 ) 991 )
995 992
996 self.assertEqual(b'\x00' * 16, f('::')) 993 self.assertEqual(b'\x00' * 16, f('::'))
997 self.assertEqual(b'\x00' * 16, f('0::0')) 994 self.assertEqual(b'\x00' * 16, f('0::0'))
998 self.assertEqual(b'\x00\x01' + b'\x00' * 14, f('1::')) 995 self.assertEqual(b'\x00\x01' + b'\x00' * 14, f('1::'))
999 self.assertEqual( 996 self.assertEqual(
(...skipping 56 matching lines...) Expand 10 before | Expand all | Expand 10 after
1056 assertInvalid(g, b'\x00' * 3) 1053 assertInvalid(g, b'\x00' * 3)
1057 assertInvalid(g, b'\x00' * 5) 1054 assertInvalid(g, b'\x00' * 5)
1058 assertInvalid(g, b'\x00' * 16) 1055 assertInvalid(g, b'\x00' * 16)
1059 1056
1060 @unittest.skipUnless(hasattr(socket, 'inet_ntop'), 1057 @unittest.skipUnless(hasattr(socket, 'inet_ntop'),
1061 'test needs socket.inet_ntop()') 1058 'test needs socket.inet_ntop()')
1062 def testStringToIPv6(self): 1059 def testStringToIPv6(self):
1063 try: 1060 try:
1064 from socket import inet_ntop, AF_INET6, has_ipv6 1061 from socket import inet_ntop, AF_INET6, has_ipv6
1065 if not has_ipv6: 1062 if not has_ipv6:
1066 return 1063 self.skipTest('IPv6 not available')
1067 except ImportError: 1064 except ImportError:
1068 return 1065 self.skipTest('could not import needed symbols from socket')
1069 1066
1070 if sys.platform == "win32": 1067 if sys.platform == "win32":
1071 try: 1068 try:
1072 inet_ntop(AF_INET6, b'\x00' * 16) 1069 inet_ntop(AF_INET6, b'\x00' * 16)
1073 except OSError as e: 1070 except OSError as e:
1074 if e.winerror == 10022: 1071 if e.winerror == 10022:
1075 return # IPv6 might not be installed on this PC 1072 self.skipTest('IPv6 might not be supported')
1076 1073
1077 f = lambda a: inet_ntop(AF_INET6, a) 1074 f = lambda a: inet_ntop(AF_INET6, a)
1078 assertInvalid = lambda a: self.assertRaises( 1075 assertInvalid = lambda a: self.assertRaises(
1079 (OSError, ValueError), f, a 1076 (OSError, ValueError), f, a
1080 ) 1077 )
1081 1078
1082 self.assertEqual('::', f(b'\x00' * 16)) 1079 self.assertEqual('::', f(b'\x00' * 16))
1083 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01')) 1080 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01'))
1084 self.assertEqual( 1081 self.assertEqual(
1085 'aef:b01:506:1001:ffff:9997:55:170', 1082 'aef:b01:506:1001:ffff:9997:55:170',
(...skipping 13 matching lines...) Expand all
1099 self.addCleanup(sock.close) 1096 self.addCleanup(sock.close)
1100 sock.bind(("0.0.0.0", port)) 1097 sock.bind(("0.0.0.0", port))
1101 name = sock.getsockname() 1098 name = sock.getsockname()
1102 # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate 1099 # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate
1103 # it reasonable to get the host's addr in addition to 0.0.0.0. 1100 # it reasonable to get the host's addr in addition to 0.0.0.0.
1104 # At least for eCos. This is required for the S/390 to pass. 1101 # At least for eCos. This is required for the S/390 to pass.
1105 try: 1102 try:
1106 my_ip_addr = socket.gethostbyname(socket.gethostname()) 1103 my_ip_addr = socket.gethostbyname(socket.gethostname())
1107 except OSError: 1104 except OSError:
1108 # Probably name lookup wasn't set up right; skip this test 1105 # Probably name lookup wasn't set up right; skip this test
1109 return 1106 self.skipTest('name lookup failure')
1110 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0]) 1107 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0])
1111 self.assertEqual(name[1], port) 1108 self.assertEqual(name[1], port)
1112 1109
1113 def testGetSockOpt(self): 1110 def testGetSockOpt(self):
1114 # Testing getsockopt() 1111 # Testing getsockopt()
1115 # We know a socket should start without reuse==0 1112 # We know a socket should start without reuse==0
1116 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1113 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1117 self.addCleanup(sock.close) 1114 self.addCleanup(sock.close)
1118 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1115 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1119 self.assertEqual(reuse, 0, "initial mode is reuse") 1116 self.assertEqual(reuse, 0, "initial mode is reuse")
(...skipping 219 matching lines...) Expand 10 before | Expand all | Expand 10 after
1339 for protocol in range(pickle.HIGHEST_PROTOCOL + 1): 1336 for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
1340 self.assertRaises(TypeError, pickle.dumps, sock, protocol) 1337 self.assertRaises(TypeError, pickle.dumps, sock, protocol)
1341 1338
1342 def test_listen_backlog(self): 1339 def test_listen_backlog(self):
1343 for backlog in 0, -1: 1340 for backlog in 0, -1:
1344 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1341 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1345 srv.bind((HOST, 0)) 1342 srv.bind((HOST, 0))
1346 srv.listen(backlog) 1343 srv.listen(backlog)
1347 srv.close() 1344 srv.close()
1348 1345
1346 @support.cpython_only
1347 def test_listen_backlog_overflow(self):
1349 # Issue 15989 1348 # Issue 15989
1349 import _testcapi
1350 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1350 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1351 srv.bind((HOST, 0)) 1351 srv.bind((HOST, 0))
1352 self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1) 1352 self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1)
1353 srv.close() 1353 srv.close()
1354 1354
1355 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.') 1355 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.')
1356 def test_flowinfo(self): 1356 def test_flowinfo(self):
1357 self.assertRaises(OverflowError, socket.getnameinfo, 1357 self.assertRaises(OverflowError, socket.getnameinfo,
1358 (support.HOSTv6, 0, 0xffffffff), 0) 1358 (support.HOSTv6, 0, 0xffffffff), 0)
1359 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: 1359 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
(...skipping 373 matching lines...) Expand 10 before | Expand all | Expand 10 after
1733 # Testing shutdown() 1733 # Testing shutdown()
1734 msg = self.cli_conn.recv(1024) 1734 msg = self.cli_conn.recv(1024)
1735 self.assertEqual(msg, MSG) 1735 self.assertEqual(msg, MSG)
1736 # wait for _testShutdown to finish: on OS X, when the server 1736 # wait for _testShutdown to finish: on OS X, when the server
1737 # closes the connection the client also becomes disconnected, 1737 # closes the connection the client also becomes disconnected,
1738 # and the client's shutdown call will fail. (Issue #4397.) 1738 # and the client's shutdown call will fail. (Issue #4397.)
1739 self.done.wait() 1739 self.done.wait()
1740 1740
1741 def _testShutdown(self): 1741 def _testShutdown(self):
1742 self.serv_conn.send(MSG) 1742 self.serv_conn.send(MSG)
1743 self.serv_conn.shutdown(2)
1744
1745 testShutdown_overflow = support.cpython_only(testShutdown)
1746
1747 @support.cpython_only
1748 def _testShutdown_overflow(self):
1749 import _testcapi
1750 self.serv_conn.send(MSG)
1743 # Issue 15989 1751 # Issue 15989
1744 self.assertRaises(OverflowError, self.serv_conn.shutdown, 1752 self.assertRaises(OverflowError, self.serv_conn.shutdown,
1745 _testcapi.INT_MAX + 1) 1753 _testcapi.INT_MAX + 1)
1746 self.assertRaises(OverflowError, self.serv_conn.shutdown, 1754 self.assertRaises(OverflowError, self.serv_conn.shutdown,
1747 2 + (_testcapi.UINT_MAX + 1)) 1755 2 + (_testcapi.UINT_MAX + 1))
1748 self.serv_conn.shutdown(2) 1756 self.serv_conn.shutdown(2)
1749 1757
1750 def testDetach(self): 1758 def testDetach(self):
1751 # Testing detach() 1759 # Testing detach()
1752 fileno = self.cli_conn.fileno() 1760 fileno = self.cli_conn.fileno()
(...skipping 748 matching lines...) Expand 10 before | Expand all | Expand 10 after
2501 def _testRecvmsgIntoScatter(self): 2509 def _testRecvmsgIntoScatter(self):
2502 self.sendToServer(b"Mary had a little lamb") 2510 self.sendToServer(b"Mary had a little lamb")
2503 2511
2504 2512
2505 class CmsgMacroTests(unittest.TestCase): 2513 class CmsgMacroTests(unittest.TestCase):
2506 # Test the functions CMSG_LEN() and CMSG_SPACE(). Tests 2514 # Test the functions CMSG_LEN() and CMSG_SPACE(). Tests
2507 # assumptions used by sendmsg() and recvmsg[_into](), which share 2515 # assumptions used by sendmsg() and recvmsg[_into](), which share
2508 # code with these functions. 2516 # code with these functions.
2509 2517
2510 # Match the definition in socketmodule.c 2518 # Match the definition in socketmodule.c
2511 socklen_t_limit = min(0x7fffffff, _testcapi.INT_MAX) 2519 try:
2520 import _testcapi
2521 except ImportError:
2522 socklen_t_limit = 0x7fffffff
2523 else:
2524 socklen_t_limit = min(0x7fffffff, _testcapi.INT_MAX)
2512 2525
2513 @requireAttrs(socket, "CMSG_LEN") 2526 @requireAttrs(socket, "CMSG_LEN")
2514 def testCMSG_LEN(self): 2527 def testCMSG_LEN(self):
2515 # Test CMSG_LEN() with various valid and invalid values, 2528 # Test CMSG_LEN() with various valid and invalid values,
2516 # checking the assumptions used by recvmsg() and sendmsg(). 2529 # checking the assumptions used by recvmsg() and sendmsg().
2517 toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1 2530 toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1
2518 values = list(range(257)) + list(range(toobig - 257, toobig)) 2531 values = list(range(257)) + list(range(toobig - 257, toobig))
2519 2532
2520 # struct cmsghdr has at least three members, two of which are ints 2533 # struct cmsghdr has at least three members, two of which are ints
2521 self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2) 2534 self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2)
(...skipping 1108 matching lines...) Expand 10 before | Expand all | Expand 10 after
3630 # Check that func(*args, **kwargs), run in a loop, raises 3643 # Check that func(*args, **kwargs), run in a loop, raises
3631 # OSError with an errno of EINTR when interrupted by a 3644 # OSError with an errno of EINTR when interrupted by a
3632 # signal. 3645 # signal.
3633 with self.assertRaises(OSError) as cm: 3646 with self.assertRaises(OSError) as cm:
3634 while True: 3647 while True:
3635 self.setAlarm(self.alarm_time) 3648 self.setAlarm(self.alarm_time)
3636 func(*args, **kwargs) 3649 func(*args, **kwargs)
3637 self.assertNotIsInstance(cm.exception, socket.timeout) 3650 self.assertNotIsInstance(cm.exception, socket.timeout)
3638 self.assertEqual(cm.exception.errno, errno.EINTR) 3651 self.assertEqual(cm.exception.errno, errno.EINTR)
3639 3652
3640 # Issue #12958: The following tests have problems on Mac OS X 3653 # Issue #12958: The following tests have problems on OS X prior to 10.7
3641 @support.anticipate_failure(sys.platform == "darwin") 3654 @support.requires_mac_ver(10, 7)
3642 def testInterruptedSendTimeout(self): 3655 def testInterruptedSendTimeout(self):
3643 self.checkInterruptedSend(self.serv_conn.send, b"a"*512) 3656 self.checkInterruptedSend(self.serv_conn.send, b"a"*512)
3644 3657
3645 @support.anticipate_failure(sys.platform == "darwin") 3658 @support.requires_mac_ver(10, 7)
3646 def testInterruptedSendtoTimeout(self): 3659 def testInterruptedSendtoTimeout(self):
3647 # Passing an actual address here as Python's wrapper for 3660 # Passing an actual address here as Python's wrapper for
3648 # sendto() doesn't allow passing a zero-length one; POSIX 3661 # sendto() doesn't allow passing a zero-length one; POSIX
3649 # requires that the address is ignored since the socket is 3662 # requires that the address is ignored since the socket is
3650 # connection-mode, however. 3663 # connection-mode, however.
3651 self.checkInterruptedSend(self.serv_conn.sendto, b"a"*512, 3664 self.checkInterruptedSend(self.serv_conn.sendto, b"a"*512,
3652 self.serv_addr) 3665 self.serv_addr)
3653 3666
3654 @support.anticipate_failure(sys.platform == "darwin") 3667 @support.requires_mac_ver(10, 7)
3655 @requireAttrs(socket.socket, "sendmsg") 3668 @requireAttrs(socket.socket, "sendmsg")
3656 def testInterruptedSendmsgTimeout(self): 3669 def testInterruptedSendmsgTimeout(self):
3657 self.checkInterruptedSend(self.serv_conn.sendmsg, [b"a"*512]) 3670 self.checkInterruptedSend(self.serv_conn.sendmsg, [b"a"*512])
3658 3671
3659 3672
3660 @unittest.skipUnless(thread, 'Threading required for this test.') 3673 @unittest.skipUnless(thread, 'Threading required for this test.')
3661 class TCPCloserTest(ThreadedTCPSocketTest): 3674 class TCPCloserTest(ThreadedTCPSocketTest):
3662 3675
3663 def testClose(self): 3676 def testClose(self):
3664 conn, addr = self.serv.accept() 3677 conn, addr = self.serv.accept()
(...skipping 61 matching lines...) Expand 10 before | Expand all | Expand 10 after
3726 self.assertIsNone(self.serv.gettimeout()) 3739 self.assertIsNone(self.serv.gettimeout())
3727 self.serv.setblocking(False) 3740 self.serv.setblocking(False)
3728 self.assertEqual(self.serv.gettimeout(), 0.0) 3741 self.assertEqual(self.serv.gettimeout(), 0.0)
3729 start = time.time() 3742 start = time.time()
3730 try: 3743 try:
3731 self.serv.accept() 3744 self.serv.accept()
3732 except OSError: 3745 except OSError:
3733 pass 3746 pass
3734 end = time.time() 3747 end = time.time()
3735 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.") 3748 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.")
3736 # Issue 15989
3737 if _testcapi.UINT_MAX < _testcapi.ULONG_MAX:
3738 self.serv.setblocking(_testcapi.UINT_MAX + 1)
3739 self.assertIsNone(self.serv.gettimeout())
3740 3749
3741 def _testSetBlocking(self): 3750 def _testSetBlocking(self):
3742 pass 3751 pass
3752
3753 @support.cpython_only
3754 def testSetBlocking_overflow(self):
3755 # Issue 15989
3756 import _testcapi
3757 if _testcapi.UINT_MAX >= _testcapi.ULONG_MAX:
3758 self.skipTest('needs UINT_MAX < ULONG_MAX')
3759 self.serv.setblocking(False)
3760 self.assertEqual(self.serv.gettimeout(), 0.0)
3761 self.serv.setblocking(_testcapi.UINT_MAX + 1)
3762 self.assertIsNone(self.serv.gettimeout())
3763
3764 _testSetBlocking_overflow = support.cpython_only(_testSetBlocking)
3743 3765
3744 @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'), 3766 @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
3745 'test needs socket.SOCK_NONBLOCK') 3767 'test needs socket.SOCK_NONBLOCK')
3746 @support.requires_linux_version(2, 6, 28) 3768 @support.requires_linux_version(2, 6, 28)
3747 def testInitNonBlocking(self): 3769 def testInitNonBlocking(self):
3748 # reinit server socket 3770 # reinit server socket
3749 self.serv.close() 3771 self.serv.close()
3750 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM | 3772 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM |
3751 socket.SOCK_NONBLOCK) 3773 socket.SOCK_NONBLOCK)
3752 self.port = support.bind_port(self.serv) 3774 self.port = support.bind_port(self.serv)
(...skipping 922 matching lines...) Expand 10 before | Expand all | Expand 10 after
4675 _testRecvFromIntoBytearray = _testRecvFromIntoArray 4697 _testRecvFromIntoBytearray = _testRecvFromIntoArray
4676 4698
4677 def testRecvFromIntoMemoryview(self): 4699 def testRecvFromIntoMemoryview(self):
4678 buf = bytearray(1024) 4700 buf = bytearray(1024)
4679 nbytes, addr = self.cli_conn.recvfrom_into(memoryview(buf)) 4701 nbytes, addr = self.cli_conn.recvfrom_into(memoryview(buf))
4680 self.assertEqual(nbytes, len(MSG)) 4702 self.assertEqual(nbytes, len(MSG))
4681 msg = buf[:len(MSG)] 4703 msg = buf[:len(MSG)]
4682 self.assertEqual(msg, MSG) 4704 self.assertEqual(msg, MSG)
4683 4705
4684 _testRecvFromIntoMemoryview = _testRecvFromIntoArray 4706 _testRecvFromIntoMemoryview = _testRecvFromIntoArray
4707
4708 def testRecvFromIntoSmallBuffer(self):
4709 # See issue #20246.
4710 buf = bytearray(8)
4711 self.assertRaises(ValueError, self.cli_conn.recvfrom_into, buf, 1024)
4712
4713 def _testRecvFromIntoSmallBuffer(self):
4714 self.serv_conn.send(MSG)
4715
4716 def testRecvFromIntoEmptyBuffer(self):
4717 buf = bytearray()
4718 self.cli_conn.recvfrom_into(buf)
4719 self.cli_conn.recvfrom_into(buf, 0)
4720
4721 _testRecvFromIntoEmptyBuffer = _testRecvFromIntoArray
4685 4722
4686 4723
4687 TIPC_STYPE = 2000 4724 TIPC_STYPE = 2000
4688 TIPC_LOWER = 200 4725 TIPC_LOWER = 200
4689 TIPC_UPPER = 210 4726 TIPC_UPPER = 210
4690 4727
4691 def isTipcAvailable(): 4728 def isTipcAvailable():
4692 """Check if the TIPC module is loaded 4729 """Check if the TIPC module is loaded
4693 4730
4694 The TIPC module is not loaded automatically on Ubuntu and probably 4731 The TIPC module is not loaded automatically on Ubuntu and probably
(...skipping 389 matching lines...) Expand 10 before | Expand all | Expand 10 after
5084 InterruptedSendTimeoutTest, 5121 InterruptedSendTimeoutTest,
5085 TestSocketSharing, 5122 TestSocketSharing,
5086 ]) 5123 ])
5087 5124
5088 thread_info = support.threading_setup() 5125 thread_info = support.threading_setup()
5089 support.run_unittest(*tests) 5126 support.run_unittest(*tests)
5090 support.threading_cleanup(*thread_info) 5127 support.threading_cleanup(*thread_info)
5091 5128
5092 if __name__ == "__main__": 5129 if __name__ == "__main__":
5093 test_main() 5130 test_main()
LEFTRIGHT

RSS Feeds Recent Issues | This issue
This is Rietveld 894c83f36cb7+