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

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

Issue 16510: Using appropriate checks in tests
Left Patch Set: Created 5 years, 11 months ago
Right Patch Set: Created 5 years, 6 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 from unittest.case import _ExpectedFailure
6 3
7 import errno 4 import errno
8 import io 5 import io
9 import socket 6 import socket
10 import select 7 import select
11 import tempfile 8 import tempfile
12 import _testcapi
13 import time 9 import time
14 import traceback 10 import traceback
15 import queue 11 import queue
16 import sys 12 import sys
17 import os 13 import os
18 import array 14 import array
19 import platform 15 import platform
20 import contextlib 16 import contextlib
21 from weakref import proxy 17 from weakref import proxy
22 import signal 18 import signal
23 import math 19 import math
24 import pickle 20 import pickle
25 import struct 21 import struct
26 try: 22 try:
27 import fcntl
28 except ImportError:
29 fcntl = False
30 try:
31 import multiprocessing 23 import multiprocessing
32 except ImportError: 24 except ImportError:
33 multiprocessing = False 25 multiprocessing = False
26 try:
27 import fcntl
28 except ImportError:
29 fcntl = None
34 30
35 HOST = support.HOST 31 HOST = support.HOST
36 MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8') ## test unicode string and carriage return 32 MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8') ## test unicode string and carriage return
37 33
38 try: 34 try:
39 import _thread as thread 35 import _thread as thread
40 import threading 36 import threading
41 except ImportError: 37 except ImportError:
42 thread = None 38 thread = None
43 threading = None 39 threading = None
(...skipping 69 matching lines...) Expand 10 before | Expand all | Expand 10 after
113 class SocketCANTest(unittest.TestCase): 109 class SocketCANTest(unittest.TestCase):
114 110
115 """To be able to run this test, a `vcan0` CAN interface can be created with 111 """To be able to run this test, a `vcan0` CAN interface can be created with
116 the following commands: 112 the following commands:
117 # modprobe vcan 113 # modprobe vcan
118 # ip link add dev vcan0 type vcan 114 # ip link add dev vcan0 type vcan
119 # ifconfig vcan0 up 115 # ifconfig vcan0 up
120 """ 116 """
121 interface = 'vcan0' 117 interface = 'vcan0'
122 bufsize = 128 118 bufsize = 128
119
120 """The CAN frame structure is defined in <linux/can.h>:
121
122 struct can_frame {
123 canid_t can_id; /* 32 bit CAN_ID + EFF/RTR/ERR flags */
124 __u8 can_dlc; /* data length code: 0 .. 8 */
125 __u8 data[8] __attribute__((aligned(8)));
126 };
127 """
128 can_frame_fmt = "=IB3x8s"
129 can_frame_size = struct.calcsize(can_frame_fmt)
130
131 """The Broadcast Management Command frame structure is defined
132 in <linux/can/bcm.h>:
133
134 struct bcm_msg_head {
135 __u32 opcode;
136 __u32 flags;
137 __u32 count;
138 struct timeval ival1, ival2;
139 canid_t can_id;
140 __u32 nframes;
141 struct can_frame frames[0];
142 }
143
144 `bcm_msg_head` must be 8 bytes aligned because of the `frames` member (see
145 `struct can_frame` definition). Must use native not standard types for packi ng.
146 """
147 bcm_cmd_msg_fmt = "@3I4l2I"
148 bcm_cmd_msg_fmt += "x" * (struct.calcsize(bcm_cmd_msg_fmt) % 8)
123 149
124 def setUp(self): 150 def setUp(self):
125 self.s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) 151 self.s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
126 self.addCleanup(self.s.close) 152 self.addCleanup(self.s.close)
127 try: 153 try:
128 self.s.bind((self.interface,)) 154 self.s.bind((self.interface,))
129 except OSError: 155 except OSError:
130 self.skipTest('network interface `%s` does not exist' % 156 self.skipTest('network interface `%s` does not exist' %
131 self.interface) 157 self.interface)
132 158
(...skipping 102 matching lines...) Expand 10 before | Expand all | Expand 10 after
235 self.server_ready.wait() 261 self.server_ready.wait()
236 self.clientSetUp() 262 self.clientSetUp()
237 self.client_ready.set() 263 self.client_ready.set()
238 if self.server_crashed: 264 if self.server_crashed:
239 self.clientTearDown() 265 self.clientTearDown()
240 return 266 return
241 if not hasattr(test_func, '__call__'): 267 if not hasattr(test_func, '__call__'):
242 raise TypeError("test_func must be a callable function") 268 raise TypeError("test_func must be a callable function")
243 try: 269 try:
244 test_func() 270 test_func()
245 except _ExpectedFailure:
246 # We deliberately ignore expected failures
247 pass
248 except BaseException as e: 271 except BaseException as e:
249 self.queue.put(e) 272 self.queue.put(e)
250 finally: 273 finally:
251 self.clientTearDown() 274 self.clientTearDown()
252 275
253 def clientSetUp(self): 276 def clientSetUp(self):
254 raise NotImplementedError("clientSetUp must be implemented.") 277 raise NotImplementedError("clientSetUp must be implemented.")
255 278
256 def clientTearDown(self): 279 def clientTearDown(self):
257 self.done.set() 280 self.done.set()
(...skipping 278 matching lines...) Expand 10 before | Expand all | Expand 10 after
536 """Base class for SCTP tests in one-to-one (SOCK_STREAM) mode.""" 559 """Base class for SCTP tests in one-to-one (SOCK_STREAM) mode."""
537 560
538 def newSocket(self): 561 def newSocket(self):
539 return socket.socket(socket.AF_INET, socket.SOCK_STREAM, 562 return socket.socket(socket.AF_INET, socket.SOCK_STREAM,
540 socket.IPPROTO_SCTP) 563 socket.IPPROTO_SCTP)
541 564
542 565
543 class Inet6TestBase(InetTestBase): 566 class Inet6TestBase(InetTestBase):
544 """Base class for IPv6 socket tests.""" 567 """Base class for IPv6 socket tests."""
545 568
546 # Don't use "localhost" here - it may not have an IPv6 address 569 host = support.HOSTv6
547 # assigned to it by default (e.g. in /etc/hosts), and if someone
548 # has assigned it an IPv4-mapped address, then it's unlikely to
549 # work with the full IPv6 API.
550 host = "::1"
551 570
552 class UDP6TestBase(Inet6TestBase): 571 class UDP6TestBase(Inet6TestBase):
553 """Base class for UDP-over-IPv6 tests.""" 572 """Base class for UDP-over-IPv6 tests."""
554 573
555 def newSocket(self): 574 def newSocket(self):
556 return socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) 575 return socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
557 576
558 577
559 # Test-skipping decorators for use with ThreadableTest. 578 # Test-skipping decorators for use with ThreadableTest.
560 579
(...skipping 58 matching lines...) Expand 10 before | Expand all | Expand 10 after
619 ", ".join(str(o) for o in args), err)) 638 ", ".join(str(o) for o in args), err))
620 639
621 640
622 ####################################################################### 641 #######################################################################
623 ## Begin Tests 642 ## Begin Tests
624 643
625 class GeneralModuleTests(unittest.TestCase): 644 class GeneralModuleTests(unittest.TestCase):
626 645
627 def test_repr(self): 646 def test_repr(self):
628 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 647 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
629 self.addCleanup(s.close) 648 with s:
630 self.assertTrue(repr(s).startswith("<socket.socket object")) 649 self.assertIn('fd=%i' % s.fileno(), repr(s))
650 self.assertIn('family=%s' % socket.AF_INET, repr(s))
651 self.assertIn('type=%s' % socket.SOCK_STREAM, repr(s))
652 self.assertIn('proto=0', repr(s))
653 self.assertNotIn('raddr', repr(s))
654 s.bind(('127.0.0.1', 0))
655 self.assertIn('laddr', repr(s))
656 self.assertIn(str(s.getsockname()), repr(s))
657 self.assertIn('[closed]', repr(s))
658 self.assertNotIn('laddr', repr(s))
631 659
632 def test_weakref(self): 660 def test_weakref(self):
633 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 661 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
634 p = proxy(s) 662 p = proxy(s)
635 self.assertEqual(p.fileno(), s.fileno()) 663 self.assertEqual(p.fileno(), s.fileno())
636 s.close() 664 s.close()
637 s = None 665 s = None
638 try: 666 try:
639 p.fileno() 667 p.fileno()
640 except ReferenceError: 668 except ReferenceError:
(...skipping 66 matching lines...) Expand 10 before | Expand all | Expand 10 after
707 socket.SOL_SOCKET 735 socket.SOL_SOCKET
708 socket.SO_REUSEADDR 736 socket.SO_REUSEADDR
709 737
710 def testHostnameRes(self): 738 def testHostnameRes(self):
711 # Testing hostname resolution mechanisms 739 # Testing hostname resolution mechanisms
712 hostname = socket.gethostname() 740 hostname = socket.gethostname()
713 try: 741 try:
714 ip = socket.gethostbyname(hostname) 742 ip = socket.gethostbyname(hostname)
715 except OSError: 743 except OSError:
716 # Probably name lookup wasn't set up right; skip this test 744 # Probably name lookup wasn't set up right; skip this test
717 return 745 self.skipTest('name lookup failure')
718 self.assertGreaterEqual(ip.find('.'), 0, "Error resolving host to ip.") 746 self.assertIn('.', ip, "Error resolving host to ip.")
719 try: 747 try:
720 hname, aliases, ipaddrs = socket.gethostbyaddr(ip) 748 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
721 except OSError: 749 except OSError:
722 # Probably a similar problem as above; skip this test 750 # Probably a similar problem as above; skip this test
723 return 751 self.skipTest('name lookup failure')
724 all_host_names = [hostname, hname] + aliases 752 all_host_names = [hostname, hname] + aliases
725 fqhn = socket.getfqdn(ip) 753 fqhn = socket.getfqdn(ip)
726 if not fqhn in all_host_names: 754 if not fqhn in all_host_names:
727 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)))
756
757 def test_host_resolution(self):
758 for addr in ['0.1.1.~1', '1+.1.1.1', '::1q', '::1::2',
759 '1:1:1:1:1:1:1:1:1']:
760 self.assertRaises(OSError, socket.gethostbyname, addr)
761 self.assertRaises(OSError, socket.gethostbyaddr, addr)
762
763 for addr in [support.HOST, '10.0.0.1', '255.255.255.255']:
764 self.assertEqual(socket.gethostbyname(addr), addr)
765
766 # we don't test support.HOSTv6 because there's a chance it doesn't have
767 # a matching name entry (e.g. 'ip6-localhost')
768 for host in [support.HOST]:
769 self.assertIn(host, socket.gethostbyaddr(host)[2])
728 770
729 @unittest.skipUnless(hasattr(socket, 'sethostname'), "test needs socket.seth ostname()") 771 @unittest.skipUnless(hasattr(socket, 'sethostname'), "test needs socket.seth ostname()")
730 @unittest.skipUnless(hasattr(socket, 'gethostname'), "test needs socket.geth ostname()") 772 @unittest.skipUnless(hasattr(socket, 'gethostname'), "test needs socket.geth ostname()")
731 def test_sethostname(self): 773 def test_sethostname(self):
732 oldhn = socket.gethostname() 774 oldhn = socket.gethostname()
733 try: 775 try:
734 socket.sethostname('new') 776 socket.sethostname('new')
735 except OSError as e: 777 except OSError as e:
736 if e.errno == errno.EPERM: 778 if e.errno == errno.EPERM:
737 self.skipTest("test should be run as root") 779 self.skipTest("test should be run as root")
(...skipping 27 matching lines...) Expand all
765 @unittest.skipUnless(hasattr(socket, 'if_nameindex'), 807 @unittest.skipUnless(hasattr(socket, 'if_nameindex'),
766 'socket.if_nameindex() not available.') 808 'socket.if_nameindex() not available.')
767 def testInvalidInterfaceNameIndex(self): 809 def testInvalidInterfaceNameIndex(self):
768 # test nonexistent interface index/name 810 # test nonexistent interface index/name
769 self.assertRaises(OSError, socket.if_indextoname, 0) 811 self.assertRaises(OSError, socket.if_indextoname, 0)
770 self.assertRaises(OSError, socket.if_nametoindex, '_DEADBEEF') 812 self.assertRaises(OSError, socket.if_nametoindex, '_DEADBEEF')
771 # test with invalid values 813 # test with invalid values
772 self.assertRaises(TypeError, socket.if_nametoindex, 0) 814 self.assertRaises(TypeError, socket.if_nametoindex, 0)
773 self.assertRaises(TypeError, socket.if_indextoname, '_DEADBEEF') 815 self.assertRaises(TypeError, socket.if_indextoname, '_DEADBEEF')
774 816
817 @unittest.skipUnless(hasattr(sys, 'getrefcount'),
818 'test needs sys.getrefcount()')
775 def testRefCountGetNameInfo(self): 819 def testRefCountGetNameInfo(self):
776 # Testing reference count for getnameinfo 820 # Testing reference count for getnameinfo
777 if hasattr(sys, "getrefcount"): 821 try:
778 try: 822 # On some versions, this loses a reference
779 # On some versions, this loses a reference 823 orig = sys.getrefcount(__name__)
780 orig = sys.getrefcount(__name__) 824 socket.getnameinfo(__name__,0)
781 socket.getnameinfo(__name__,0) 825 except TypeError:
782 except TypeError: 826 if sys.getrefcount(__name__) != orig:
783 if sys.getrefcount(__name__) != orig: 827 self.fail("socket.getnameinfo loses a reference")
784 self.fail("socket.getnameinfo loses a reference")
785 828
786 def testInterpreterCrash(self): 829 def testInterpreterCrash(self):
787 # Making sure getnameinfo doesn't crash the interpreter 830 # Making sure getnameinfo doesn't crash the interpreter
788 try: 831 try:
789 # On some versions, this crashes the interpreter. 832 # On some versions, this crashes the interpreter.
790 socket.getnameinfo(('x', 0, 0, 0), 0) 833 socket.getnameinfo(('x', 0, 0, 0), 0)
791 except OSError: 834 except OSError:
792 pass 835 pass
793 836
794 def testNtoH(self): 837 def testNtoH(self):
(...skipping 40 matching lines...) Expand 10 before | Expand all | Expand 10 after
835 try: 878 try:
836 port = socket.getservbyname(service, 'tcp') 879 port = socket.getservbyname(service, 'tcp')
837 break 880 break
838 except OSError: 881 except OSError:
839 pass 882 pass
840 else: 883 else:
841 raise OSError 884 raise OSError
842 # Try same call with optional protocol omitted 885 # Try same call with optional protocol omitted
843 port2 = socket.getservbyname(service) 886 port2 = socket.getservbyname(service)
844 eq(port, port2) 887 eq(port, port2)
845 # Try udp, but don't barf it it doesn't exist 888 # Try udp, but don't barf if it doesn't exist
846 try: 889 try:
847 udpport = socket.getservbyname(service, 'udp') 890 udpport = socket.getservbyname(service, 'udp')
848 except OSError: 891 except OSError:
849 udpport = None 892 udpport = None
850 else: 893 else:
851 eq(udpport, port) 894 eq(udpport, port)
852 # Now make sure the lookup by port returns the same service name 895 # Now make sure the lookup by port returns the same service name
853 eq(socket.getservbyport(port2), service) 896 eq(socket.getservbyport(port2), service)
854 eq(socket.getservbyport(port, 'tcp'), service) 897 eq(socket.getservbyport(port, 'tcp'), service)
855 if udpport is not None: 898 if udpport is not None:
(...skipping 23 matching lines...) Expand all
879 s = socket.socket() 922 s = socket.socket()
880 self.assertEqual(s.gettimeout(), None) 923 self.assertEqual(s.gettimeout(), None)
881 s.close() 924 s.close()
882 925
883 # Check that setting it to an invalid value raises ValueError 926 # Check that setting it to an invalid value raises ValueError
884 self.assertRaises(ValueError, socket.setdefaulttimeout, -1) 927 self.assertRaises(ValueError, socket.setdefaulttimeout, -1)
885 928
886 # Check that setting it to an invalid type raises TypeError 929 # Check that setting it to an invalid type raises TypeError
887 self.assertRaises(TypeError, socket.setdefaulttimeout, "spam") 930 self.assertRaises(TypeError, socket.setdefaulttimeout, "spam")
888 931
932 @unittest.skipUnless(hasattr(socket, 'inet_aton'),
933 'test needs socket.inet_aton()')
889 def testIPv4_inet_aton_fourbytes(self): 934 def testIPv4_inet_aton_fourbytes(self):
890 if not hasattr(socket, 'inet_aton'):
891 return # No inet_aton, nothing to check
892 # Test that issue1008086 and issue767150 are fixed. 935 # Test that issue1008086 and issue767150 are fixed.
893 # It must return 4 bytes. 936 # It must return 4 bytes.
894 self.assertEqual(b'\x00'*4, socket.inet_aton('0.0.0.0')) 937 self.assertEqual(b'\x00'*4, socket.inet_aton('0.0.0.0'))
895 self.assertEqual(b'\xff'*4, socket.inet_aton('255.255.255.255')) 938 self.assertEqual(b'\xff'*4, socket.inet_aton('255.255.255.255'))
896 939
940 @unittest.skipUnless(hasattr(socket, 'inet_pton'),
941 'test needs socket.inet_pton()')
897 def testIPv4toString(self): 942 def testIPv4toString(self):
898 if not hasattr(socket, 'inet_pton'):
899 return # No inet_pton() on this platform
900 from socket import inet_aton as f, inet_pton, AF_INET 943 from socket import inet_aton as f, inet_pton, AF_INET
901 g = lambda a: inet_pton(AF_INET, a) 944 g = lambda a: inet_pton(AF_INET, a)
902 945
903 assertInvalid = lambda func,a: self.assertRaises( 946 assertInvalid = lambda func,a: self.assertRaises(
904 (OSError, ValueError), func, a 947 (OSError, ValueError), func, a
905 ) 948 )
906 949
907 self.assertEqual(b'\x00\x00\x00\x00', f('0.0.0.0')) 950 self.assertEqual(b'\x00\x00\x00\x00', f('0.0.0.0'))
908 self.assertEqual(b'\xff\x00\xff\x00', f('255.0.255.0')) 951 self.assertEqual(b'\xff\x00\xff\x00', f('255.0.255.0'))
909 self.assertEqual(b'\xaa\xaa\xaa\xaa', f('170.170.170.170')) 952 self.assertEqual(b'\xaa\xaa\xaa\xaa', f('170.170.170.170'))
910 self.assertEqual(b'\x01\x02\x03\x04', f('1.2.3.4')) 953 self.assertEqual(b'\x01\x02\x03\x04', f('1.2.3.4'))
911 self.assertEqual(b'\xff\xff\xff\xff', f('255.255.255.255')) 954 self.assertEqual(b'\xff\xff\xff\xff', f('255.255.255.255'))
912 assertInvalid(f, '0.0.0.') 955 assertInvalid(f, '0.0.0.')
913 assertInvalid(f, '300.0.0.0') 956 assertInvalid(f, '300.0.0.0')
914 assertInvalid(f, 'a.0.0.0') 957 assertInvalid(f, 'a.0.0.0')
915 assertInvalid(f, '1.2.3.4.5') 958 assertInvalid(f, '1.2.3.4.5')
916 assertInvalid(f, '::1') 959 assertInvalid(f, '::1')
917 960
918 self.assertEqual(b'\x00\x00\x00\x00', g('0.0.0.0')) 961 self.assertEqual(b'\x00\x00\x00\x00', g('0.0.0.0'))
919 self.assertEqual(b'\xff\x00\xff\x00', g('255.0.255.0')) 962 self.assertEqual(b'\xff\x00\xff\x00', g('255.0.255.0'))
920 self.assertEqual(b'\xaa\xaa\xaa\xaa', g('170.170.170.170')) 963 self.assertEqual(b'\xaa\xaa\xaa\xaa', g('170.170.170.170'))
921 self.assertEqual(b'\xff\xff\xff\xff', g('255.255.255.255')) 964 self.assertEqual(b'\xff\xff\xff\xff', g('255.255.255.255'))
922 assertInvalid(g, '0.0.0.') 965 assertInvalid(g, '0.0.0.')
923 assertInvalid(g, '300.0.0.0') 966 assertInvalid(g, '300.0.0.0')
924 assertInvalid(g, 'a.0.0.0') 967 assertInvalid(g, 'a.0.0.0')
925 assertInvalid(g, '1.2.3.4.5') 968 assertInvalid(g, '1.2.3.4.5')
926 assertInvalid(g, '::1') 969 assertInvalid(g, '::1')
927 970
971 @unittest.skipUnless(hasattr(socket, 'inet_pton'),
972 'test needs socket.inet_pton()')
928 def testIPv6toString(self): 973 def testIPv6toString(self):
929 if not hasattr(socket, 'inet_pton'):
930 return # No inet_pton() on this platform
931 try: 974 try:
932 from socket import inet_pton, AF_INET6, has_ipv6 975 from socket import inet_pton, AF_INET6, has_ipv6
933 if not has_ipv6: 976 if not has_ipv6:
934 return 977 self.skipTest('IPv6 not available')
935 except ImportError: 978 except ImportError:
936 return 979 self.skipTest('could not import needed symbols from socket')
980
981 if sys.platform == "win32":
982 try:
983 inet_pton(AF_INET6, '::')
984 except OSError as e:
985 if e.winerror == 10022:
986 self.skipTest('IPv6 might not be supported')
987
937 f = lambda a: inet_pton(AF_INET6, a) 988 f = lambda a: inet_pton(AF_INET6, a)
938 assertInvalid = lambda a: self.assertRaises( 989 assertInvalid = lambda a: self.assertRaises(
939 (OSError, ValueError), f, a 990 (OSError, ValueError), f, a
940 ) 991 )
941 992
942 self.assertEqual(b'\x00' * 16, f('::')) 993 self.assertEqual(b'\x00' * 16, f('::'))
943 self.assertEqual(b'\x00' * 16, f('0::0')) 994 self.assertEqual(b'\x00' * 16, f('0::0'))
944 self.assertEqual(b'\x00\x01' + b'\x00' * 14, f('1::')) 995 self.assertEqual(b'\x00\x01' + b'\x00' * 14, f('1::'))
945 self.assertEqual( 996 self.assertEqual(
946 b'\x45\xef\x76\xcb\x00\x1a\x56\xef\xaf\xeb\x0b\xac\x19\x24\xae\xae', 997 b'\x45\xef\x76\xcb\x00\x1a\x56\xef\xaf\xeb\x0b\xac\x19\x24\xae\xae',
(...skipping 25 matching lines...) Expand all
972 b'\x00\x42\xa8\xb9\x00\x00\x00\x02\xff\xff\xa2\x9b\xfe\x2a\x17\x40', 1023 b'\x00\x42\xa8\xb9\x00\x00\x00\x02\xff\xff\xa2\x9b\xfe\x2a\x17\x40',
973 f('42:a8b9:0:2:ffff:a29b:254.42.23.64') 1024 f('42:a8b9:0:2:ffff:a29b:254.42.23.64')
974 ) 1025 )
975 assertInvalid('255.254.253.252') 1026 assertInvalid('255.254.253.252')
976 assertInvalid('1::260.2.3.0') 1027 assertInvalid('1::260.2.3.0')
977 assertInvalid('1::0.be.e.0') 1028 assertInvalid('1::0.be.e.0')
978 assertInvalid('1:2:3:4:5:6:7:1.2.3.4') 1029 assertInvalid('1:2:3:4:5:6:7:1.2.3.4')
979 assertInvalid('::1.2.3.4:0') 1030 assertInvalid('::1.2.3.4:0')
980 assertInvalid('0.100.200.0:3:4:5:6:7:8') 1031 assertInvalid('0.100.200.0:3:4:5:6:7:8')
981 1032
1033 @unittest.skipUnless(hasattr(socket, 'inet_ntop'),
1034 'test needs socket.inet_ntop()')
982 def testStringToIPv4(self): 1035 def testStringToIPv4(self):
983 if not hasattr(socket, 'inet_ntop'):
984 return # No inet_ntop() on this platform
985 from socket import inet_ntoa as f, inet_ntop, AF_INET 1036 from socket import inet_ntoa as f, inet_ntop, AF_INET
986 g = lambda a: inet_ntop(AF_INET, a) 1037 g = lambda a: inet_ntop(AF_INET, a)
987 assertInvalid = lambda func,a: self.assertRaises( 1038 assertInvalid = lambda func,a: self.assertRaises(
988 (OSError, ValueError), func, a 1039 (OSError, ValueError), func, a
989 ) 1040 )
990 1041
991 self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00')) 1042 self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00'))
992 self.assertEqual('170.85.170.85', f(b'\xaa\x55\xaa\x55')) 1043 self.assertEqual('170.85.170.85', f(b'\xaa\x55\xaa\x55'))
993 self.assertEqual('255.255.255.255', f(b'\xff\xff\xff\xff')) 1044 self.assertEqual('255.255.255.255', f(b'\xff\xff\xff\xff'))
994 self.assertEqual('1.2.3.4', f(b'\x01\x02\x03\x04')) 1045 self.assertEqual('1.2.3.4', f(b'\x01\x02\x03\x04'))
995 assertInvalid(f, b'\x00' * 3) 1046 assertInvalid(f, b'\x00' * 3)
996 assertInvalid(f, b'\x00' * 5) 1047 assertInvalid(f, b'\x00' * 5)
997 assertInvalid(f, b'\x00' * 16) 1048 assertInvalid(f, b'\x00' * 16)
998 1049
999 self.assertEqual('1.0.1.0', g(b'\x01\x00\x01\x00')) 1050 self.assertEqual('1.0.1.0', g(b'\x01\x00\x01\x00'))
1000 self.assertEqual('170.85.170.85', g(b'\xaa\x55\xaa\x55')) 1051 self.assertEqual('170.85.170.85', g(b'\xaa\x55\xaa\x55'))
1001 self.assertEqual('255.255.255.255', g(b'\xff\xff\xff\xff')) 1052 self.assertEqual('255.255.255.255', g(b'\xff\xff\xff\xff'))
1002 assertInvalid(g, b'\x00' * 3) 1053 assertInvalid(g, b'\x00' * 3)
1003 assertInvalid(g, b'\x00' * 5) 1054 assertInvalid(g, b'\x00' * 5)
1004 assertInvalid(g, b'\x00' * 16) 1055 assertInvalid(g, b'\x00' * 16)
1005 1056
1057 @unittest.skipUnless(hasattr(socket, 'inet_ntop'),
1058 'test needs socket.inet_ntop()')
1006 def testStringToIPv6(self): 1059 def testStringToIPv6(self):
1007 if not hasattr(socket, 'inet_ntop'):
1008 return # No inet_ntop() on this platform
1009 try: 1060 try:
1010 from socket import inet_ntop, AF_INET6, has_ipv6 1061 from socket import inet_ntop, AF_INET6, has_ipv6
1011 if not has_ipv6: 1062 if not has_ipv6:
1012 return 1063 self.skipTest('IPv6 not available')
1013 except ImportError: 1064 except ImportError:
1014 return 1065 self.skipTest('could not import needed symbols from socket')
1066
1067 if sys.platform == "win32":
1068 try:
1069 inet_ntop(AF_INET6, b'\x00' * 16)
1070 except OSError as e:
1071 if e.winerror == 10022:
1072 self.skipTest('IPv6 might not be supported')
1073
1015 f = lambda a: inet_ntop(AF_INET6, a) 1074 f = lambda a: inet_ntop(AF_INET6, a)
1016 assertInvalid = lambda a: self.assertRaises( 1075 assertInvalid = lambda a: self.assertRaises(
1017 (OSError, ValueError), f, a 1076 (OSError, ValueError), f, a
1018 ) 1077 )
1019 1078
1020 self.assertEqual('::', f(b'\x00' * 16)) 1079 self.assertEqual('::', f(b'\x00' * 16))
1021 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01')) 1080 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01'))
1022 self.assertEqual( 1081 self.assertEqual(
1023 'aef:b01:506:1001:ffff:9997:55:170', 1082 'aef:b01:506:1001:ffff:9997:55:170',
1024 f(b'\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70 ') 1083 f(b'\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70 ')
(...skipping 12 matching lines...) Expand all
1037 self.addCleanup(sock.close) 1096 self.addCleanup(sock.close)
1038 sock.bind(("0.0.0.0", port)) 1097 sock.bind(("0.0.0.0", port))
1039 name = sock.getsockname() 1098 name = sock.getsockname()
1040 # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate 1099 # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate
1041 # 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.
1042 # 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.
1043 try: 1102 try:
1044 my_ip_addr = socket.gethostbyname(socket.gethostname()) 1103 my_ip_addr = socket.gethostbyname(socket.gethostname())
1045 except OSError: 1104 except OSError:
1046 # Probably name lookup wasn't set up right; skip this test 1105 # Probably name lookup wasn't set up right; skip this test
1047 return 1106 self.skipTest('name lookup failure')
1048 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])
1049 self.assertEqual(name[1], port) 1108 self.assertEqual(name[1], port)
1050 1109
1051 def testGetSockOpt(self): 1110 def testGetSockOpt(self):
1052 # Testing getsockopt() 1111 # Testing getsockopt()
1053 # We know a socket should start without reuse==0 1112 # We know a socket should start without reuse==0
1054 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1113 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1055 self.addCleanup(sock.close) 1114 self.addCleanup(sock.close)
1056 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1115 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1057 self.assertEqual(reuse, 0, "initial mode is reuse") 1116 self.assertEqual(reuse, 0, "initial mode is reuse")
1058 1117
1059 def testSetSockOpt(self): 1118 def testSetSockOpt(self):
1060 # Testing setsockopt() 1119 # Testing setsockopt()
1061 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1120 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1062 self.addCleanup(sock.close) 1121 self.addCleanup(sock.close)
1063 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 1122 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1064 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1123 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1065 self.assertNotEqual(reuse, 0, "failed to set reuse mode") 1124 self.assertNotEqual(reuse, 0, "failed to set reuse mode")
1066 1125
1067 def testSendAfterClose(self): 1126 def testSendAfterClose(self):
1068 # testing send() after close() with timeout 1127 # testing send() after close() with timeout
1069 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1128 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1070 sock.settimeout(1) 1129 sock.settimeout(1)
1071 sock.close() 1130 sock.close()
1072 self.assertRaises(OSError, sock.send, b"spam") 1131 self.assertRaises(OSError, sock.send, b"spam")
1073 1132
1074 def testNewAttributes(self): 1133 def testNewAttributes(self):
1075 # testing .family, .type and .protocol 1134 # testing .family, .type and .protocol
1135
1076 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1136 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1077 self.assertEqual(sock.family, socket.AF_INET) 1137 self.assertEqual(sock.family, socket.AF_INET)
1078 self.assertEqual(sock.type, socket.SOCK_STREAM) 1138 if hasattr(socket, 'SOCK_CLOEXEC'):
1139 self.assertIn(sock.type,
1140 (socket.SOCK_STREAM | socket.SOCK_CLOEXEC,
1141 socket.SOCK_STREAM))
1142 else:
1143 self.assertEqual(sock.type, socket.SOCK_STREAM)
1079 self.assertEqual(sock.proto, 0) 1144 self.assertEqual(sock.proto, 0)
1080 sock.close() 1145 sock.close()
1081 1146
1082 def test_getsockaddrarg(self): 1147 def test_getsockaddrarg(self):
1083 host = '0.0.0.0' 1148 host = '0.0.0.0'
1084 port = support.find_unused_port() 1149 port = support.find_unused_port()
1085 big_port = port + 65536 1150 big_port = port + 65536
1086 neg_port = port - 65536 1151 neg_port = port - 65536
1087 sock = socket.socket() 1152 sock = socket.socket()
1088 try: 1153 try:
(...skipping 32 matching lines...) Expand 10 before | Expand all | Expand 10 after
1121 socket.getaddrinfo('127.0.0.1', 80) 1186 socket.getaddrinfo('127.0.0.1', 80)
1122 socket.getaddrinfo(None, 80) 1187 socket.getaddrinfo(None, 80)
1123 if support.IPV6_ENABLED: 1188 if support.IPV6_ENABLED:
1124 socket.getaddrinfo('::1', 80) 1189 socket.getaddrinfo('::1', 80)
1125 # port can be a string service name such as "http", a numeric 1190 # port can be a string service name such as "http", a numeric
1126 # port number or None 1191 # port number or None
1127 socket.getaddrinfo(HOST, "http") 1192 socket.getaddrinfo(HOST, "http")
1128 socket.getaddrinfo(HOST, 80) 1193 socket.getaddrinfo(HOST, 80)
1129 socket.getaddrinfo(HOST, None) 1194 socket.getaddrinfo(HOST, None)
1130 # test family and socktype filters 1195 # test family and socktype filters
1131 infos = socket.getaddrinfo(HOST, None, socket.AF_INET) 1196 infos = socket.getaddrinfo(HOST, 80, socket.AF_INET, socket.SOCK_STREAM)
1132 for family, _, _, _, _ in infos: 1197 for family, type, _, _, _ in infos:
1133 self.assertEqual(family, socket.AF_INET) 1198 self.assertEqual(family, socket.AF_INET)
1199 self.assertEqual(str(family), 'AddressFamily.AF_INET')
1200 self.assertEqual(type, socket.SOCK_STREAM)
1201 self.assertEqual(str(type), 'SocketType.SOCK_STREAM')
1134 infos = socket.getaddrinfo(HOST, None, 0, socket.SOCK_STREAM) 1202 infos = socket.getaddrinfo(HOST, None, 0, socket.SOCK_STREAM)
1135 for _, socktype, _, _, _ in infos: 1203 for _, socktype, _, _, _ in infos:
1136 self.assertEqual(socktype, socket.SOCK_STREAM) 1204 self.assertEqual(socktype, socket.SOCK_STREAM)
1137 # test proto and flags arguments 1205 # test proto and flags arguments
1138 socket.getaddrinfo(HOST, None, 0, 0, socket.SOL_TCP) 1206 socket.getaddrinfo(HOST, None, 0, 0, socket.SOL_TCP)
1139 socket.getaddrinfo(HOST, None, 0, 0, 0, socket.AI_PASSIVE) 1207 socket.getaddrinfo(HOST, None, 0, 0, 0, socket.AI_PASSIVE)
1140 # a server willing to support both IPv4 and IPv6 will 1208 # a server willing to support both IPv4 and IPv6 will
1141 # usually do this 1209 # usually do this
1142 socket.getaddrinfo(None, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, 1210 socket.getaddrinfo(None, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0,
1143 socket.AI_PASSIVE) 1211 socket.AI_PASSIVE)
(...skipping 15 matching lines...) Expand all
1159 self.assertEqual(a, b) 1227 self.assertEqual(a, b)
1160 a = socket.getaddrinfo(None, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, 1228 a = socket.getaddrinfo(None, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0,
1161 socket.AI_PASSIVE) 1229 socket.AI_PASSIVE)
1162 b = socket.getaddrinfo(host=None, port=0, family=socket.AF_UNSPEC, 1230 b = socket.getaddrinfo(host=None, port=0, family=socket.AF_UNSPEC,
1163 type=socket.SOCK_STREAM, proto=0, 1231 type=socket.SOCK_STREAM, proto=0,
1164 flags=socket.AI_PASSIVE) 1232 flags=socket.AI_PASSIVE)
1165 self.assertEqual(a, b) 1233 self.assertEqual(a, b)
1166 # Issue #6697. 1234 # Issue #6697.
1167 self.assertRaises(UnicodeEncodeError, socket.getaddrinfo, 'localhost', ' \uD800') 1235 self.assertRaises(UnicodeEncodeError, socket.getaddrinfo, 'localhost', ' \uD800')
1168 1236
1237 # Issue 17269
1238 if hasattr(socket, 'AI_NUMERICSERV'):
1239 socket.getaddrinfo("localhost", None, 0, 0, 0, socket.AI_NUMERICSERV )
1240
1169 def test_getnameinfo(self): 1241 def test_getnameinfo(self):
1170 # only IP addresses are allowed 1242 # only IP addresses are allowed
1171 self.assertRaises(OSError, socket.getnameinfo, ('mail.python.org',0), 0) 1243 self.assertRaises(OSError, socket.getnameinfo, ('mail.python.org',0), 0)
1172 1244
1173 @unittest.skipUnless(support.is_resource_enabled('network'), 1245 @unittest.skipUnless(support.is_resource_enabled('network'),
1174 'network is not enabled') 1246 'network is not enabled')
1175 def test_idna(self): 1247 def test_idna(self):
1176 # Check for internet access before running test (issue #12804). 1248 # Check for internet access before running test (issue #12804).
1177 try: 1249 try:
1178 socket.gethostbyname('python.org') 1250 socket.gethostbyname('python.org')
(...skipping 20 matching lines...) Expand all
1199 self.assertRaises(ValueError, math.acosh, 0) 1271 self.assertRaises(ValueError, math.acosh, 0)
1200 1 // 0 1272 1 // 0
1201 c, s = socket.socketpair() 1273 c, s = socket.socketpair()
1202 old_alarm = signal.signal(signal.SIGALRM, raising_handler) 1274 old_alarm = signal.signal(signal.SIGALRM, raising_handler)
1203 try: 1275 try:
1204 if with_timeout: 1276 if with_timeout:
1205 # Just above the one second minimum for signal.alarm 1277 # Just above the one second minimum for signal.alarm
1206 c.settimeout(1.5) 1278 c.settimeout(1.5)
1207 with self.assertRaises(ZeroDivisionError): 1279 with self.assertRaises(ZeroDivisionError):
1208 signal.alarm(1) 1280 signal.alarm(1)
1209 c.sendall(b"x" * (1024**2)) 1281 c.sendall(b"x" * support.SOCK_MAX_SIZE)
1210 if with_timeout: 1282 if with_timeout:
1211 signal.signal(signal.SIGALRM, ok_handler) 1283 signal.signal(signal.SIGALRM, ok_handler)
1212 signal.alarm(1) 1284 signal.alarm(1)
1213 self.assertRaises(socket.timeout, c.sendall, b"x" * (1024**2)) 1285 self.assertRaises(socket.timeout, c.sendall,
1286 b"x" * support.SOCK_MAX_SIZE)
1214 finally: 1287 finally:
1288 signal.alarm(0)
1215 signal.signal(signal.SIGALRM, old_alarm) 1289 signal.signal(signal.SIGALRM, old_alarm)
1216 c.close() 1290 c.close()
1217 s.close() 1291 s.close()
1218 1292
1219 def test_sendall_interrupted(self): 1293 def test_sendall_interrupted(self):
1220 self.check_sendall_interrupted(False) 1294 self.check_sendall_interrupted(False)
1221 1295
1222 def test_sendall_interrupted_with_timeout(self): 1296 def test_sendall_interrupted_with_timeout(self):
1223 self.check_sendall_interrupted(True) 1297 self.check_sendall_interrupted(True)
1224 1298
(...skipping 37 matching lines...) Expand 10 before | Expand all | Expand 10 after
1262 for protocol in range(pickle.HIGHEST_PROTOCOL + 1): 1336 for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
1263 self.assertRaises(TypeError, pickle.dumps, sock, protocol) 1337 self.assertRaises(TypeError, pickle.dumps, sock, protocol)
1264 1338
1265 def test_listen_backlog(self): 1339 def test_listen_backlog(self):
1266 for backlog in 0, -1: 1340 for backlog in 0, -1:
1267 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1341 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1268 srv.bind((HOST, 0)) 1342 srv.bind((HOST, 0))
1269 srv.listen(backlog) 1343 srv.listen(backlog)
1270 srv.close() 1344 srv.close()
1271 1345
1346 @support.cpython_only
1347 def test_listen_backlog_overflow(self):
1272 # Issue 15989 1348 # Issue 15989
1349 import _testcapi
1273 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1350 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1274 srv.bind((HOST, 0)) 1351 srv.bind((HOST, 0))
1275 self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1) 1352 self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1)
1276 srv.close() 1353 srv.close()
1277 1354
1278 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.') 1355 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.')
1279 def test_flowinfo(self): 1356 def test_flowinfo(self):
1280 self.assertRaises(OverflowError, socket.getnameinfo, 1357 self.assertRaises(OverflowError, socket.getnameinfo,
1281 ('::1',0, 0xffffffff), 0) 1358 (support.HOSTv6, 0, 0xffffffff), 0)
1282 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: 1359 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
1283 self.assertRaises(OverflowError, s.bind, ('::1', 0, -10)) 1360 self.assertRaises(OverflowError, s.bind, (support.HOSTv6, 0, -10))
1284 1361
1362 def test_str_for_enums(self):
1363 # Make sure that the AF_* and SOCK_* constants have enum-like string
1364 # reprs.
1365 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
1366 self.assertEqual(str(s.family), 'AddressFamily.AF_INET')
1367 self.assertEqual(str(s.type), 'SocketType.SOCK_STREAM')
1368
1369 @unittest.skipIf(os.name == 'nt', 'Will not work on Windows')
1370 def test_uknown_socket_family_repr(self):
1371 # Test that when created with a family that's not one of the known
1372 # AF_*/SOCK_* constants, socket.family just returns the number.
1373 #
1374 # To do this we fool socket.socket into believing it already has an
1375 # open fd because on this path it doesn't actually verify the family and
1376 # type and populates the socket object.
1377 #
1378 # On Windows this trick won't work, so the test is skipped.
1379 fd, _ = tempfile.mkstemp()
1380 with socket.socket(family=42424, type=13331, fileno=fd) as s:
1381 self.assertEqual(s.family, 42424)
1382 self.assertEqual(s.type, 13331)
1285 1383
1286 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.') 1384 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.')
1287 class BasicCANTest(unittest.TestCase): 1385 class BasicCANTest(unittest.TestCase):
1288 1386
1289 def testCrucialConstants(self): 1387 def testCrucialConstants(self):
1290 socket.AF_CAN 1388 socket.AF_CAN
1291 socket.PF_CAN 1389 socket.PF_CAN
1292 socket.CAN_RAW 1390 socket.CAN_RAW
1293 1391
1392 @unittest.skipUnless(hasattr(socket, "CAN_BCM"),
1393 'socket.CAN_BCM required for this test.')
1394 def testBCMConstants(self):
1395 socket.CAN_BCM
1396
1397 # opcodes
1398 socket.CAN_BCM_TX_SETUP # create (cyclic) transmission task
1399 socket.CAN_BCM_TX_DELETE # remove (cyclic) transmission task
1400 socket.CAN_BCM_TX_READ # read properties of (cyclic) transmission t ask
1401 socket.CAN_BCM_TX_SEND # send one CAN frame
1402 socket.CAN_BCM_RX_SETUP # create RX content filter subscription
1403 socket.CAN_BCM_RX_DELETE # remove RX content filter subscription
1404 socket.CAN_BCM_RX_READ # read properties of RX content filter subsc ription
1405 socket.CAN_BCM_TX_STATUS # reply to TX_READ request
1406 socket.CAN_BCM_TX_EXPIRED # notification on performed transmissions (c ount=0)
1407 socket.CAN_BCM_RX_STATUS # reply to RX_READ request
1408 socket.CAN_BCM_RX_TIMEOUT # cyclic message is absent
1409 socket.CAN_BCM_RX_CHANGED # updated CAN frame (detected content change )
1410
1294 def testCreateSocket(self): 1411 def testCreateSocket(self):
1295 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s: 1412 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
1413 pass
1414
1415 @unittest.skipUnless(hasattr(socket, "CAN_BCM"),
1416 'socket.CAN_BCM required for this test.')
1417 def testCreateBCMSocket(self):
1418 with socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_BCM) as s:
1296 pass 1419 pass
1297 1420
1298 def testBindAny(self): 1421 def testBindAny(self):
1299 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s: 1422 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
1300 s.bind(('', )) 1423 s.bind(('', ))
1301 1424
1302 def testTooLongInterfaceName(self): 1425 def testTooLongInterfaceName(self):
1303 # most systems limit IFNAMSIZ to 16, take 1024 to be sure 1426 # most systems limit IFNAMSIZ to 16, take 1024 to be sure
1304 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s: 1427 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
1305 self.assertRaisesRegex(OSError, 'interface name too long', 1428 self.assertRaisesRegex(OSError, 'interface name too long',
(...skipping 14 matching lines...) Expand all
1320 def testFilter(self): 1443 def testFilter(self):
1321 can_id, can_mask = 0x200, 0x700 1444 can_id, can_mask = 0x200, 0x700
1322 can_filter = struct.pack("=II", can_id, can_mask) 1445 can_filter = struct.pack("=II", can_id, can_mask)
1323 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s: 1446 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
1324 s.setsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, can_filter) 1447 s.setsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, can_filter)
1325 self.assertEqual(can_filter, 1448 self.assertEqual(can_filter,
1326 s.getsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, 8)) 1449 s.getsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, 8))
1327 1450
1328 1451
1329 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.') 1452 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.')
1330 @unittest.skipUnless(thread, 'Threading required for this test.')
1331 class CANTest(ThreadedCANSocketTest): 1453 class CANTest(ThreadedCANSocketTest):
1332
1333 """The CAN frame structure is defined in <linux/can.h>:
1334
1335 struct can_frame {
1336 canid_t can_id; /* 32 bit CAN_ID + EFF/RTR/ERR flags */
1337 __u8 can_dlc; /* data length code: 0 .. 8 */
1338 __u8 data[8] __attribute__((aligned(8)));
1339 };
1340 """
1341 can_frame_fmt = "=IB3x8s"
1342 1454
1343 def __init__(self, methodName='runTest'): 1455 def __init__(self, methodName='runTest'):
1344 ThreadedCANSocketTest.__init__(self, methodName=methodName) 1456 ThreadedCANSocketTest.__init__(self, methodName=methodName)
1345 1457
1346 @classmethod 1458 @classmethod
1347 def build_can_frame(cls, can_id, data): 1459 def build_can_frame(cls, can_id, data):
1348 """Build a CAN frame.""" 1460 """Build a CAN frame."""
1349 can_dlc = len(data) 1461 can_dlc = len(data)
1350 data = data.ljust(8, b'\x00') 1462 data = data.ljust(8, b'\x00')
1351 return struct.pack(cls.can_frame_fmt, can_id, can_dlc, data) 1463 return struct.pack(cls.can_frame_fmt, can_id, can_dlc, data)
(...skipping 28 matching lines...) Expand all
1380 1492
1381 cf, addr = self.s.recvfrom(self.bufsize) 1493 cf, addr = self.s.recvfrom(self.bufsize)
1382 self.assertEqual(self.cf2, cf) 1494 self.assertEqual(self.cf2, cf)
1383 1495
1384 def _testSendMultiFrames(self): 1496 def _testSendMultiFrames(self):
1385 self.cf1 = self.build_can_frame(0x07, b'\x44\x33\x22\x11') 1497 self.cf1 = self.build_can_frame(0x07, b'\x44\x33\x22\x11')
1386 self.cli.send(self.cf1) 1498 self.cli.send(self.cf1)
1387 1499
1388 self.cf2 = self.build_can_frame(0x12, b'\x99\x22\x33') 1500 self.cf2 = self.build_can_frame(0x12, b'\x99\x22\x33')
1389 self.cli.send(self.cf2) 1501 self.cli.send(self.cf2)
1502
1503 @unittest.skipUnless(hasattr(socket, "CAN_BCM"),
1504 'socket.CAN_BCM required for this test.')
1505 def _testBCM(self):
1506 cf, addr = self.cli.recvfrom(self.bufsize)
1507 self.assertEqual(self.cf, cf)
1508 can_id, can_dlc, data = self.dissect_can_frame(cf)
1509 self.assertEqual(self.can_id, can_id)
1510 self.assertEqual(self.data, data)
1511
1512 @unittest.skipUnless(hasattr(socket, "CAN_BCM"),
1513 'socket.CAN_BCM required for this test.')
1514 def testBCM(self):
1515 bcm = socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_BCM)
1516 self.addCleanup(bcm.close)
1517 bcm.connect((self.interface,))
1518 self.can_id = 0x123
1519 self.data = bytes([0xc0, 0xff, 0xee])
1520 self.cf = self.build_can_frame(self.can_id, self.data)
1521 opcode = socket.CAN_BCM_TX_SEND
1522 flags = 0
1523 count = 0
1524 ival1_seconds = ival1_usec = ival2_seconds = ival2_usec = 0
1525 bcm_can_id = 0x0222
1526 nframes = 1
1527 assert len(self.cf) == 16
1528 header = struct.pack(self.bcm_cmd_msg_fmt,
1529 opcode,
1530 flags,
1531 count,
1532 ival1_seconds,
1533 ival1_usec,
1534 ival2_seconds,
1535 ival2_usec,
1536 bcm_can_id,
1537 nframes,
1538 )
1539 header_plus_frame = header + self.cf
1540 bytes_sent = bcm.send(header_plus_frame)
1541 self.assertEqual(bytes_sent, len(header_plus_frame))
1390 1542
1391 1543
1392 @unittest.skipUnless(HAVE_SOCKET_RDS, 'RDS sockets required for this test.') 1544 @unittest.skipUnless(HAVE_SOCKET_RDS, 'RDS sockets required for this test.')
1393 class BasicRDSTest(unittest.TestCase): 1545 class BasicRDSTest(unittest.TestCase):
1394 1546
1395 def testCrucialConstants(self): 1547 def testCrucialConstants(self):
1396 socket.AF_RDS 1548 socket.AF_RDS
1397 socket.PF_RDS 1549 socket.PF_RDS
1398 1550
1399 def testCreateSocket(self): 1551 def testCreateSocket(self):
(...skipping 180 matching lines...) Expand 10 before | Expand all | Expand 10 after
1580 def testShutdown(self): 1732 def testShutdown(self):
1581 # Testing shutdown() 1733 # Testing shutdown()
1582 msg = self.cli_conn.recv(1024) 1734 msg = self.cli_conn.recv(1024)
1583 self.assertEqual(msg, MSG) 1735 self.assertEqual(msg, MSG)
1584 # wait for _testShutdown to finish: on OS X, when the server 1736 # wait for _testShutdown to finish: on OS X, when the server
1585 # closes the connection the client also becomes disconnected, 1737 # closes the connection the client also becomes disconnected,
1586 # and the client's shutdown call will fail. (Issue #4397.) 1738 # and the client's shutdown call will fail. (Issue #4397.)
1587 self.done.wait() 1739 self.done.wait()
1588 1740
1589 def _testShutdown(self): 1741 def _testShutdown(self):
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
1590 self.serv_conn.send(MSG) 1750 self.serv_conn.send(MSG)
1591 # Issue 15989 1751 # Issue 15989
1592 self.assertRaises(OverflowError, self.serv_conn.shutdown, 1752 self.assertRaises(OverflowError, self.serv_conn.shutdown,
1593 _testcapi.INT_MAX + 1) 1753 _testcapi.INT_MAX + 1)
1594 self.assertRaises(OverflowError, self.serv_conn.shutdown, 1754 self.assertRaises(OverflowError, self.serv_conn.shutdown,
1595 2 + (_testcapi.UINT_MAX + 1)) 1755 2 + (_testcapi.UINT_MAX + 1))
1596 self.serv_conn.shutdown(2) 1756 self.serv_conn.shutdown(2)
1597 1757
1598 def testDetach(self): 1758 def testDetach(self):
1599 # Testing detach() 1759 # Testing detach()
(...skipping 749 matching lines...) Expand 10 before | Expand all | Expand 10 after
2349 def _testRecvmsgIntoScatter(self): 2509 def _testRecvmsgIntoScatter(self):
2350 self.sendToServer(b"Mary had a little lamb") 2510 self.sendToServer(b"Mary had a little lamb")
2351 2511
2352 2512
2353 class CmsgMacroTests(unittest.TestCase): 2513 class CmsgMacroTests(unittest.TestCase):
2354 # Test the functions CMSG_LEN() and CMSG_SPACE(). Tests 2514 # Test the functions CMSG_LEN() and CMSG_SPACE(). Tests
2355 # assumptions used by sendmsg() and recvmsg[_into](), which share 2515 # assumptions used by sendmsg() and recvmsg[_into](), which share
2356 # code with these functions. 2516 # code with these functions.
2357 2517
2358 # Match the definition in socketmodule.c 2518 # Match the definition in socketmodule.c
2359 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)
2360 2525
2361 @requireAttrs(socket, "CMSG_LEN") 2526 @requireAttrs(socket, "CMSG_LEN")
2362 def testCMSG_LEN(self): 2527 def testCMSG_LEN(self):
2363 # Test CMSG_LEN() with various valid and invalid values, 2528 # Test CMSG_LEN() with various valid and invalid values,
2364 # checking the assumptions used by recvmsg() and sendmsg(). 2529 # checking the assumptions used by recvmsg() and sendmsg().
2365 toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1 2530 toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1
2366 values = list(range(257)) + list(range(toobig - 257, toobig)) 2531 values = list(range(257)) + list(range(toobig - 257, toobig))
2367 2532
2368 # 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
2369 self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2) 2534 self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2)
(...skipping 152 matching lines...) Expand 10 before | Expand all | Expand 10 after
2522 self.doRecvmsg(self.serv_sock, len(MSG), 2687 self.doRecvmsg(self.serv_sock, len(MSG),
2523 socket.CMSG_LEN(4 * SIZEOF_INT)), 2688 socket.CMSG_LEN(4 * SIZEOF_INT)),
2524 # RFC 3542 says implementations may set 2689 # RFC 3542 says implementations may set
2525 # MSG_CTRUNC if there isn't enough space 2690 # MSG_CTRUNC if there isn't enough space
2526 # for trailing padding. 2691 # for trailing padding.
2527 ignoreflags=socket.MSG_CTRUNC) 2692 ignoreflags=socket.MSG_CTRUNC)
2528 2693
2529 def _testFDPassCMSG_LEN(self): 2694 def _testFDPassCMSG_LEN(self):
2530 self.createAndSendFDs(1) 2695 self.createAndSendFDs(1)
2531 2696
2532 # Issue #12958: The following test has problems on Mac OS X 2697 @unittest.skipIf(sys.platform == "darwin", "skipping, see issue #12958")
2533 @support.anticipate_failure(sys.platform == "darwin")
2534 @requireAttrs(socket, "CMSG_SPACE") 2698 @requireAttrs(socket, "CMSG_SPACE")
2535 def testFDPassSeparate(self): 2699 def testFDPassSeparate(self):
2536 # Pass two FDs in two separate arrays. Arrays may be combined 2700 # Pass two FDs in two separate arrays. Arrays may be combined
2537 # into a single control message by the OS. 2701 # into a single control message by the OS.
2538 self.checkRecvmsgFDs(2, 2702 self.checkRecvmsgFDs(2,
2539 self.doRecvmsg(self.serv_sock, len(MSG), 10240), 2703 self.doRecvmsg(self.serv_sock, len(MSG), 10240),
2540 maxcmsgs=2) 2704 maxcmsgs=2)
2541 2705
2542 @testFDPassSeparate.client_skip 2706 @testFDPassSeparate.client_skip
2543 @support.anticipate_failure(sys.platform == "darwin") 2707 @unittest.skipIf(sys.platform == "darwin", "skipping, see issue #12958")
2544 def _testFDPassSeparate(self): 2708 def _testFDPassSeparate(self):
2545 fd0, fd1 = self.newFDs(2) 2709 fd0, fd1 = self.newFDs(2)
2546 self.assertEqual( 2710 self.assertEqual(
2547 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET, 2711 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET,
2548 socket.SCM_RIGHTS, 2712 socket.SCM_RIGHTS,
2549 array.array("i", [fd0])), 2713 array.array("i", [fd0])),
2550 (socket.SOL_SOCKET, 2714 (socket.SOL_SOCKET,
2551 socket.SCM_RIGHTS, 2715 socket.SCM_RIGHTS,
2552 array.array("i", [fd1]))]), 2716 array.array("i", [fd1]))]),
2553 len(MSG)) 2717 len(MSG))
2554 2718
2555 # Issue #12958: The following test has problems on Mac OS X 2719 @unittest.skipIf(sys.platform == "darwin", "skipping, see issue #12958")
2556 @support.anticipate_failure(sys.platform == "darwin")
2557 @requireAttrs(socket, "CMSG_SPACE") 2720 @requireAttrs(socket, "CMSG_SPACE")
2558 def testFDPassSeparateMinSpace(self): 2721 def testFDPassSeparateMinSpace(self):
2559 # Pass two FDs in two separate arrays, receiving them into the 2722 # Pass two FDs in two separate arrays, receiving them into the
2560 # minimum space for two arrays. 2723 # minimum space for two arrays.
2561 self.checkRecvmsgFDs(2, 2724 self.checkRecvmsgFDs(2,
2562 self.doRecvmsg(self.serv_sock, len(MSG), 2725 self.doRecvmsg(self.serv_sock, len(MSG),
2563 socket.CMSG_SPACE(SIZEOF_INT) + 2726 socket.CMSG_SPACE(SIZEOF_INT) +
2564 socket.CMSG_LEN(SIZEOF_INT)), 2727 socket.CMSG_LEN(SIZEOF_INT)),
2565 maxcmsgs=2, ignoreflags=socket.MSG_CTRUNC) 2728 maxcmsgs=2, ignoreflags=socket.MSG_CTRUNC)
2566 2729
2567 @testFDPassSeparateMinSpace.client_skip 2730 @testFDPassSeparateMinSpace.client_skip
2568 @support.anticipate_failure(sys.platform == "darwin") 2731 @unittest.skipIf(sys.platform == "darwin", "skipping, see issue #12958")
2569 def _testFDPassSeparateMinSpace(self): 2732 def _testFDPassSeparateMinSpace(self):
2570 fd0, fd1 = self.newFDs(2) 2733 fd0, fd1 = self.newFDs(2)
2571 self.assertEqual( 2734 self.assertEqual(
2572 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET, 2735 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET,
2573 socket.SCM_RIGHTS, 2736 socket.SCM_RIGHTS,
2574 array.array("i", [fd0])), 2737 array.array("i", [fd0])),
2575 (socket.SOL_SOCKET, 2738 (socket.SOL_SOCKET,
2576 socket.SCM_RIGHTS, 2739 socket.SCM_RIGHTS,
2577 array.array("i", [fd1]))]), 2740 array.array("i", [fd1]))]),
2578 len(MSG)) 2741 len(MSG))
(...skipping 635 matching lines...) Expand 10 before | Expand all | Expand 10 after
3214 3377
3215 @requireAttrs(socket.socket, "recvmsg_into") 3378 @requireAttrs(socket.socket, "recvmsg_into")
3216 @unittest.skipUnless(thread, 'Threading required for this test.') 3379 @unittest.skipUnless(thread, 'Threading required for this test.')
3217 class RecvmsgIntoUDPTest(RecvmsgIntoTests, SendrecvmsgUDPTestBase): 3380 class RecvmsgIntoUDPTest(RecvmsgIntoTests, SendrecvmsgUDPTestBase):
3218 pass 3381 pass
3219 3382
3220 3383
3221 class SendrecvmsgUDP6TestBase(SendrecvmsgDgramFlagsBase, 3384 class SendrecvmsgUDP6TestBase(SendrecvmsgDgramFlagsBase,
3222 SendrecvmsgConnectionlessBase, 3385 SendrecvmsgConnectionlessBase,
3223 ThreadedSocketTestMixin, UDP6TestBase): 3386 ThreadedSocketTestMixin, UDP6TestBase):
3224 pass 3387
3388 def checkRecvmsgAddress(self, addr1, addr2):
3389 # Called to compare the received address with the address of
3390 # the peer, ignoring scope ID
3391 self.assertEqual(addr1[:-1], addr2[:-1])
3225 3392
3226 @requireAttrs(socket.socket, "sendmsg") 3393 @requireAttrs(socket.socket, "sendmsg")
3227 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.') 3394 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.')
3228 @requireSocket("AF_INET6", "SOCK_DGRAM") 3395 @requireSocket("AF_INET6", "SOCK_DGRAM")
3229 @unittest.skipUnless(thread, 'Threading required for this test.') 3396 @unittest.skipUnless(thread, 'Threading required for this test.')
3230 class SendmsgUDP6Test(SendmsgConnectionlessTests, SendrecvmsgUDP6TestBase): 3397 class SendmsgUDP6Test(SendmsgConnectionlessTests, SendrecvmsgUDP6TestBase):
3231 pass 3398 pass
3232 3399
3233 @requireAttrs(socket.socket, "recvmsg") 3400 @requireAttrs(socket.socket, "recvmsg")
3234 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.') 3401 @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 required for this test.')
(...skipping 241 matching lines...) Expand 10 before | Expand all | Expand 10 after
3476 # Check that func(*args, **kwargs), run in a loop, raises 3643 # Check that func(*args, **kwargs), run in a loop, raises
3477 # OSError with an errno of EINTR when interrupted by a 3644 # OSError with an errno of EINTR when interrupted by a
3478 # signal. 3645 # signal.
3479 with self.assertRaises(OSError) as cm: 3646 with self.assertRaises(OSError) as cm:
3480 while True: 3647 while True:
3481 self.setAlarm(self.alarm_time) 3648 self.setAlarm(self.alarm_time)
3482 func(*args, **kwargs) 3649 func(*args, **kwargs)
3483 self.assertNotIsInstance(cm.exception, socket.timeout) 3650 self.assertNotIsInstance(cm.exception, socket.timeout)
3484 self.assertEqual(cm.exception.errno, errno.EINTR) 3651 self.assertEqual(cm.exception.errno, errno.EINTR)
3485 3652
3486 # 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
3487 @support.anticipate_failure(sys.platform == "darwin") 3654 @support.requires_mac_ver(10, 7)
3488 def testInterruptedSendTimeout(self): 3655 def testInterruptedSendTimeout(self):
3489 self.checkInterruptedSend(self.serv_conn.send, b"a"*512) 3656 self.checkInterruptedSend(self.serv_conn.send, b"a"*512)
3490 3657
3491 @support.anticipate_failure(sys.platform == "darwin") 3658 @support.requires_mac_ver(10, 7)
3492 def testInterruptedSendtoTimeout(self): 3659 def testInterruptedSendtoTimeout(self):
3493 # Passing an actual address here as Python's wrapper for 3660 # Passing an actual address here as Python's wrapper for
3494 # sendto() doesn't allow passing a zero-length one; POSIX 3661 # sendto() doesn't allow passing a zero-length one; POSIX
3495 # requires that the address is ignored since the socket is 3662 # requires that the address is ignored since the socket is
3496 # connection-mode, however. 3663 # connection-mode, however.
3497 self.checkInterruptedSend(self.serv_conn.sendto, b"a"*512, 3664 self.checkInterruptedSend(self.serv_conn.sendto, b"a"*512,
3498 self.serv_addr) 3665 self.serv_addr)
3499 3666
3500 @support.anticipate_failure(sys.platform == "darwin") 3667 @support.requires_mac_ver(10, 7)
3501 @requireAttrs(socket.socket, "sendmsg") 3668 @requireAttrs(socket.socket, "sendmsg")
3502 def testInterruptedSendmsgTimeout(self): 3669 def testInterruptedSendmsgTimeout(self):
3503 self.checkInterruptedSend(self.serv_conn.sendmsg, [b"a"*512]) 3670 self.checkInterruptedSend(self.serv_conn.sendmsg, [b"a"*512])
3504 3671
3505 3672
3506 @unittest.skipUnless(thread, 'Threading required for this test.') 3673 @unittest.skipUnless(thread, 'Threading required for this test.')
3507 class TCPCloserTest(ThreadedTCPSocketTest): 3674 class TCPCloserTest(ThreadedTCPSocketTest):
3508 3675
3509 def testClose(self): 3676 def testClose(self):
3510 conn, addr = self.serv.accept() 3677 conn, addr = self.serv.accept()
3511 conn.close() 3678 conn.close()
3512 3679
3513 sd = self.cli 3680 sd = self.cli
3514 read, write, err = select.select([sd], [], [], 1.0) 3681 read, write, err = select.select([sd], [], [], 1.0)
3515 self.assertEqual(read, [sd]) 3682 self.assertEqual(read, [sd])
3516 self.assertEqual(sd.recv(1), b'') 3683 self.assertEqual(sd.recv(1), b'')
3517 3684
3518 # Calling close() many times should be safe. 3685 # Calling close() many times should be safe.
3519 conn.close() 3686 conn.close()
3520 conn.close() 3687 conn.close()
3521 3688
3522 def _testClose(self): 3689 def _testClose(self):
3523 self.cli.connect((HOST, self.port)) 3690 self.cli.connect((HOST, self.port))
3524 time.sleep(1.0) 3691 time.sleep(1.0)
3525 3692
3693 @unittest.skipUnless(hasattr(socket, 'socketpair'),
3694 'test needs socket.socketpair()')
3526 @unittest.skipUnless(thread, 'Threading required for this test.') 3695 @unittest.skipUnless(thread, 'Threading required for this test.')
3527 class BasicSocketPairTest(SocketPairTest): 3696 class BasicSocketPairTest(SocketPairTest):
3528 3697
3529 def __init__(self, methodName='runTest'): 3698 def __init__(self, methodName='runTest'):
3530 SocketPairTest.__init__(self, methodName=methodName) 3699 SocketPairTest.__init__(self, methodName=methodName)
3531 3700
3532 def _check_defaults(self, sock): 3701 def _check_defaults(self, sock):
3533 self.assertIsInstance(sock, socket.socket) 3702 self.assertIsInstance(sock, socket.socket)
3534 if hasattr(socket, 'AF_UNIX'): 3703 if hasattr(socket, 'AF_UNIX'):
3535 self.assertEqual(sock.family, socket.AF_UNIX) 3704 self.assertEqual(sock.family, socket.AF_UNIX)
(...skipping 34 matching lines...) Expand 10 before | Expand all | Expand 10 after
3570 self.assertIsNone(self.serv.gettimeout()) 3739 self.assertIsNone(self.serv.gettimeout())
3571 self.serv.setblocking(False) 3740 self.serv.setblocking(False)
3572 self.assertEqual(self.serv.gettimeout(), 0.0) 3741 self.assertEqual(self.serv.gettimeout(), 0.0)
3573 start = time.time() 3742 start = time.time()
3574 try: 3743 try:
3575 self.serv.accept() 3744 self.serv.accept()
3576 except OSError: 3745 except OSError:
3577 pass 3746 pass
3578 end = time.time() 3747 end = time.time()
3579 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.") 3748 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.")
3580 # Issue 15989
3581 if _testcapi.UINT_MAX < _testcapi.ULONG_MAX:
3582 self.serv.setblocking(_testcapi.UINT_MAX + 1)
3583 self.assertIsNone(self.serv.gettimeout())
3584 3749
3585 def _testSetBlocking(self): 3750 def _testSetBlocking(self):
3586 pass 3751 pass
3587 3752
3588 if hasattr(socket, "SOCK_NONBLOCK"): 3753 @support.cpython_only
3589 @support.requires_linux_version(2, 6, 28) 3754 def testSetBlocking_overflow(self):
3590 def testInitNonBlocking(self): 3755 # Issue 15989
3591 # reinit server socket 3756 import _testcapi
3592 self.serv.close() 3757 if _testcapi.UINT_MAX >= _testcapi.ULONG_MAX:
3593 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM | 3758 self.skipTest('needs UINT_MAX < ULONG_MAX')
3594 socket.SOCK_NONBLOCK) 3759 self.serv.setblocking(False)
3595 self.port = support.bind_port(self.serv) 3760 self.assertEqual(self.serv.gettimeout(), 0.0)
3596 self.serv.listen(1) 3761 self.serv.setblocking(_testcapi.UINT_MAX + 1)
3597 # actual testing 3762 self.assertIsNone(self.serv.gettimeout())
3598 start = time.time() 3763
3599 try: 3764 _testSetBlocking_overflow = support.cpython_only(_testSetBlocking)
3600 self.serv.accept() 3765
3601 except OSError: 3766 @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
3602 pass 3767 'test needs socket.SOCK_NONBLOCK')
3603 end = time.time() 3768 @support.requires_linux_version(2, 6, 28)
3604 self.assertLess(end - start, 1.0, "Error creating with non-blocking mode.") 3769 def testInitNonBlocking(self):
3605 3770 # reinit server socket
3606 def _testInitNonBlocking(self): 3771 self.serv.close()
3772 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM |
3773 socket.SOCK_NONBLOCK)
3774 self.port = support.bind_port(self.serv)
3775 self.serv.listen(1)
3776 # actual testing
3777 start = time.time()
3778 try:
3779 self.serv.accept()
3780 except OSError:
3607 pass 3781 pass
3782 end = time.time()
3783 self.assertLess(end - start, 1.0, "Error creating with non-blocking mode .")
3784
3785 def _testInitNonBlocking(self):
3786 pass
3608 3787
3609 def testInheritFlags(self): 3788 def testInheritFlags(self):
3610 # Issue #7995: when calling accept() on a listening socket with a 3789 # Issue #7995: when calling accept() on a listening socket with a
3611 # timeout, the resulting socket should not be non-blocking. 3790 # timeout, the resulting socket should not be non-blocking.
3612 self.serv.settimeout(10) 3791 self.serv.settimeout(10)
3613 try: 3792 try:
3614 conn, addr = self.serv.accept() 3793 conn, addr = self.serv.accept()
3615 message = conn.recv(len(MSG)) 3794 message = conn.recv(len(MSG))
3616 finally: 3795 finally:
3617 conn.close() 3796 conn.close()
(...skipping 191 matching lines...) Expand 10 before | Expand all | Expand 10 after
3809 msg = self.cli_conn.recv(len(MSG)) 3988 msg = self.cli_conn.recv(len(MSG))
3810 if isinstance(self.read_msg, str): 3989 if isinstance(self.read_msg, str):
3811 msg = msg.decode() 3990 msg = msg.decode()
3812 self.assertEqual(msg, self.read_msg) 3991 self.assertEqual(msg, self.read_msg)
3813 3992
3814 def _testMakefileAfterMakefileClose(self): 3993 def _testMakefileAfterMakefileClose(self):
3815 self.write_file.write(self.write_msg) 3994 self.write_file.write(self.write_msg)
3816 self.write_file.flush() 3995 self.write_file.flush()
3817 3996
3818 def testClosedAttr(self): 3997 def testClosedAttr(self):
3819 self.assertFalse(self.read_file.closed) 3998 self.assertTrue(not self.read_file.closed)
3820 3999
3821 def _testClosedAttr(self): 4000 def _testClosedAttr(self):
3822 self.assertFalse(self.write_file.closed) 4001 self.assertTrue(not self.write_file.closed)
3823 4002
3824 def testAttributes(self): 4003 def testAttributes(self):
3825 self.assertEqual(self.read_file.mode, self.read_mode) 4004 self.assertEqual(self.read_file.mode, self.read_mode)
3826 self.assertEqual(self.read_file.name, self.cli_conn.fileno()) 4005 self.assertEqual(self.read_file.name, self.cli_conn.fileno())
3827 4006
3828 def _testAttributes(self): 4007 def _testAttributes(self):
3829 self.assertEqual(self.write_file.mode, self.write_mode) 4008 self.assertEqual(self.write_file.mode, self.write_mode)
3830 self.assertEqual(self.write_file.name, self.serv_conn.fileno()) 4009 self.assertEqual(self.write_file.name, self.serv_conn.fileno())
3831 4010
3832 def testRealClose(self): 4011 def testRealClose(self):
(...skipping 200 matching lines...) Expand 10 before | Expand all | Expand 10 after
4033 self.cli_finished.wait(5.0) 4212 self.cli_finished.wait(5.0)
4034 # The client thread can't skip directly - the SkipTest exception 4213 # The client thread can't skip directly - the SkipTest exception
4035 # would appear as a failure. 4214 # would appear as a failure.
4036 if self.serv_skipped: 4215 if self.serv_skipped:
4037 self.skipTest(self.serv_skipped) 4216 self.skipTest(self.serv_skipped)
4038 4217
4039 def _testWriteNonBlocking(self): 4218 def _testWriteNonBlocking(self):
4040 self.serv_skipped = None 4219 self.serv_skipped = None
4041 self.serv_conn.setblocking(False) 4220 self.serv_conn.setblocking(False)
4042 # Try to saturate the socket buffer pipe with repeated large writes. 4221 # Try to saturate the socket buffer pipe with repeated large writes.
4043 BIG = b"x" * (1024 ** 2) 4222 BIG = b"x" * support.SOCK_MAX_SIZE
4044 LIMIT = 10 4223 LIMIT = 10
4045 # The first write() succeeds since a chunk of data can be buffered 4224 # The first write() succeeds since a chunk of data can be buffered
4046 n = self.write_file.write(BIG) 4225 n = self.write_file.write(BIG)
4047 self.assertGreater(n, 0) 4226 self.assertGreater(n, 0)
4048 for i in range(LIMIT): 4227 for i in range(LIMIT):
4049 n = self.write_file.write(BIG) 4228 n = self.write_file.write(BIG)
4050 if n is None: 4229 if n is None:
4051 # Succeeded 4230 # Succeeded
4052 break 4231 break
4053 self.assertGreater(n, 0) 4232 self.assertGreater(n, 0)
(...skipping 233 matching lines...) Expand 10 before | Expand all | Expand 10 after
4287 foo = self.serv.accept() 4466 foo = self.serv.accept()
4288 except socket.timeout: 4467 except socket.timeout:
4289 self.fail("caught timeout instead of error (TCP)") 4468 self.fail("caught timeout instead of error (TCP)")
4290 except OSError: 4469 except OSError:
4291 ok = True 4470 ok = True
4292 except: 4471 except:
4293 self.fail("caught unexpected exception (TCP)") 4472 self.fail("caught unexpected exception (TCP)")
4294 if not ok: 4473 if not ok:
4295 self.fail("accept() returned success when we did not expect it") 4474 self.fail("accept() returned success when we did not expect it")
4296 4475
4476 @unittest.skipUnless(hasattr(signal, 'alarm'),
4477 'test needs signal.alarm()')
4297 def testInterruptedTimeout(self): 4478 def testInterruptedTimeout(self):
4298 # XXX I don't know how to do this test on MSWindows or any other 4479 # XXX I don't know how to do this test on MSWindows or any other
4299 # plaform that doesn't support signal.alarm() or os.kill(), though 4480 # plaform that doesn't support signal.alarm() or os.kill(), though
4300 # the bug should have existed on all platforms. 4481 # the bug should have existed on all platforms.
4301 if not hasattr(signal, "alarm"):
4302 return # can only test on *nix
4303 self.serv.settimeout(5.0) # must be longer than alarm 4482 self.serv.settimeout(5.0) # must be longer than alarm
4304 class Alarm(Exception): 4483 class Alarm(Exception):
4305 pass 4484 pass
4306 def alarm_handler(signal, frame): 4485 def alarm_handler(signal, frame):
4307 raise Alarm 4486 raise Alarm
4308 old_alarm = signal.signal(signal.SIGALRM, alarm_handler) 4487 old_alarm = signal.signal(signal.SIGALRM, alarm_handler)
4309 try: 4488 try:
4310 signal.alarm(2) # POSIX allows alarm to be up to 1 second early 4489 signal.alarm(2) # POSIX allows alarm to be up to 1 second early
4311 try: 4490 try:
4312 foo = self.serv.accept() 4491 foo = self.serv.accept()
(...skipping 39 matching lines...) Expand 10 before | Expand all | Expand 10 after
4352 self.fail("recv() returned success when we did not expect it") 4531 self.fail("recv() returned success when we did not expect it")
4353 4532
4354 class TestExceptions(unittest.TestCase): 4533 class TestExceptions(unittest.TestCase):
4355 4534
4356 def testExceptionTree(self): 4535 def testExceptionTree(self):
4357 self.assertTrue(issubclass(OSError, Exception)) 4536 self.assertTrue(issubclass(OSError, Exception))
4358 self.assertTrue(issubclass(socket.herror, OSError)) 4537 self.assertTrue(issubclass(socket.herror, OSError))
4359 self.assertTrue(issubclass(socket.gaierror, OSError)) 4538 self.assertTrue(issubclass(socket.gaierror, OSError))
4360 self.assertTrue(issubclass(socket.timeout, OSError)) 4539 self.assertTrue(issubclass(socket.timeout, OSError))
4361 4540
4541 @unittest.skipUnless(sys.platform == 'linux', 'Linux specific test')
4362 class TestLinuxAbstractNamespace(unittest.TestCase): 4542 class TestLinuxAbstractNamespace(unittest.TestCase):
4363 4543
4364 UNIX_PATH_MAX = 108 4544 UNIX_PATH_MAX = 108
4365 4545
4366 def testLinuxAbstractNamespace(self): 4546 def testLinuxAbstractNamespace(self):
4367 address = b"\x00python-test-hello\x00\xff" 4547 address = b"\x00python-test-hello\x00\xff"
4368 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s1: 4548 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s1:
4369 s1.bind(address) 4549 s1.bind(address)
4370 s1.listen(1) 4550 s1.listen(1)
4371 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s2: 4551 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s2:
(...skipping 15 matching lines...) Expand all
4387 4567
4388 def testStrName(self): 4568 def testStrName(self):
4389 # Check that an abstract name can be passed as a string. 4569 # Check that an abstract name can be passed as a string.
4390 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 4570 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
4391 try: 4571 try:
4392 s.bind("\x00python\x00test\x00") 4572 s.bind("\x00python\x00test\x00")
4393 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00") 4573 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00")
4394 finally: 4574 finally:
4395 s.close() 4575 s.close()
4396 4576
4577 @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'test needs socket.AF_UNIX')
4397 class TestUnixDomain(unittest.TestCase): 4578 class TestUnixDomain(unittest.TestCase):
4398 4579
4399 def setUp(self): 4580 def setUp(self):
4400 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 4581 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
4401 4582
4402 def tearDown(self): 4583 def tearDown(self):
4403 self.sock.close() 4584 self.sock.close()
4404 4585
4405 def encoded(self, path): 4586 def encoded(self, path):
4406 # Return the given path encoded in the file system encoding, 4587 # Return the given path encoded in the file system encoding,
(...skipping 110 matching lines...) Expand 10 before | Expand all | Expand 10 after
4517 4698
4518 def testRecvFromIntoMemoryview(self): 4699 def testRecvFromIntoMemoryview(self):
4519 buf = bytearray(1024) 4700 buf = bytearray(1024)
4520 nbytes, addr = self.cli_conn.recvfrom_into(memoryview(buf)) 4701 nbytes, addr = self.cli_conn.recvfrom_into(memoryview(buf))
4521 self.assertEqual(nbytes, len(MSG)) 4702 self.assertEqual(nbytes, len(MSG))
4522 msg = buf[:len(MSG)] 4703 msg = buf[:len(MSG)]
4523 self.assertEqual(msg, MSG) 4704 self.assertEqual(msg, MSG)
4524 4705
4525 _testRecvFromIntoMemoryview = _testRecvFromIntoArray 4706 _testRecvFromIntoMemoryview = _testRecvFromIntoArray
4526 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
4722
4527 4723
4528 TIPC_STYPE = 2000 4724 TIPC_STYPE = 2000
4529 TIPC_LOWER = 200 4725 TIPC_LOWER = 200
4530 TIPC_UPPER = 210 4726 TIPC_UPPER = 210
4531 4727
4532 def isTipcAvailable(): 4728 def isTipcAvailable():
4533 """Check if the TIPC module is loaded 4729 """Check if the TIPC module is loaded
4534 4730
4535 The TIPC module is not loaded automatically on Ubuntu and probably 4731 The TIPC module is not loaded automatically on Ubuntu and probably
4536 other Linux distros. 4732 other Linux distros.
4537 """ 4733 """
4538 if not hasattr(socket, "AF_TIPC"): 4734 if not hasattr(socket, "AF_TIPC"):
4539 return False 4735 return False
4540 if not os.path.isfile("/proc/modules"): 4736 if not os.path.isfile("/proc/modules"):
4541 return False 4737 return False
4542 with open("/proc/modules") as f: 4738 with open("/proc/modules") as f:
4543 for line in f: 4739 for line in f:
4544 if line.startswith("tipc "): 4740 if line.startswith("tipc "):
4545 return True 4741 return True
4546 if support.verbose:
4547 print("TIPC module is not loaded, please 'sudo modprobe tipc'")
4548 return False 4742 return False
4549 4743
4744 @unittest.skipUnless(isTipcAvailable(),
4745 "TIPC module is not loaded, please 'sudo modprobe tipc'")
4550 class TIPCTest(unittest.TestCase): 4746 class TIPCTest(unittest.TestCase):
4551 def testRDM(self): 4747 def testRDM(self):
4552 srv = socket.socket(socket.AF_TIPC, socket.SOCK_RDM) 4748 srv = socket.socket(socket.AF_TIPC, socket.SOCK_RDM)
4553 cli = socket.socket(socket.AF_TIPC, socket.SOCK_RDM) 4749 cli = socket.socket(socket.AF_TIPC, socket.SOCK_RDM)
4554 self.addCleanup(srv.close) 4750 self.addCleanup(srv.close)
4555 self.addCleanup(cli.close) 4751 self.addCleanup(cli.close)
4556 4752
4557 srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 4753 srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
4558 srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE, 4754 srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE,
4559 TIPC_LOWER, TIPC_UPPER) 4755 TIPC_LOWER, TIPC_UPPER)
4560 srv.bind(srvaddr) 4756 srv.bind(srvaddr)
4561 4757
4562 sendaddr = (socket.TIPC_ADDR_NAME, TIPC_STYPE, 4758 sendaddr = (socket.TIPC_ADDR_NAME, TIPC_STYPE,
4563 TIPC_LOWER + int((TIPC_UPPER - TIPC_LOWER) / 2), 0) 4759 TIPC_LOWER + int((TIPC_UPPER - TIPC_LOWER) / 2), 0)
4564 cli.sendto(MSG, sendaddr) 4760 cli.sendto(MSG, sendaddr)
4565 4761
4566 msg, recvaddr = srv.recvfrom(1024) 4762 msg, recvaddr = srv.recvfrom(1024)
4567 4763
4568 self.assertEqual(cli.getsockname(), recvaddr) 4764 self.assertEqual(cli.getsockname(), recvaddr)
4569 self.assertEqual(msg, MSG) 4765 self.assertEqual(msg, MSG)
4570 4766
4571 4767
4768 @unittest.skipUnless(isTipcAvailable(),
4769 "TIPC module is not loaded, please 'sudo modprobe tipc'")
4572 class TIPCThreadableTest(unittest.TestCase, ThreadableTest): 4770 class TIPCThreadableTest(unittest.TestCase, ThreadableTest):
4573 def __init__(self, methodName = 'runTest'): 4771 def __init__(self, methodName = 'runTest'):
4574 unittest.TestCase.__init__(self, methodName = methodName) 4772 unittest.TestCase.__init__(self, methodName = methodName)
4575 ThreadableTest.__init__(self) 4773 ThreadableTest.__init__(self)
4576 4774
4577 def setUp(self): 4775 def setUp(self):
4578 self.srv = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM) 4776 self.srv = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM)
4579 self.addCleanup(self.srv.close) 4777 self.addCleanup(self.srv.close)
4580 self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 4778 self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
4581 srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE, 4779 srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE,
(...skipping 64 matching lines...) Expand 10 before | Expand all | Expand 10 after
4646 conn.sendall(data) 4844 conn.sendall(data)
4647 4845
4648 def _testCreateConnectionClose(self): 4846 def _testCreateConnectionClose(self):
4649 address = self.serv.getsockname() 4847 address = self.serv.getsockname()
4650 with socket.create_connection(address) as sock: 4848 with socket.create_connection(address) as sock:
4651 sock.close() 4849 sock.close()
4652 self.assertTrue(sock._closed) 4850 self.assertTrue(sock._closed)
4653 self.assertRaises(OSError, sock.sendall, b'foo') 4851 self.assertRaises(OSError, sock.sendall, b'foo')
4654 4852
4655 4853
4656 @unittest.skipUnless(hasattr(socket, "SOCK_CLOEXEC"), 4854 class InheritanceTest(unittest.TestCase):
4657 "SOCK_CLOEXEC not defined") 4855 @unittest.skipUnless(hasattr(socket, "SOCK_CLOEXEC"),
4658 @unittest.skipUnless(fcntl, "module fcntl not available") 4856 "SOCK_CLOEXEC not defined")
4659 class CloexecConstantTest(unittest.TestCase):
4660 @support.requires_linux_version(2, 6, 28) 4857 @support.requires_linux_version(2, 6, 28)
4661 def test_SOCK_CLOEXEC(self): 4858 def test_SOCK_CLOEXEC(self):
4662 with socket.socket(socket.AF_INET, 4859 with socket.socket(socket.AF_INET,
4663 socket.SOCK_STREAM | socket.SOCK_CLOEXEC) as s: 4860 socket.SOCK_STREAM | socket.SOCK_CLOEXEC) as s:
4664 self.assertTrue(s.type & socket.SOCK_CLOEXEC) 4861 self.assertTrue(s.type & socket.SOCK_CLOEXEC)
4665 self.assertTrue(fcntl.fcntl(s, fcntl.F_GETFD) & fcntl.FD_CLOEXEC) 4862 self.assertFalse(s.get_inheritable())
4863
4864 def test_default_inheritable(self):
4865 sock = socket.socket()
4866 with sock:
4867 self.assertEqual(sock.get_inheritable(), False)
4868
4869 def test_dup(self):
4870 sock = socket.socket()
4871 with sock:
4872 newsock = sock.dup()
4873 sock.close()
4874 with newsock:
4875 self.assertEqual(newsock.get_inheritable(), False)
4876
4877 def test_set_inheritable(self):
4878 sock = socket.socket()
4879 with sock:
4880 sock.set_inheritable(True)
4881 self.assertEqual(sock.get_inheritable(), True)
4882
4883 sock.set_inheritable(False)
4884 self.assertEqual(sock.get_inheritable(), False)
4885
4886 @unittest.skipIf(fcntl is None, "need fcntl")
4887 def test_get_inheritable_cloexec(self):
4888 sock = socket.socket()
4889 with sock:
4890 fd = sock.fileno()
4891 self.assertEqual(sock.get_inheritable(), False)
4892
4893 # clear FD_CLOEXEC flag
4894 flags = fcntl.fcntl(fd, fcntl.F_GETFD)
4895 flags &= ~fcntl.FD_CLOEXEC
4896 fcntl.fcntl(fd, fcntl.F_SETFD, flags)
4897
4898 self.assertEqual(sock.get_inheritable(), True)
4899
4900 @unittest.skipIf(fcntl is None, "need fcntl")
4901 def test_set_inheritable_cloexec(self):
4902 sock = socket.socket()
4903 with sock:
4904 fd = sock.fileno()
4905 self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
4906 fcntl.FD_CLOEXEC)
4907
4908 sock.set_inheritable(True)
4909 self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
4910 0)
4911
4912
4913 @unittest.skipUnless(hasattr(socket, "socketpair"),
4914 "need socket.socketpair()")
4915 def test_socketpair(self):
4916 s1, s2 = socket.socketpair()
4917 self.addCleanup(s1.close)
4918 self.addCleanup(s2.close)
4919 self.assertEqual(s1.get_inheritable(), False)
4920 self.assertEqual(s2.get_inheritable(), False)
4666 4921
4667 4922
4668 @unittest.skipUnless(hasattr(socket, "SOCK_NONBLOCK"), 4923 @unittest.skipUnless(hasattr(socket, "SOCK_NONBLOCK"),
4669 "SOCK_NONBLOCK not defined") 4924 "SOCK_NONBLOCK not defined")
4670 class NonblockConstantTest(unittest.TestCase): 4925 class NonblockConstantTest(unittest.TestCase):
4671 def checkNonblock(self, s, nonblock=True, timeout=0.0): 4926 def checkNonblock(self, s, nonblock=True, timeout=0.0):
4672 if nonblock: 4927 if nonblock:
4673 self.assertTrue(s.type & socket.SOCK_NONBLOCK) 4928 self.assertTrue(s.type & socket.SOCK_NONBLOCK)
4674 self.assertEqual(s.gettimeout(), timeout) 4929 self.assertEqual(s.gettimeout(), timeout)
4675 else: 4930 else:
(...skipping 148 matching lines...) Expand 10 before | Expand all | Expand 10 after
4824 UnbufferedFileObjectClassTestCase, 5079 UnbufferedFileObjectClassTestCase,
4825 LineBufferedFileObjectClassTestCase, 5080 LineBufferedFileObjectClassTestCase,
4826 SmallBufferedFileObjectClassTestCase, 5081 SmallBufferedFileObjectClassTestCase,
4827 UnicodeReadFileObjectClassTestCase, 5082 UnicodeReadFileObjectClassTestCase,
4828 UnicodeWriteFileObjectClassTestCase, 5083 UnicodeWriteFileObjectClassTestCase,
4829 UnicodeReadWriteFileObjectClassTestCase, 5084 UnicodeReadWriteFileObjectClassTestCase,
4830 NetworkConnectionNoServer, 5085 NetworkConnectionNoServer,
4831 NetworkConnectionAttributesTest, 5086 NetworkConnectionAttributesTest,
4832 NetworkConnectionBehaviourTest, 5087 NetworkConnectionBehaviourTest,
4833 ContextManagersTest, 5088 ContextManagersTest,
4834 CloexecConstantTest, 5089 InheritanceTest,
4835 NonblockConstantTest 5090 NonblockConstantTest
4836 ]) 5091 ])
4837 if hasattr(socket, "socketpair"): 5092 tests.append(BasicSocketPairTest)
4838 tests.append(BasicSocketPairTest) 5093 tests.append(TestUnixDomain)
4839 if hasattr(socket, "AF_UNIX"): 5094 tests.append(TestLinuxAbstractNamespace)
4840 tests.append(TestUnixDomain) 5095 tests.extend([TIPCTest, TIPCThreadableTest])
4841 if sys.platform == 'linux':
4842 tests.append(TestLinuxAbstractNamespace)
4843 if isTipcAvailable():
4844 tests.append(TIPCTest)
4845 tests.append(TIPCThreadableTest)
4846 tests.extend([BasicCANTest, CANTest]) 5096 tests.extend([BasicCANTest, CANTest])
4847 tests.extend([BasicRDSTest, RDSTest]) 5097 tests.extend([BasicRDSTest, RDSTest])
4848 tests.extend([ 5098 tests.extend([
4849 CmsgMacroTests, 5099 CmsgMacroTests,
4850 SendmsgUDPTest, 5100 SendmsgUDPTest,
4851 RecvmsgUDPTest, 5101 RecvmsgUDPTest,
4852 RecvmsgIntoUDPTest, 5102 RecvmsgIntoUDPTest,
4853 SendmsgUDP6Test, 5103 SendmsgUDP6Test,
4854 RecvmsgUDP6Test, 5104 RecvmsgUDP6Test,
4855 RecvmsgRFC3542AncillaryUDP6Test, 5105 RecvmsgRFC3542AncillaryUDP6Test,
(...skipping 15 matching lines...) Expand all
4871 InterruptedSendTimeoutTest, 5121 InterruptedSendTimeoutTest,
4872 TestSocketSharing, 5122 TestSocketSharing,
4873 ]) 5123 ])
4874 5124
4875 thread_info = support.threading_setup() 5125 thread_info = support.threading_setup()
4876 support.run_unittest(*tests) 5126 support.run_unittest(*tests)
4877 support.threading_cleanup(*thread_info) 5127 support.threading_cleanup(*thread_info)
4878 5128
4879 if __name__ == "__main__": 5129 if __name__ == "__main__":
4880 test_main() 5130 test_main()
LEFTRIGHT

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