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

Side by Side Diff: Lib/test/test_socket.py

Issue 23001: Accept mutable bytes-like objects
Patch Set: Created 4 years, 8 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:
View unified diff | Download patch
OLDNEW
1 import unittest 1 import unittest
2 from test import support 2 from test import support
3 3
4 import errno 4 import errno
5 import io 5 import io
6 import itertools 6 import itertools
7 import socket 7 import socket
8 import select 8 import select
9 import tempfile 9 import tempfile
10 import time 10 import time
(...skipping 1056 matching lines...) Expand 10 before | Expand all | Expand 10 after
1067 (OSError, ValueError), func, a 1067 (OSError, ValueError), func, a
1068 ) 1068 )
1069 1069
1070 self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00')) 1070 self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00'))
1071 self.assertEqual('170.85.170.85', f(b'\xaa\x55\xaa\x55')) 1071 self.assertEqual('170.85.170.85', f(b'\xaa\x55\xaa\x55'))
1072 self.assertEqual('255.255.255.255', f(b'\xff\xff\xff\xff')) 1072 self.assertEqual('255.255.255.255', f(b'\xff\xff\xff\xff'))
1073 self.assertEqual('1.2.3.4', f(b'\x01\x02\x03\x04')) 1073 self.assertEqual('1.2.3.4', f(b'\x01\x02\x03\x04'))
1074 assertInvalid(f, b'\x00' * 3) 1074 assertInvalid(f, b'\x00' * 3)
1075 assertInvalid(f, b'\x00' * 5) 1075 assertInvalid(f, b'\x00' * 5)
1076 assertInvalid(f, b'\x00' * 16) 1076 assertInvalid(f, b'\x00' * 16)
1077 self.assertEqual('170.85.170.85', f(bytearray(b'\xaa\x55\xaa\x55')))
1077 1078
1078 self.assertEqual('1.0.1.0', g(b'\x01\x00\x01\x00')) 1079 self.assertEqual('1.0.1.0', g(b'\x01\x00\x01\x00'))
1079 self.assertEqual('170.85.170.85', g(b'\xaa\x55\xaa\x55')) 1080 self.assertEqual('170.85.170.85', g(b'\xaa\x55\xaa\x55'))
1080 self.assertEqual('255.255.255.255', g(b'\xff\xff\xff\xff')) 1081 self.assertEqual('255.255.255.255', g(b'\xff\xff\xff\xff'))
1081 assertInvalid(g, b'\x00' * 3) 1082 assertInvalid(g, b'\x00' * 3)
1082 assertInvalid(g, b'\x00' * 5) 1083 assertInvalid(g, b'\x00' * 5)
1083 assertInvalid(g, b'\x00' * 16) 1084 assertInvalid(g, b'\x00' * 16)
1085 self.assertEqual('170.85.170.85', g(bytearray(b'\xaa\x55\xaa\x55')))
1084 1086
1085 @unittest.skipUnless(hasattr(socket, 'inet_ntop'), 1087 @unittest.skipUnless(hasattr(socket, 'inet_ntop'),
1086 'test needs socket.inet_ntop()') 1088 'test needs socket.inet_ntop()')
1087 def testStringToIPv6(self): 1089 def testStringToIPv6(self):
1088 try: 1090 try:
1089 from socket import inet_ntop, AF_INET6, has_ipv6 1091 from socket import inet_ntop, AF_INET6, has_ipv6
1090 if not has_ipv6: 1092 if not has_ipv6:
1091 self.skipTest('IPv6 not available') 1093 self.skipTest('IPv6 not available')
1092 except ImportError: 1094 except ImportError:
1093 self.skipTest('could not import needed symbols from socket') 1095 self.skipTest('could not import needed symbols from socket')
1094 1096
1095 if sys.platform == "win32": 1097 if sys.platform == "win32":
1096 try: 1098 try:
1097 inet_ntop(AF_INET6, b'\x00' * 16) 1099 inet_ntop(AF_INET6, b'\x00' * 16)
1098 except OSError as e: 1100 except OSError as e:
1099 if e.winerror == 10022: 1101 if e.winerror == 10022:
1100 self.skipTest('IPv6 might not be supported') 1102 self.skipTest('IPv6 might not be supported')
1101 1103
1102 f = lambda a: inet_ntop(AF_INET6, a) 1104 f = lambda a: inet_ntop(AF_INET6, a)
1103 assertInvalid = lambda a: self.assertRaises( 1105 assertInvalid = lambda a: self.assertRaises(
1104 (OSError, ValueError), f, a 1106 (OSError, ValueError), f, a
1105 ) 1107 )
1106 1108
1107 self.assertEqual('::', f(b'\x00' * 16)) 1109 self.assertEqual('::', f(b'\x00' * 16))
1108 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01')) 1110 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01'))
1109 self.assertEqual( 1111 self.assertEqual(
1110 'aef:b01:506:1001:ffff:9997:55:170', 1112 'aef:b01:506:1001:ffff:9997:55:170',
1111 f(b'\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70 ') 1113 f(b'\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70 ')
1112 ) 1114 )
1115 self.assertEqual('::1', f(bytearray(b'\x00' * 15 + b'\x01')))
1113 1116
1114 assertInvalid(b'\x12' * 15) 1117 assertInvalid(b'\x12' * 15)
1115 assertInvalid(b'\x12' * 17) 1118 assertInvalid(b'\x12' * 17)
1116 assertInvalid(b'\x12' * 4) 1119 assertInvalid(b'\x12' * 4)
1117 1120
1118 # XXX The following don't test module-level functionality... 1121 # XXX The following don't test module-level functionality...
1119 1122
1120 def testSockName(self): 1123 def testSockName(self):
1121 # Testing getsockname() 1124 # Testing getsockname()
1122 port = support.find_unused_port() 1125 port = support.find_unused_port()
(...skipping 367 matching lines...) Expand 10 before | Expand all | Expand 10 after
1490 1493
1491 @unittest.skipUnless(hasattr(socket, "CAN_RAW_FILTER"), 1494 @unittest.skipUnless(hasattr(socket, "CAN_RAW_FILTER"),
1492 'socket.CAN_RAW_FILTER required for this test.') 1495 'socket.CAN_RAW_FILTER required for this test.')
1493 def testFilter(self): 1496 def testFilter(self):
1494 can_id, can_mask = 0x200, 0x700 1497 can_id, can_mask = 0x200, 0x700
1495 can_filter = struct.pack("=II", can_id, can_mask) 1498 can_filter = struct.pack("=II", can_id, can_mask)
1496 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s: 1499 with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
1497 s.setsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, can_filter) 1500 s.setsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, can_filter)
1498 self.assertEqual(can_filter, 1501 self.assertEqual(can_filter,
1499 s.getsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, 8)) 1502 s.getsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, 8))
1503 s.setsockopt(socket.SOL_CAN_RAW, socket.CAN_RAW_FILTER, bytearray(ca n_filter))
1500 1504
1501 1505
1502 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.') 1506 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.')
1503 @unittest.skipUnless(thread, 'Threading required for this test.') 1507 @unittest.skipUnless(thread, 'Threading required for this test.')
1504 class CANTest(ThreadedCANSocketTest): 1508 class CANTest(ThreadedCANSocketTest):
1505 1509
1506 def __init__(self, methodName='runTest'): 1510 def __init__(self, methodName='runTest'):
1507 ThreadedCANSocketTest.__init__(self, methodName=methodName) 1511 ThreadedCANSocketTest.__init__(self, methodName=methodName)
1508 1512
1509 @classmethod 1513 @classmethod
(...skipping 2990 matching lines...) Expand 10 before | Expand all | Expand 10 after
4500 self.assertRaises(OSError, s.bind, address) 4504 self.assertRaises(OSError, s.bind, address)
4501 4505
4502 def testStrName(self): 4506 def testStrName(self):
4503 # Check that an abstract name can be passed as a string. 4507 # Check that an abstract name can be passed as a string.
4504 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 4508 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
4505 try: 4509 try:
4506 s.bind("\x00python\x00test\x00") 4510 s.bind("\x00python\x00test\x00")
4507 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00") 4511 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00")
4508 finally: 4512 finally:
4509 s.close() 4513 s.close()
4514
4515 def testBytearrayName(self):
4516 # Check that an abstract name can be passed as a bytearray.
4517 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
4518 s.bind(bytearray(b"\x00python\x00test\x00"))
4519 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00")
4510 4520
4511 @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'test needs socket.AF_UNIX') 4521 @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'test needs socket.AF_UNIX')
4512 class TestUnixDomain(unittest.TestCase): 4522 class TestUnixDomain(unittest.TestCase):
4513 4523
4514 def setUp(self): 4524 def setUp(self):
4515 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 4525 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
4516 4526
4517 def tearDown(self): 4527 def tearDown(self):
4518 self.sock.close() 4528 self.sock.close()
4519 4529
(...skipping 805 matching lines...) Expand 10 before | Expand all | Expand 10 after
5325 SendfileUsingSendTest, 5335 SendfileUsingSendTest,
5326 SendfileUsingSendfileTest, 5336 SendfileUsingSendfileTest,
5327 ]) 5337 ])
5328 5338
5329 thread_info = support.threading_setup() 5339 thread_info = support.threading_setup()
5330 support.run_unittest(*tests) 5340 support.run_unittest(*tests)
5331 support.threading_cleanup(*thread_info) 5341 support.threading_cleanup(*thread_info)
5332 5342
5333 if __name__ == "__main__": 5343 if __name__ == "__main__":
5334 test_main() 5344 test_main()
OLDNEW

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