OLD | NEW |
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 Loading... |
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 Loading... |
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 Loading... |
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 Loading... |
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() |
OLD | NEW |