Lines Matching refs:self
179 def setUp(self):
180 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
181 self.port = socket_helper.bind_port(self.serv)
182 self.serv.listen()
184 def tearDown(self):
185 self.serv.close()
186 self.serv = None
190 def setUp(self):
191 self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
192 self.port = socket_helper.bind_port(self.serv)
194 def tearDown(self):
195 self.serv.close()
196 self.serv = None
200 def setUp(self):
201 self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDPLITE)
202 self.port = socket_helper.bind_port(self.serv)
211 def __init__(self, *args, **kwargs):
213 self._cleanup_lock = threading.RLock()
215 def addCleanup(self, *args, **kwargs):
216 with self._cleanup_lock:
219 def doCleanups(self, *args, **kwargs):
220 with self._cleanup_lock:
264 def setUp(self):
265 self.s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
266 self.addCleanup(self.s.close)
268 self.s.bind((self.interface,))
270 self.skipTest('network interface `%s` does not exist' %
271 self.interface)
281 def setUp(self):
282 self.serv = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)
283 self.addCleanup(self.serv.close)
285 self.port = socket_helper.bind_port(self.serv)
287 self.skipTest('unable to bind RDS socket')
311 def testFoo(self):
314 def _testFoo(self):
328 def __init__(self):
330 self.__setUp = self.setUp
331 self.setUp = self._setUp
333 def serverExplicitReady(self):
338 self.server_ready.set()
340 def _setUp(self):
341 self.enterContext(threading_helper.wait_threads_exit())
343 self.server_ready = threading.Event()
344 self.client_ready = threading.Event()
345 self.done = threading.Event()
346 self.queue = queue.Queue(1)
347 self.server_crashed = False
350 if self.queue.qsize():
351 raise self.queue.get()
352 self.addCleanup(raise_queued_exception)
355 methodname = self.id()
358 test_method = getattr(self, '_' + methodname)
359 self.client_thread = thread.start_new_thread(
360 self.clientRun, (test_method,))
363 self.__setUp()
365 self.server_crashed = True
368 self.server_ready.set()
369 self.client_ready.wait()
370 self.addCleanup(self.done.wait)
372 def clientRun(self, test_func):
373 self.server_ready.wait()
375 self.clientSetUp()
377 self.queue.put(e)
378 self.clientTearDown()
381 self.client_ready.set()
382 if self.server_crashed:
383 self.clientTearDown()
390 self.queue.put(e)
392 self.clientTearDown()
394 def clientSetUp(self):
397 def clientTearDown(self):
398 self.done.set()
403 def __init__(self, methodName='runTest'):
404 SocketTCPTest.__init__(self, methodName=methodName)
405 ThreadableTest.__init__(self)
407 def clientSetUp(self):
408 self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
410 def clientTearDown(self):
411 self.cli.close()
412 self.cli = None
413 ThreadableTest.clientTearDown(self)
417 def __init__(self, methodName='runTest'):
418 SocketUDPTest.__init__(self, methodName=methodName)
419 ThreadableTest.__init__(self)
421 def clientSetUp(self):
422 self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
424 def clientTearDown(self):
425 self.cli.close()
426 self.cli = None
427 ThreadableTest.clientTearDown(self)
433 def __init__(self, methodName='runTest'):
434 SocketUDPLITETest.__init__(self, methodName=methodName)
435 ThreadableTest.__init__(self)
437 def clientSetUp(self):
438 self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDPLITE)
440 def clientTearDown(self):
441 self.cli.close()
442 self.cli = None
443 ThreadableTest.clientTearDown(self)
447 def __init__(self, methodName='runTest'):
448 SocketCANTest.__init__(self, methodName=methodName)
449 ThreadableTest.__init__(self)
451 def clientSetUp(self):
452 self.cli = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
454 self.cli.bind((self.interface,))
460 def clientTearDown(self):
461 self.cli.close()
462 self.cli = None
463 ThreadableTest.clientTearDown(self)
467 def __init__(self, methodName='runTest'):
468 SocketRDSTest.__init__(self, methodName=methodName)
469 ThreadableTest.__init__(self)
471 def clientSetUp(self):
472 self.cli = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)
475 self.cli.bind((HOST, 0))
476 self.cli_addr = self.cli.getsockname()
482 def clientTearDown(self):
483 self.cli.close()
484 self.cli = None
485 ThreadableTest.clientTearDown(self)
494 def __init__(self, methodName='runTest'):
495 unittest.TestCase.__init__(self, methodName=methodName)
496 ThreadableTest.__init__(self)
498 def setUp(self):
499 self.serv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
500 self.addCleanup(self.serv.close)
501 self.serv.bind((socket.VMADDR_CID_ANY, VSOCKPORT))
502 self.serv.listen()
503 self.serverExplicitReady()
504 self.conn, self.connaddr = self.serv.accept()
505 self.addCleanup(self.conn.close)
507 def clientSetUp(self):
509 self.cli = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
510 self.addCleanup(self.cli.close)
512 self.cli.connect((cid, VSOCKPORT))
514 def testStream(self):
515 msg = self.conn.recv(1024)
516 self.assertEqual(msg, MSG)
518 def _testStream(self):
519 self.cli.send(MSG)
520 self.cli.close()
525 self.cli_conn is a client socket connected to the server. The
529 def __init__(self, methodName='runTest'):
530 ThreadedTCPSocketTest.__init__(self, methodName=methodName)
532 def setUp(self):
533 ThreadedTCPSocketTest.setUp(self)
536 self.serverExplicitReady()
537 conn, addr = self.serv.accept()
538 self.cli_conn = conn
540 def tearDown(self):
541 self.cli_conn.close()
542 self.cli_conn = None
543 ThreadedTCPSocketTest.tearDown(self)
545 def clientSetUp(self):
546 ThreadedTCPSocketTest.clientSetUp(self)
547 self.cli.connect((HOST, self.port))
548 self.serv_conn = self.cli
550 def clientTearDown(self):
551 self.serv_conn.close()
552 self.serv_conn = None
553 ThreadedTCPSocketTest.clientTearDown(self)
557 def __init__(self, methodName='runTest'):
558 unittest.TestCase.__init__(self, methodName=methodName)
559 ThreadableTest.__init__(self)
561 def setUp(self):
562 self.serv, self.cli = socket.socketpair()
564 def tearDown(self):
565 self.serv.close()
566 self.serv = None
568 def clientSetUp(self):
571 def clientTearDown(self):
572 self.cli.close()
573 self.cli = None
574 ThreadableTest.clientTearDown(self)
589 Creates a socket self.serv and sets self.serv_addr to its address.
592 def setUp(self):
593 self.serv = self.newSocket()
594 self.bindServer()
596 def bindServer(self):
597 """Bind server socket and set self.serv_addr to its address."""
598 self.bindSock(self.serv)
599 self.serv_addr = self.serv.getsockname()
601 def tearDown(self):
602 self.serv.close()
603 self.serv = None
609 def setUp(self):
611 self.serv.listen()
618 Client socket is self.cli and its address is self.cli_addr. See
622 def __init__(self, *args, **kwargs):
624 ThreadableTest.__init__(self)
626 def clientSetUp(self):
627 self.cli = self.newClientSocket()
628 self.bindClient()
630 def newClientSocket(self):
632 return self.newSocket()
634 def bindClient(self):
635 """Bind client socket and set self.cli_addr to its address."""
636 self.bindSock(self.cli)
637 self.cli_addr = self.cli.getsockname()
639 def clientTearDown(self):
640 self.cli.close()
641 self.cli = None
642 ThreadableTest.clientTearDown(self)
649 Server's socket representing connection to client is self.cli_conn
650 and client's connection to server is self.serv_conn. (Based on
654 def setUp(self):
658 self.serverExplicitReady()
659 conn, addr = self.serv.accept()
660 self.cli_conn = conn
662 def tearDown(self):
663 self.cli_conn.close()
664 self.cli_conn = None
667 def clientSetUp(self):
669 self.cli.connect(self.serv_addr)
670 self.serv_conn = self.cli
672 def clientTearDown(self):
674 self.serv_conn.close()
675 self.serv_conn = None
689 def setUp(self):
690 self.dir_path = tempfile.mkdtemp()
691 self.addCleanup(os.rmdir, self.dir_path)
694 def bindSock(self, sock):
695 path = tempfile.mktemp(dir=self.dir_path)
697 self.addCleanup(os_helper.unlink, path)
702 def newSocket(self):
711 def setUp(self):
713 self.port = self.serv_addr[1]
715 def bindSock(self, sock):
716 socket_helper.bind_port(sock, host=self.host)
721 def newSocket(self):
727 def newSocket(self):
733 def newSocket(self):
739 def newSocket(self):
752 def newSocket(self):
758 def newSocket(self):
830 def test_SocketType_is_socketobject(self):
832 self.assertTrue(socket.SocketType is _socket.socket)
834 self.assertIsInstance(s, socket.SocketType)
837 def test_repr(self):
840 self.assertIn('fd=%i' % s.fileno(), repr(s))
841 self.assertIn('family=%s' % socket.AF_INET, repr(s))
842 self.assertIn('type=%s' % socket.SOCK_STREAM, repr(s))
843 self.assertIn('proto=0', repr(s))
844 self.assertNotIn('raddr', repr(s))
846 self.assertIn('laddr', repr(s))
847 self.assertIn(str(s.getsockname()), repr(s))
848 self.assertIn('[closed]', repr(s))
849 self.assertNotIn('laddr', repr(s))
852 def test_csocket_repr(self):
857 self.assertEqual(repr(s), expected)
862 self.assertEqual(repr(s), expected)
864 def test_weakref(self):
867 self.assertEqual(p.fileno(), s.fileno())
875 self.fail('Socket proxy still exists')
877 def testSocketError(self):
880 with self.assertRaises(OSError, msg=msg % 'OSError'):
882 with self.assertRaises(OSError, msg=msg % 'socket.herror'):
884 with self.assertRaises(OSError, msg=msg % 'socket.gaierror'):
887 def testSendtoErrors(self):
890 self.addCleanup(s.close)
894 with self.assertRaises(TypeError) as cm:
896 self.assertEqual(str(cm.exception),
898 with self.assertRaises(TypeError) as cm:
900 self.assertEqual(str(cm.exception),
902 with self.assertRaises(TypeError) as cm:
904 self.assertIn('not NoneType',str(cm.exception))
906 with self.assertRaises(TypeError) as cm:
908 self.assertEqual(str(cm.exception),
910 with self.assertRaises(TypeError) as cm:
912 self.assertEqual(str(cm.exception),
914 with self.assertRaises(TypeError) as cm:
916 self.assertIn('not NoneType', str(cm.exception))
917 with self.assertRaises(TypeError) as cm:
919 with self.assertRaises(TypeError) as cm:
922 with self.assertRaises(TypeError) as cm:
924 self.assertIn('(1 given)', str(cm.exception))
925 with self.assertRaises(TypeError) as cm:
927 self.assertIn('(4 given)', str(cm.exception))
929 def testCrucialConstants(self):
942 def testCrucialIpProtoConstants(self):
949 def testWindowsSpecificConstants(self):
960 def test_socket_methods(self):
970 self.fail(f"socket method {name} is missing")
974 def test3542SocketOptions(self):
999 self.assertTrue(
1003 def testHostnameRes(self):
1010 self.skipTest('name lookup failure')
1011 self.assertTrue(ip.find('.') >= 0, "Error resolving host to ip.")
1016 self.skipTest('name lookup failure')
1020 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names)))
1022 def test_host_resolution(self):
1024 self.assertEqual(socket.gethostbyname(addr), addr)
1029 self.assertIn(host, socket.gethostbyaddr(host)[2])
1031 def test_host_resolution_bad_address(self):
1043 with self.assertRaises(OSError, msg=addr):
1045 with self.assertRaises(OSError, msg=explanation):
1050 def test_sethostname(self):
1056 self.skipTest("test should be run as root")
1061 self.assertEqual(socket.gethostname(), 'new')
1064 self.assertEqual(socket.gethostname(), 'bar')
1070 def testInterfaceNameIndex(self):
1073 self.assertIsInstance(index, int)
1074 self.assertIsInstance(name, str)
1076 self.assertGreater(index, 0)
1078 self.assertIsInstance(_index, int)
1079 self.assertEqual(index, _index)
1081 self.assertIsInstance(_name, str)
1082 self.assertEqual(name, _name)
1086 def testInvalidInterfaceIndexToName(self):
1087 self.assertRaises(OSError, socket.if_indextoname, 0)
1088 self.assertRaises(TypeError, socket.if_indextoname, '_DEADBEEF')
1092 def testInvalidInterfaceNameToIndex(self):
1093 self.assertRaises(TypeError, socket.if_nametoindex, 0)
1094 self.assertRaises(OSError, socket.if_nametoindex, '_DEADBEEF')
1098 def testRefCountGetNameInfo(self):
1106 self.fail("socket.getnameinfo loses a reference")
1108 def testInterpreterCrash(self):
1116 def testNtoH(self):
1124 self.assertEqual(i & mask, func(func(i&mask)) & mask)
1127 self.assertEqual(swapped & mask, mask)
1128 self.assertRaises(OverflowError, func, 1<<34)
1131 def testNtoHErrors(self):
1148 self.assertRaises(OverflowError, socket.ntohs, k)
1149 self.assertRaises(OverflowError, socket.htons, k)
1151 self.assertRaises(OverflowError, socket.ntohl, k)
1152 self.assertRaises(OverflowError, socket.htonl, k)
1154 def testGetServBy(self):
1155 eq = self.assertEqual
1195 self.assertRaises(OverflowError, socket.getservbyport, -1)
1196 self.assertRaises(OverflowError, socket.getservbyport, 65536)
1198 def testDefaultTimeout(self):
1201 self.assertEqual(socket.getdefaulttimeout(), None)
1203 self.assertEqual(s.gettimeout(), None)
1207 self.assertEqual(socket.getdefaulttimeout(), 10)
1209 self.assertEqual(sock.gettimeout(), 10)
1213 self.assertEqual(socket.getdefaulttimeout(), None)
1215 self.assertEqual(sock.gettimeout(), None)
1218 self.assertRaises(ValueError, socket.setdefaulttimeout, -1)
1221 self.assertRaises(TypeError, socket.setdefaulttimeout, "spam")
1225 def testIPv4_inet_aton_fourbytes(self):
1228 self.assertEqual(b'\x00'*4, socket.inet_aton('0.0.0.0'))
1229 self.assertEqual(b'\xff'*4, socket.inet_aton('255.255.255.255'))
1233 def testIPv4toString(self):
1237 assertInvalid = lambda func,a: self.assertRaises(
1241 self.assertEqual(b'\x00\x00\x00\x00', f('0.0.0.0'))
1242 self.assertEqual(b'\xff\x00\xff\x00', f('255.0.255.0'))
1243 self.assertEqual(b'\xaa\xaa\xaa\xaa', f('170.170.170.170'))
1244 self.assertEqual(b'\x01\x02\x03\x04', f('1.2.3.4'))
1245 self.assertEqual(b'\xff\xff\xff\xff', f('255.255.255.255'))
1254 self.assertEqual(b'\x00\x00\x00\x00', g('0.0.0.0'))
1255 self.assertEqual(b'\xff\x00\xff\x00', g('255.0.255.0'))
1256 self.assertEqual(b'\xaa\xaa\xaa\xaa', g('170.170.170.170'))
1257 self.assertEqual(b'\xff\xff\xff\xff', g('255.255.255.255'))
1266 def testIPv6toString(self):
1270 self.skipTest('IPv6 not available')
1272 self.skipTest('could not import needed symbols from socket')
1279 self.skipTest('IPv6 might not be supported')
1282 assertInvalid = lambda a: self.assertRaises(
1286 self.assertEqual(b'\x00' * 16, f('::'))
1287 self.assertEqual(b'\x00' * 16, f('0::0'))
1288 self.assertEqual(b'\x00\x01' + b'\x00' * 14, f('1::'))
1289 self.assertEqual(
1293 self.assertEqual(
1297 self.assertEqual(b'\x00\x12\x00\x0a' + b'\x00' * 12, f('12:a::'))
1310 self.assertEqual(b'\x00' * 12 + b'\xfe\x2a\x17\x40',
1313 self.assertEqual(
1317 self.assertEqual(
1330 def testStringToIPv4(self):
1333 assertInvalid = lambda func,a: self.assertRaises(
1337 self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00'))
1338 self.assertEqual('170.85.170.85', f(b'\xaa\x55\xaa\x55'))
1339 self.assertEqual('255.255.255.255', f(b'\xff\xff\xff\xff'))
1340 self.assertEqual('1.2.3.4', f(b'\x01\x02\x03\x04'))
1344 self.assertEqual('170.85.170.85', f(bytearray(b'\xaa\x55\xaa\x55')))
1346 self.assertEqual('1.0.1.0', g(b'\x01\x00\x01\x00'))
1347 self.assertEqual('170.85.170.85', g(b'\xaa\x55\xaa\x55'))
1348 self.assertEqual('255.255.255.255', g(b'\xff\xff\xff\xff'))
1352 self.assertEqual('170.85.170.85', g(bytearray(b'\xaa\x55\xaa\x55')))
1356 def testStringToIPv6(self):
1360 self.skipTest('IPv6 not available')
1362 self.skipTest('could not import needed symbols from socket')
1369 self.skipTest('IPv6 might not be supported')
1372 assertInvalid = lambda a: self.assertRaises(
1376 self.assertEqual('::', f(b'\x00' * 16))
1377 self.assertEqual('::1', f(b'\x00' * 15 + b'\x01'))
1378 self.assertEqual(
1382 self.assertEqual('::1', f(bytearray(b'\x00' * 15 + b'\x01')))
1390 def testSockName(self):
1393 self.addCleanup(sock.close)
1415 self.skipTest('name lookup failure')
1416 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0])
1417 self.assertEqual(name[1], port)
1419 def testGetSockOpt(self):
1423 self.addCleanup(sock.close)
1425 self.assertFalse(reuse != 0, "initial mode is reuse")
1427 def testSetSockOpt(self):
1430 self.addCleanup(sock.close)
1433 self.assertFalse(reuse == 0, "failed to set reuse mode")
1435 def testSendAfterClose(self):
1439 self.assertRaises(OSError, sock.send, b"spam")
1441 def testCloseException(self):
1449 self.assertIn(err.errno, (errno.EBADF, errno.ENOTSOCK))
1451 self.fail("close() should raise EBADF/ENOTSOCK")
1453 def testNewAttributes(self):
1457 self.assertEqual(sock.family, socket.AF_INET)
1459 self.assertIn(sock.type,
1463 self.assertEqual(sock.type, socket.SOCK_STREAM)
1464 self.assertEqual(sock.proto, 0)
1466 def test_getsockaddrarg(self):
1468 self.addCleanup(sock.close)
1472 self.assertRaises(OverflowError, sock.bind, (HOST, big_port))
1473 self.assertRaises(OverflowError, sock.bind, (HOST, neg_port))
1487 def test_sock_ioctl(self):
1488 self.assertTrue(hasattr(socket.socket, 'ioctl'))
1489 self.assertTrue(hasattr(socket, 'SIO_RCVALL'))
1490 self.assertTrue(hasattr(socket, 'RCVALL_ON'))
1491 self.assertTrue(hasattr(socket, 'RCVALL_OFF'))
1492 self.assertTrue(hasattr(socket, 'SIO_KEEPALIVE_VALS'))
1494 self.addCleanup(s.close)
1495 self.assertRaises(ValueError, s.ioctl, -1, None)
1501 def test_sio_loopback_fast_path(self):
1503 self.addCleanup(s.close)
1509 self.skipTest("SIO_LOOPBACK_FAST_PATH is defined but "
1512 self.assertRaises(TypeError, s.ioctl, socket.SIO_LOOPBACK_FAST_PATH, None)
1514 def testGetaddrinfo(self):
1520 self.skipTest("buggy libc version")
1524 self.assertEqual(len(info), 5)
1543 self.assertEqual(family, socket.AF_INET)
1544 self.assertEqual(repr(family), '<AddressFamily.AF_INET: %r>' % family.value)
1545 self.assertEqual(str(family), str(family.value))
1546 self.assertEqual(type, socket.SOCK_STREAM)
1547 self.assertEqual(repr(type), '<SocketKind.SOCK_STREAM: %r>' % type.value)
1548 self.assertEqual(str(type), str(type.value))
1551 self.assertEqual(socktype, socket.SOCK_STREAM)
1562 self.assertEqual(a, b)
1565 self.assertEqual(a, b)
1568 self.assertEqual(a, b)
1571 self.assertEqual(a, b)
1574 self.assertEqual(a, b)
1580 self.assertEqual(a, b)
1582 self.assertRaises(UnicodeEncodeError, socket.getaddrinfo, 'localhost', '\uD800')
1594 def test_getnameinfo(self):
1596 self.assertRaises(OSError, socket.getnameinfo, ('mail.python.org',0), 0)
1600 def test_idna(self):
1615 def check_sendall_interrupted(self, with_timeout):
1618 self.skipTest("signal.alarm and socket.socketpair required for this test")
1622 self.assertRaises(ValueError, math.acosh, 0)
1624 self.assertRaises(ValueError, math.acosh, 0)
1632 with self.assertRaises(ZeroDivisionError):
1638 self.assertRaises(TimeoutError, c.sendall,
1646 def test_sendall_interrupted(self):
1647 self.check_sendall_interrupted(False)
1649 def test_sendall_interrupted_with_timeout(self):
1650 self.check_sendall_interrupted(True)
1652 def test_dealloc_warn(self):
1655 with self.assertWarns(ResourceWarning) as cm:
1658 self.assertIn(r, str(cm.warning.args[0]))
1665 with self.assertWarns(ResourceWarning):
1669 def test_name_closed_socketio(self):
1673 self.assertEqual(repr(fp), "<_io.BufferedReader name=-1>")
1675 def test_unusable_closed_socketio(self):
1678 self.assertTrue(fp.readable())
1679 self.assertFalse(fp.writable())
1680 self.assertFalse(fp.seekable())
1682 self.assertRaises(ValueError, fp.readable)
1683 self.assertRaises(ValueError, fp.writable)
1684 self.assertRaises(ValueError, fp.seekable)
1686 def test_socket_close(self):
1691 with self.assertRaises(OSError):
1694 with self.assertRaises(OSError):
1697 with self.assertRaises(TypeError):
1699 with self.assertRaises(OSError):
1702 def test_makefile_mode(self):
1704 with self.subTest(mode=mode):
1708 self.assertEqual(fp.mode, mode)
1710 def test_makefile_invalid_mode(self):
1712 with self.subTest(mode=mode):
1714 with self.assertRaisesRegex(ValueError, 'invalid mode'):
1717 def test_pickle(self):
1721 self.assertRaises(TypeError, pickle.dumps, sock, protocol)
1724 self.assertEqual(family, socket.AF_INET)
1726 self.assertEqual(type, socket.SOCK_STREAM)
1728 def test_listen_backlog(self):
1739 def test_listen_backlog_overflow(self):
1744 self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1)
1747 def test_flowinfo(self):
1748 self.assertRaises(OverflowError, socket.getnameinfo,
1751 self.assertRaises(OverflowError, s.bind, (socket_helper.HOSTv6, 0, -10))
1754 def test_getaddrinfo_ipv6_basic(self):
1761 self.assertEqual(sockaddr, ('ff02::1de:c0:face:8d', 1234, 0, 0))
1763 def test_getfqdn_filter_localhost(self):
1764 self.assertEqual(socket.getfqdn(), socket.getfqdn("0.0.0.0"))
1765 self.assertEqual(socket.getfqdn(), socket.getfqdn("::"))
1771 def test_getaddrinfo_ipv6_scopeid_symbolic(self):
1781 self.assertEqual(sockaddr, ('ff02::1de:c0:face:8d', 1234, 0, ifindex))
1787 def test_getaddrinfo_ipv6_scopeid_numeric(self):
1798 self.assertEqual(sockaddr, ('ff02::1de:c0:face:8d', 1234, 0, ifindex))
1804 def test_getnameinfo_ipv6_scopeid_symbolic(self):
1809 self.assertEqual(nameinfo, ('ff02::1de:c0:face:8d%' + test_interface, '1234'))
1814 def test_getnameinfo_ipv6_scopeid_numeric(self):
1820 self.assertEqual(nameinfo, ('ff02::1de:c0:face:8d%' + str(ifindex), '1234'))
1822 def test_str_for_enums(self):
1826 self.assertEqual(repr(s.family), '<AddressFamily.AF_INET: %r>' % s.family.value)
1827 self.assertEqual(repr(s.type), '<SocketKind.SOCK_STREAM: %r>' % s.type.value)
1828 self.assertEqual(str(s.family), str(s.family.value))
1829 self.assertEqual(str(s.type), str(s.type.value))
1831 def test_socket_consistent_sock_type(self):
1837 self.assertEqual(s.type, socket.SOCK_STREAM)
1839 self.assertEqual(s.type, socket.SOCK_STREAM)
1841 self.assertEqual(s.type, socket.SOCK_STREAM)
1843 self.assertEqual(s.type, socket.SOCK_STREAM)
1845 self.assertEqual(s.type, socket.SOCK_STREAM)
1847 def test_unknown_socket_family_repr(self):
1867 self.assertEqual(s.family, unknown_family)
1868 self.assertEqual(s.type, unknown_type)
1870 self.assertIn(s.proto, {0, 23})
1873 def test__sendfile_use_sendfile(self):
1875 def __init__(self, fd):
1876 self.fd = fd
1878 def fileno(self):
1879 return self.fd
1883 with self.assertRaises(socket._GiveupOnSendfile):
1885 with self.assertRaises(OverflowError):
1887 with self.assertRaises(TypeError):
1890 def _test_socket_fileno(self, s, family, stype):
1891 self.assertEqual(s.family, family)
1892 self.assertEqual(s.type, stype)
1896 self.addCleanup(s2.close)
1899 self.assertEqual(s2.family, family)
1900 self.assertEqual(s2.type, stype)
1901 self.assertEqual(s2.fileno(), fd)
1903 def test_socket_fileno(self):
1905 self.addCleanup(s.close)
1907 self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_STREAM)
1911 self.addCleanup(s.close)
1913 self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_DGRAM)
1917 self.addCleanup(s.close)
1919 self._test_socket_fileno(s, socket.AF_INET6, socket.SOCK_STREAM)
1923 self.addCleanup(shutil.rmtree, tmpdir)
1925 self.addCleanup(s.close)
1931 self._test_socket_fileno(s, socket.AF_UNIX,
1934 def test_socket_fileno_rejects_float(self):
1935 with self.assertRaises(TypeError):
1938 def test_socket_fileno_rejects_other_types(self):
1939 with self.assertRaises(TypeError):
1942 def test_socket_fileno_rejects_invalid_socket(self):
1943 with self.assertRaisesRegex(ValueError, "negative file descriptor"):
1947 def test_socket_fileno_rejects_negative(self):
1948 with self.assertRaisesRegex(ValueError, "negative file descriptor"):
1951 def test_socket_fileno_requires_valid_fd(self):
1953 with self.assertRaises(OSError) as cm:
1955 self.assertIn(cm.exception.errno, (errno.EBADF, WSAENOTSOCK))
1957 with self.assertRaises(OSError) as cm:
1962 self.assertIn(cm.exception.errno, (errno.EBADF, WSAENOTSOCK))
1964 def test_socket_fileno_requires_socket_fd(self):
1966 with self.assertRaises(OSError):
1969 with self.assertRaises(OSError) as cm:
1974 self.assertEqual(cm.exception.errno, errno.ENOTSOCK)
1976 def test_addressfamily_enum(self):
1985 def test_socketkind_enum(self):
1994 def test_msgflag_enum(self):
2003 def test_addressinfo_enum(self):
2015 def testCrucialConstants(self):
2022 def testBCMConstants(self):
2052 def testCreateSocket(self):
2058 def testCreateBCMSocket(self):
2062 def testBindAny(self):
2066 self.assertEqual(s.getsockname(), address)
2068 def testTooLongInterfaceName(self):
2071 self.assertRaisesRegex(OSError, 'interface name too long',
2076 def testLoopback(self):
2081 self.assertEqual(loopback,
2086 def testFilter(self):
2091 self.assertEqual(can_filter,
2099 def __init__(self, methodName='runTest'):
2100 ThreadedCANSocketTest.__init__(self, methodName=methodName)
2115 def testSendFrame(self):
2116 cf, addr = self.s.recvfrom(self.bufsize)
2117 self.assertEqual(self.cf, cf)
2118 self.assertEqual(addr[0], self.interface)
2120 def _testSendFrame(self):
2121 self.cf = self.build_can_frame(0x00, b'\x01\x02\x03\x04\x05')
2122 self.cli.send(self.cf)
2124 def testSendMaxFrame(self):
2125 cf, addr = self.s.recvfrom(self.bufsize)
2126 self.assertEqual(self.cf, cf)
2128 def _testSendMaxFrame(self):
2129 self.cf = self.build_can_frame(0x00, b'\x07' * 8)
2130 self.cli.send(self.cf)
2132 def testSendMultiFrames(self):
2133 cf, addr = self.s.recvfrom(self.bufsize)
2134 self.assertEqual(self.cf1, cf)
2136 cf, addr = self.s.recvfrom(self.bufsize)
2137 self.assertEqual(self.cf2, cf)
2139 def _testSendMultiFrames(self):
2140 self.cf1 = self.build_can_frame(0x07, b'\x44\x33\x22\x11')
2141 self.cli.send(self.cf1)
2143 self.cf2 = self.build_can_frame(0x12, b'\x99\x22\x33')
2144 self.cli.send(self.cf2)
2148 def _testBCM(self):
2149 cf, addr = self.cli.recvfrom(self.bufsize)
2150 self.assertEqual(self.cf, cf)
2151 can_id, can_dlc, data = self.dissect_can_frame(cf)
2152 self.assertEqual(self.can_id, can_id)
2153 self.assertEqual(self.data, data)
2157 def testBCM(self):
2159 self.addCleanup(bcm.close)
2160 bcm.connect((self.interface,))
2161 self.can_id = 0x123
2162 self.data = bytes([0xc0, 0xff, 0xee])
2163 self.cf = self.build_can_frame(self.can_id, self.data)
2170 assert len(self.cf) == 16
2171 header = struct.pack(self.bcm_cmd_msg_fmt,
2182 header_plus_frame = header + self.cf
2184 self.assertEqual(bytes_sent, len(header_plus_frame))
2190 def __init__(self, *args, **kwargs):
2192 self.interface = "vcan0"
2194 def testCrucialConstants(self):
2200 def testCreateSocket(self):
2206 def testCreateISOTPSocket(self):
2210 def testTooLongInterfaceName(self):
2213 with self.assertRaisesRegex(OSError, 'interface name too long'):
2216 def testBind(self):
2219 addr = self.interface, 0x123, 0x456
2221 self.assertEqual(s.getsockname(), addr)
2224 self.skipTest('network interface `%s` does not exist' %
2225 self.interface)
2233 def __init__(self, *args, **kwargs):
2235 self.interface = "vcan0"
2239 def testJ1939Constants(self):
2274 def testCreateJ1939Socket(self):
2278 def testBind(self):
2281 addr = self.interface, socket.J1939_NO_NAME, socket.J1939_NO_PGN, socket.J1939_NO_ADDR
2283 self.assertEqual(s.getsockname(), addr)
2286 self.skipTest('network interface `%s` does not exist' %
2287 self.interface)
2295 def testCrucialConstants(self):
2299 def testCreateSocket(self):
2303 def testSocketBufferSize(self):
2313 def __init__(self, methodName='runTest'):
2314 ThreadedRDSSocketTest.__init__(self, methodName=methodName)
2316 def setUp(self):
2318 self.evt = threading.Event()
2320 def testSendAndRecv(self):
2321 data, addr = self.serv.recvfrom(self.bufsize)
2322 self.assertEqual(self.data, data)
2323 self.assertEqual(self.cli_addr, addr)
2325 def _testSendAndRecv(self):
2326 self.data = b'spam'
2327 self.cli.sendto(self.data, 0, (HOST, self.port))
2329 def testPeek(self):
2330 data, addr = self.serv.recvfrom(self.bufsize, socket.MSG_PEEK)
2331 self.assertEqual(self.data, data)
2332 data, addr = self.serv.recvfrom(self.bufsize)
2333 self.assertEqual(self.data, data)
2335 def _testPeek(self):
2336 self.data = b'spam'
2337 self.cli.sendto(self.data, 0, (HOST, self.port))
2340 def testSendAndRecvMsg(self):
2341 data, ancdata, msg_flags, addr = self.serv.recvmsg(self.bufsize)
2342 self.assertEqual(self.data, data)
2345 def _testSendAndRecvMsg(self):
2346 self.data = b'hello ' * 10
2347 self.cli.sendmsg([self.data], (), 0, (HOST, self.port))
2349 def testSendAndRecvMulti(self):
2350 data, addr = self.serv.recvfrom(self.bufsize)
2351 self.assertEqual(self.data1, data)
2353 data, addr = self.serv.recvfrom(self.bufsize)
2354 self.assertEqual(self.data2, data)
2356 def _testSendAndRecvMulti(self):
2357 self.data1 = b'bacon'
2358 self.cli.sendto(self.data1, 0, (HOST, self.port))
2360 self.data2 = b'egg'
2361 self.cli.sendto(self.data2, 0, (HOST, self.port))
2363 def testSelect(self):
2364 r, w, x = select.select([self.serv], [], [], 3.0)
2365 self.assertIn(self.serv, r)
2366 data, addr = self.serv.recvfrom(self.bufsize)
2367 self.assertEqual(self.data, data)
2369 def _testSelect(self):
2370 self.data = b'select'
2371 self.cli.sendto(self.data, 0, (HOST, self.port))
2377 def testCrucialConstants(self):
2380 def testCreateSocket(self):
2384 def testUnbound(self):
2386 self.assertEqual(s.getsockname()[1], 0)
2388 def testBindSock(self):
2391 self.assertNotEqual(s.getsockname()[1], 0)
2393 def testInvalidBindSock(self):
2395 self.assertRaises(OSError, socket_helper.bind_port, s, host=-2)
2397 def testAutoBindSock(self):
2400 self.assertNotEqual(s.getsockname()[1], 0)
2407 def testCrucialConstants(self):
2410 def testVSOCKConstants(self):
2420 def testCreateSocket(self):
2424 def testSocketBufferSize(self):
2440 self.assertEqual(orig_max * 2,
2443 self.assertEqual(orig * 2,
2446 self.assertEqual(orig_min * 2,
2455 def testBluetoothConstants(self):
2469 def testCreateRfcommSocket(self):
2474 def testCreateL2capSocket(self):
2479 def testCreateHciSocket(self):
2485 def testCreateScoSocket(self):
2492 def __init__(self, methodName='runTest'):
2493 SocketConnectedTest.__init__(self, methodName=methodName)
2495 def testRecv(self):
2497 msg = self.cli_conn.recv(1024)
2498 self.assertEqual(msg, MSG)
2500 def _testRecv(self):
2501 self.serv_conn.send(MSG)
2503 def testOverFlowRecv(self):
2505 seg1 = self.cli_conn.recv(len(MSG) - 3)
2506 seg2 = self.cli_conn.recv(1024)
2508 self.assertEqual(msg, MSG)
2510 def _testOverFlowRecv(self):
2511 self.serv_conn.send(MSG)
2513 def testRecvFrom(self):
2515 msg, addr = self.cli_conn.recvfrom(1024)
2516 self.assertEqual(msg, MSG)
2518 def _testRecvFrom(self):
2519 self.serv_conn.send(MSG)
2521 def testOverFlowRecvFrom(self):
2523 seg1, addr = self.cli_conn.recvfrom(len(MSG)-3)
2524 seg2, addr = self.cli_conn.recvfrom(1024)
2526 self.assertEqual(msg, MSG)
2528 def _testOverFlowRecvFrom(self):
2529 self.serv_conn.send(MSG)
2531 def testSendAll(self):
2535 read = self.cli_conn.recv(1024)
2539 self.assertEqual(msg, b'f' * 2048)
2541 def _testSendAll(self):
2543 self.serv_conn.sendall(big_chunk)
2545 def testFromFd(self):
2547 fd = self.cli_conn.fileno()
2549 self.addCleanup(sock.close)
2550 self.assertIsInstance(sock, socket.socket)
2552 self.assertEqual(msg, MSG)
2554 def _testFromFd(self):
2555 self.serv_conn.send(MSG)
2557 def testDup(self):
2559 sock = self.cli_conn.dup()
2560 self.addCleanup(sock.close)
2562 self.assertEqual(msg, MSG)
2564 def _testDup(self):
2565 self.serv_conn.send(MSG)
2567 def testShutdown(self):
2569 msg = self.cli_conn.recv(1024)
2570 self.assertEqual(msg, MSG)
2574 self.done.wait()
2576 def _testShutdown(self):
2577 self.serv_conn.send(MSG)
2578 self.serv_conn.shutdown(2)
2583 def _testShutdown_overflow(self):
2585 self.serv_conn.send(MSG)
2587 self.assertRaises(OverflowError, self.serv_conn.shutdown,
2589 self.assertRaises(OverflowError, self.serv_conn.shutdown,
2591 self.serv_conn.shutdown(2)
2593 def testDetach(self):
2595 fileno = self.cli_conn.fileno()
2596 f = self.cli_conn.detach()
2597 self.assertEqual(f, fileno)
2599 self.assertTrue(self.cli_conn._closed)
2600 self.assertRaises(OSError, self.cli_conn.recv, 1024)
2601 self.cli_conn.close()
2605 self.addCleanup(sock.close)
2607 self.assertEqual(msg, MSG)
2609 def _testDetach(self):
2610 self.serv_conn.send(MSG)
2615 def __init__(self, methodName='runTest'):
2616 ThreadedUDPSocketTest.__init__(self, methodName=methodName)
2618 def testSendtoAndRecv(self):
2620 msg = self.serv.recv(len(MSG))
2621 self.assertEqual(msg, MSG)
2623 def _testSendtoAndRecv(self):
2624 self.cli.sendto(MSG, 0, (HOST, self.port))
2626 def testRecvFrom(self):
2628 msg, addr = self.serv.recvfrom(len(MSG))
2629 self.assertEqual(msg, MSG)
2631 def _testRecvFrom(self):
2632 self.cli.sendto(MSG, 0, (HOST, self.port))
2634 def testRecvFromNegative(self):
2636 self.assertRaises(ValueError, self.serv.recvfrom, -1)
2638 def _testRecvFromNegative(self):
2639 self.cli.sendto(MSG, 0, (HOST, self.port))
2646 def __init__(self, methodName='runTest'):
2647 ThreadedUDPLITESocketTest.__init__(self, methodName=methodName)
2649 def testSendtoAndRecv(self):
2651 msg = self.serv.recv(len(MSG))
2652 self.assertEqual(msg, MSG)
2654 def _testSendtoAndRecv(self):
2655 self.cli.sendto(MSG, 0, (HOST, self.port))
2657 def testRecvFrom(self):
2659 msg, addr = self.serv.recvfrom(len(MSG))
2660 self.assertEqual(msg, MSG)
2662 def _testRecvFrom(self):
2663 self.cli.sendto(MSG, 0, (HOST, self.port))
2665 def testRecvFromNegative(self):
2667 self.assertRaises(ValueError, self.serv.recvfrom, -1)
2669 def _testRecvFromNegative(self):
2670 self.cli.sendto(MSG, 0, (HOST, self.port))
2706 def setUp(self):
2707 self.misc_event = threading.Event()
2710 def sendToServer(self, msg):
2712 return self.cli_sock.send(msg)
2718 def sendmsgToServer(self, *args):
2719 # Call sendmsg() on self.cli_sock with the given arguments,
2721 # corresponding items of self.sendmsg_to_server_defaults, if
2723 return self.cli_sock.sendmsg(
2724 *(args + self.sendmsg_to_server_defaults[len(args):]))
2726 def doRecvmsg(self, sock, bufsize, *args):
2733 self.registerRecvmsgResult(result)
2736 def registerRecvmsgResult(self, result):
2742 def checkRecvmsgAddress(self, addr1, addr2):
2745 self.assertEqual(addr1, addr2)
2763 def checkFlags(self, flags, eor=None, checkset=0, checkunset=0, ignore=0):
2791 defaultset = self.msg_flags_common_set
2792 defaultunset = self.msg_flags_common_unset
2795 defaultset |= self.msg_flags_eor_indicator
2796 defaultunset |= self.msg_flags_non_eor_indicator
2798 defaultset |= self.msg_flags_non_eor_indicator
2799 defaultunset |= self.msg_flags_eor_indicator
2815 self.assertEqual(flags & mask, checkset & mask)
2821 def doRecvmsg(self, sock, bufsize, *args):
2824 self.registerRecvmsgResult(result)
2825 self.assertGreaterEqual(result[0], 0)
2826 self.assertLessEqual(result[0], bufsize)
2834 def msg_flags_non_eor_indicator(self):
2842 def msg_flags_eor_indicator(self):
2852 def serv_sock(self):
2853 return self.serv
2856 def cli_sock(self):
2857 return self.cli
2860 def sendmsg_to_server_defaults(self):
2861 return ([], [], 0, self.serv_addr)
2863 def sendToServer(self, msg):
2864 return self.cli_sock.sendto(msg, self.serv_addr)
2874 def serv_sock(self):
2875 return self.cli_conn
2878 def cli_sock(self):
2879 return self.serv_conn
2881 def checkRecvmsgAddress(self, addr1, addr2):
2890 def setUp(self):
2892 self.serv_sock.settimeout(self.fail_timeout)
2899 def testSendmsg(self):
2901 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
2903 def _testSendmsg(self):
2904 self.assertEqual(self.sendmsgToServer([MSG]), len(MSG))
2906 def testSendmsgDataGenerator(self):
2908 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
2910 def _testSendmsgDataGenerator(self):
2911 self.assertEqual(self.sendmsgToServer((o for o in [MSG])),
2914 def testSendmsgAncillaryGenerator(self):
2916 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
2918 def _testSendmsgAncillaryGenerator(self):
2919 self.assertEqual(self.sendmsgToServer([MSG], (o for o in [])),
2922 def testSendmsgArray(self):
2924 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
2926 def _testSendmsgArray(self):
2927 self.assertEqual(self.sendmsgToServer([array.array("B", MSG)]),
2930 def testSendmsgGather(self):
2932 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
2934 def _testSendmsgGather(self):
2935 self.assertEqual(self.sendmsgToServer([MSG[:3], MSG[3:]]), len(MSG))
2937 def testSendmsgBadArgs(self):
2939 self.assertEqual(self.serv_sock.recv(1000), b"done")
2941 def _testSendmsgBadArgs(self):
2942 self.assertRaises(TypeError, self.cli_sock.sendmsg)
2943 self.assertRaises(TypeError, self.sendmsgToServer,
2945 self.assertRaises(TypeError, self.sendmsgToServer,
2947 self.assertRaises(TypeError, self.sendmsgToServer,
2949 self.assertRaises(TypeError, self.sendmsgToServer,
2951 self.assertRaises(TypeError, self.sendmsgToServer,
2953 self.assertRaises(TypeError, self.sendmsgToServer,
2955 self.assertRaises(TypeError, self.sendmsgToServer,
2957 self.sendToServer(b"done")
2959 def testSendmsgBadCmsg(self):
2961 self.assertEqual(self.serv_sock.recv(1000), b"done")
2963 def _testSendmsgBadCmsg(self):
2964 self.assertRaises(TypeError, self.sendmsgToServer,
2966 self.assertRaises(TypeError, self.sendmsgToServer,
2968 self.assertRaises(TypeError, self.sendmsgToServer,
2970 self.assertRaises(TypeError, self.sendmsgToServer,
2972 self.assertRaises(TypeError, self.sendmsgToServer,
2974 self.assertRaises(TypeError, self.sendmsgToServer,
2976 self.sendToServer(b"done")
2979 def testSendmsgBadMultiCmsg(self):
2982 self.assertEqual(self.serv_sock.recv(1000), b"done")
2985 def _testSendmsgBadMultiCmsg(self):
2986 self.assertRaises(TypeError, self.sendmsgToServer,
2988 self.assertRaises(TypeError, self.sendmsgToServer,
2990 self.sendToServer(b"done")
2992 def testSendmsgExcessCmsgReject(self):
2995 self.assertEqual(self.serv_sock.recv(1000), b"done")
2997 def _testSendmsgExcessCmsgReject(self):
3000 with self.assertRaises(OSError) as cm:
3001 self.sendmsgToServer([MSG], [(0, 0, b""), (0, 0, b"")])
3002 self.assertIsNone(cm.exception.errno)
3003 self.sendToServer(b"done")
3005 def testSendmsgAfterClose(self):
3009 def _testSendmsgAfterClose(self):
3010 self.cli_sock.close()
3011 self.assertRaises(OSError, self.sendmsgToServer, [MSG])
3018 def testSendmsgExplicitNoneAddr(self):
3020 self.assertEqual(self.serv_sock.recv(len(MSG)), MSG)
3022 def _testSendmsgExplicitNoneAddr(self):
3023 self.assertEqual(self.sendmsgToServer([MSG], [], 0, None), len(MSG))
3025 def testSendmsgTimeout(self):
3027 self.assertEqual(self.serv_sock.recv(512), b"a"*512)
3028 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3030 def _testSendmsgTimeout(self):
3032 self.cli_sock.settimeout(0.03)
3035 self.sendmsgToServer([b"a"*512])
3044 self.fail("TimeoutError not raised")
3046 self.misc_event.set()
3056 def testSendmsgDontWait(self):
3058 self.assertEqual(self.serv_sock.recv(512), b"a"*512)
3059 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3062 def _testSendmsgDontWait(self):
3064 with self.assertRaises(OSError) as cm:
3066 self.sendmsgToServer([b"a"*512], [], socket.MSG_DONTWAIT)
3069 self.assertIn(cm.exception.errno,
3072 self.misc_event.set()
3080 def testSendmsgNoDestAddr(self):
3085 def _testSendmsgNoDestAddr(self):
3086 self.assertRaises(OSError, self.cli_sock.sendmsg,
3088 self.assertRaises(OSError, self.cli_sock.sendmsg,
3096 def testRecvmsg(self):
3098 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG))
3099 self.assertEqual(msg, MSG)
3100 self.checkRecvmsgAddress(addr, self.cli_addr)
3101 self.assertEqual(ancdata, [])
3102 self.checkFlags(flags, eor=True)
3104 def _testRecvmsg(self):
3105 self.sendToServer(MSG)
3107 def testRecvmsgExplicitDefaults(self):
3109 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3111 self.assertEqual(msg, MSG)
3112 self.checkRecvmsgAddress(addr, self.cli_addr)
3113 self.assertEqual(ancdata, [])
3114 self.checkFlags(flags, eor=True)
3116 def _testRecvmsgExplicitDefaults(self):
3117 self.sendToServer(MSG)
3119 def testRecvmsgShorter(self):
3121 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3123 self.assertEqual(msg, MSG)
3124 self.checkRecvmsgAddress(addr, self.cli_addr)
3125 self.assertEqual(ancdata, [])
3126 self.checkFlags(flags, eor=True)
3128 def _testRecvmsgShorter(self):
3129 self.sendToServer(MSG)
3131 def testRecvmsgTrunc(self):
3133 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3135 self.assertEqual(msg, MSG[:-3])
3136 self.checkRecvmsgAddress(addr, self.cli_addr)
3137 self.assertEqual(ancdata, [])
3138 self.checkFlags(flags, eor=False)
3140 def _testRecvmsgTrunc(self):
3141 self.sendToServer(MSG)
3143 def testRecvmsgShortAncillaryBuf(self):
3145 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3147 self.assertEqual(msg, MSG)
3148 self.checkRecvmsgAddress(addr, self.cli_addr)
3149 self.assertEqual(ancdata, [])
3150 self.checkFlags(flags, eor=True)
3152 def _testRecvmsgShortAncillaryBuf(self):
3153 self.sendToServer(MSG)
3155 def testRecvmsgLongAncillaryBuf(self):
3157 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3159 self.assertEqual(msg, MSG)
3160 self.checkRecvmsgAddress(addr, self.cli_addr)
3161 self.assertEqual(ancdata, [])
3162 self.checkFlags(flags, eor=True)
3164 def _testRecvmsgLongAncillaryBuf(self):
3165 self.sendToServer(MSG)
3167 def testRecvmsgAfterClose(self):
3169 self.serv_sock.close()
3170 self.assertRaises(OSError, self.doRecvmsg, self.serv_sock, 1024)
3172 def _testRecvmsgAfterClose(self):
3175 def testRecvmsgTimeout(self):
3178 self.serv_sock.settimeout(0.03)
3179 self.assertRaises(TimeoutError,
3180 self.doRecvmsg, self.serv_sock, len(MSG))
3182 self.misc_event.set()
3184 def _testRecvmsgTimeout(self):
3185 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3188 def testRecvmsgPeek(self):
3193 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3196 self.assertEqual(msg, MSG[:-3])
3197 self.checkRecvmsgAddress(addr, self.cli_addr)
3198 self.assertEqual(ancdata, [])
3203 self.checkFlags(flags, eor=False,
3207 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3210 self.assertEqual(msg, MSG)
3211 self.checkRecvmsgAddress(addr, self.cli_addr)
3212 self.assertEqual(ancdata, [])
3213 self.checkFlags(flags, eor=True)
3216 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG))
3217 self.assertEqual(msg, MSG)
3218 self.checkRecvmsgAddress(addr, self.cli_addr)
3219 self.assertEqual(ancdata, [])
3220 self.checkFlags(flags, eor=True)
3223 def _testRecvmsgPeek(self):
3224 self.sendToServer(MSG)
3227 def testRecvmsgFromSendmsg(self):
3230 self.serv_sock.settimeout(self.fail_timeout)
3231 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG))
3232 self.assertEqual(msg, MSG)
3233 self.checkRecvmsgAddress(addr, self.cli_addr)
3234 self.assertEqual(ancdata, [])
3235 self.checkFlags(flags, eor=True)
3238 def _testRecvmsgFromSendmsg(self):
3239 self.assertEqual(self.sendmsgToServer([MSG[:3], MSG[3:]]), len(MSG))
3246 def testRecvmsgEOF(self):
3248 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, 1024)
3249 self.assertEqual(msg, b"")
3250 self.checkRecvmsgAddress(addr, self.cli_addr)
3251 self.assertEqual(ancdata, [])
3252 self.checkFlags(flags, eor=None) # Might not have end-of-record marker
3254 def _testRecvmsgEOF(self):
3255 self.cli_sock.close()
3257 def testRecvmsgOverflow(self):
3259 seg1, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3261 self.checkRecvmsgAddress(addr, self.cli_addr)
3262 self.assertEqual(ancdata, [])
3263 self.checkFlags(flags, eor=False)
3265 seg2, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, 1024)
3266 self.checkRecvmsgAddress(addr, self.cli_addr)
3267 self.assertEqual(ancdata, [])
3268 self.checkFlags(flags, eor=True)
3271 self.assertEqual(msg, MSG)
3273 def _testRecvmsgOverflow(self):
3274 self.sendToServer(MSG)
3280 def testRecvmsgBadArgs(self):
3282 self.assertRaises(TypeError, self.serv_sock.recvmsg)
3283 self.assertRaises(ValueError, self.serv_sock.recvmsg,
3285 self.assertRaises(ValueError, self.serv_sock.recvmsg,
3287 self.assertRaises(TypeError, self.serv_sock.recvmsg,
3289 self.assertRaises(TypeError, self.serv_sock.recvmsg,
3291 self.assertRaises(TypeError, self.serv_sock.recvmsg,
3293 self.assertRaises(TypeError, self.serv_sock.recvmsg,
3296 msg, ancdata, flags, addr = self.serv_sock.recvmsg(len(MSG), 0, 0)
3297 self.assertEqual(msg, MSG)
3298 self.checkRecvmsgAddress(addr, self.cli_addr)
3299 self.assertEqual(ancdata, [])
3300 self.checkFlags(flags, eor=True)
3302 def _testRecvmsgBadArgs(self):
3303 self.sendToServer(MSG)
3309 def testRecvmsgIntoBadArgs(self):
3312 self.assertRaises(TypeError, self.serv_sock.recvmsg_into)
3313 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3315 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3317 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3319 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3321 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3323 self.assertRaises(ValueError, self.serv_sock.recvmsg_into,
3325 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3327 self.assertRaises(TypeError, self.serv_sock.recvmsg_into,
3330 nbytes, ancdata, flags, addr = self.serv_sock.recvmsg_into([buf], 0, 0)
3331 self.assertEqual(nbytes, len(MSG))
3332 self.assertEqual(buf, bytearray(MSG))
3333 self.checkRecvmsgAddress(addr, self.cli_addr)
3334 self.assertEqual(ancdata, [])
3335 self.checkFlags(flags, eor=True)
3337 def _testRecvmsgIntoBadArgs(self):
3338 self.sendToServer(MSG)
3340 def testRecvmsgIntoGenerator(self):
3343 nbytes, ancdata, flags, addr = self.serv_sock.recvmsg_into(
3345 self.assertEqual(nbytes, len(MSG))
3346 self.assertEqual(buf, bytearray(MSG))
3347 self.checkRecvmsgAddress(addr, self.cli_addr)
3348 self.assertEqual(ancdata, [])
3349 self.checkFlags(flags, eor=True)
3351 def _testRecvmsgIntoGenerator(self):
3352 self.sendToServer(MSG)
3354 def testRecvmsgIntoArray(self):
3357 nbytes, ancdata, flags, addr = self.serv_sock.recvmsg_into([buf])
3358 self.assertEqual(nbytes, len(MSG))
3359 self.assertEqual(buf.tobytes(), MSG)
3360 self.checkRecvmsgAddress(addr, self.cli_addr)
3361 self.assertEqual(ancdata, [])
3362 self.checkFlags(flags, eor=True)
3364 def _testRecvmsgIntoArray(self):
3365 self.sendToServer(MSG)
3367 def testRecvmsgIntoScatter(self):
3372 nbytes, ancdata, flags, addr = self.serv_sock.recvmsg_into(
3374 self.assertEqual(nbytes, len(b"Mary had a little lamb"))
3375 self.assertEqual(b1, bytearray(b"Mary"))
3376 self.assertEqual(b2, bytearray(b"01 had a 9"))
3377 self.assertEqual(b3, bytearray(b"little lamb---"))
3378 self.checkRecvmsgAddress(addr, self.cli_addr)
3379 self.assertEqual(ancdata, [])
3380 self.checkFlags(flags, eor=True)
3382 def _testRecvmsgIntoScatter(self):
3383 self.sendToServer(b"Mary had a little lamb")
3400 def testCMSG_LEN(self):
3403 toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1
3407 self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2)
3411 self.assertEqual(ret - socket.CMSG_LEN(0), n)
3412 self.assertLessEqual(ret, self.socklen_t_limit)
3414 self.assertRaises(OverflowError, socket.CMSG_LEN, -1)
3417 self.assertRaises(OverflowError, socket.CMSG_LEN, toobig)
3418 self.assertRaises(OverflowError, socket.CMSG_LEN, sys.maxsize)
3421 def testCMSG_SPACE(self):
3424 toobig = self.socklen_t_limit - socket.CMSG_SPACE(1) + 1
3429 self.assertGreater(last, array.array("i").itemsize * 2)
3432 self.assertGreaterEqual(ret, last)
3433 self.assertGreaterEqual(ret, socket.CMSG_LEN(n))
3434 self.assertGreaterEqual(ret, n + socket.CMSG_LEN(0))
3435 self.assertLessEqual(ret, self.socklen_t_limit)
3438 self.assertRaises(OverflowError, socket.CMSG_SPACE, -1)
3441 self.assertRaises(OverflowError, socket.CMSG_SPACE, toobig)
3442 self.assertRaises(OverflowError, socket.CMSG_SPACE, sys.maxsize)
3453 def newFDs(self, n):
3459 self.addCleanup(os.unlink, path)
3460 self.addCleanup(os.close, fd)
3465 def checkFDs(self, fds):
3470 self.assertEqual(os.read(fd, 1024), str(n).encode())
3472 def registerRecvmsgResult(self, result):
3473 self.addCleanup(self.closeRecvmsgFDs, result)
3475 def closeRecvmsgFDs(self, recvmsg_result):
3487 def createAndSendFDs(self, n):
3490 self.assertEqual(
3491 self.sendmsgToServer([MSG],
3494 array.array("i", self.newFDs(n)))]),
3497 def checkRecvmsgFDs(self, numfds, result, maxcmsgs=1, ignoreflags=0):
3504 self.assertEqual(msg, MSG)
3505 self.checkRecvmsgAddress(addr, self.cli_addr)
3506 self.checkFlags(flags, eor=True, checkunset=socket.MSG_CTRUNC,
3509 self.assertIsInstance(ancdata, list)
3510 self.assertLessEqual(len(ancdata), maxcmsgs)
3513 self.assertIsInstance(item, tuple)
3515 self.assertEqual(cmsg_level, socket.SOL_SOCKET)
3516 self.assertEqual(cmsg_type, socket.SCM_RIGHTS)
3517 self.assertIsInstance(cmsg_data, bytes)
3518 self.assertEqual(len(cmsg_data) % SIZEOF_INT, 0)
3521 self.assertEqual(len(fds), numfds)
3522 self.checkFDs(fds)
3524 def testFDPassSimple(self):
3526 self.checkRecvmsgFDs(1, self.doRecvmsg(self.serv_sock,
3529 def _testFDPassSimple(self):
3530 self.assertEqual(
3531 self.sendmsgToServer(
3535 array.array("i", self.newFDs(1)).tobytes())]),
3538 def testMultipleFDPass(self):
3540 self.checkRecvmsgFDs(4, self.doRecvmsg(self.serv_sock,
3543 def _testMultipleFDPass(self):
3544 self.createAndSendFDs(4)
3547 def testFDPassCMSG_SPACE(self):
3549 self.checkRecvmsgFDs(
3550 4, self.doRecvmsg(self.serv_sock, len(MSG),
3554 def _testFDPassCMSG_SPACE(self):
3555 self.createAndSendFDs(4)
3557 def testFDPassCMSG_LEN(self):
3559 self.checkRecvmsgFDs(1,
3560 self.doRecvmsg(self.serv_sock, len(MSG),
3567 def _testFDPassCMSG_LEN(self):
3568 self.createAndSendFDs(1)
3573 def testFDPassSeparate(self):
3576 self.checkRecvmsgFDs(2,
3577 self.doRecvmsg(self.serv_sock, len(MSG), 10240),
3583 def _testFDPassSeparate(self):
3584 fd0, fd1 = self.newFDs(2)
3585 self.assertEqual(
3586 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET,
3597 def testFDPassSeparateMinSpace(self):
3601 self.checkRecvmsgFDs(num_fds,
3602 self.doRecvmsg(self.serv_sock, len(MSG),
3610 def _testFDPassSeparateMinSpace(self):
3611 fd0, fd1 = self.newFDs(2)
3612 self.assertEqual(
3613 self.sendmsgToServer([MSG], [(socket.SOL_SOCKET,
3621 def sendAncillaryIfPossible(self, msg, ancdata):
3625 nbytes = self.sendmsgToServer([msg], ancdata)
3628 self.assertIsInstance(e.errno, int)
3629 nbytes = self.sendmsgToServer([msg])
3630 self.assertEqual(nbytes, len(msg))
3633 def testFDPassEmpty(self):
3636 self.checkRecvmsgFDs(0, self.doRecvmsg(self.serv_sock,
3640 def _testFDPassEmpty(self):
3641 self.sendAncillaryIfPossible(MSG, [(socket.SOL_SOCKET,
3645 def testFDPassPartialInt(self):
3647 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3649 self.assertEqual(msg, MSG)
3650 self.checkRecvmsgAddress(addr, self.cli_addr)
3651 self.checkFlags(flags, eor=True, ignore=socket.MSG_CTRUNC)
3652 self.assertLessEqual(len(ancdata), 1)
3654 self.assertEqual(cmsg_level, socket.SOL_SOCKET)
3655 self.assertEqual(cmsg_type, socket.SCM_RIGHTS)
3656 self.assertLess(len(cmsg_data), SIZEOF_INT)
3658 def _testFDPassPartialInt(self):
3659 self.sendAncillaryIfPossible(
3663 array.array("i", [self.badfd]).tobytes()[:-1])])
3666 def testFDPassPartialIntInMiddle(self):
3668 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3670 self.assertEqual(msg, MSG)
3671 self.checkRecvmsgAddress(addr, self.cli_addr)
3672 self.checkFlags(flags, eor=True, ignore=socket.MSG_CTRUNC)
3673 self.assertLessEqual(len(ancdata), 2)
3677 self.assertEqual(cmsg_level, socket.SOL_SOCKET)
3678 self.assertEqual(cmsg_type, socket.SCM_RIGHTS)
3681 self.assertLessEqual(len(fds), 2)
3682 self.checkFDs(fds)
3685 def _testFDPassPartialIntInMiddle(self):
3686 fd0, fd1 = self.newFDs(2)
3687 self.sendAncillaryIfPossible(
3691 array.array("i", [fd0, self.badfd]).tobytes()[:-1]),
3696 def checkTruncatedHeader(self, result, ignoreflags=0):
3700 self.assertEqual(msg, MSG)
3701 self.checkRecvmsgAddress(addr, self.cli_addr)
3702 self.assertEqual(ancdata, [])
3703 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC,
3706 def testCmsgTruncNoBufSize(self):
3709 self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG)),
3715 def _testCmsgTruncNoBufSize(self):
3716 self.createAndSendFDs(1)
3718 def testCmsgTrunc0(self):
3720 self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG), 0),
3723 def _testCmsgTrunc0(self):
3724 self.createAndSendFDs(1)
3729 def testCmsgTrunc1(self):
3730 self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG), 1))
3732 def _testCmsgTrunc1(self):
3733 self.createAndSendFDs(1)
3735 def testCmsgTrunc2Int(self):
3739 self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG),
3742 def _testCmsgTrunc2Int(self):
3743 self.createAndSendFDs(1)
3745 def testCmsgTruncLen0Minus1(self):
3746 self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG),
3749 def _testCmsgTruncLen0Minus1(self):
3750 self.createAndSendFDs(1)
3755 def checkTruncatedArray(self, ancbuf, maxdata, mindata=0):
3760 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3762 self.assertEqual(msg, MSG)
3763 self.checkRecvmsgAddress(addr, self.cli_addr)
3764 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC)
3768 self.assertEqual(len(ancdata), 1)
3770 self.assertEqual(cmsg_level, socket.SOL_SOCKET)
3771 self.assertEqual(cmsg_type, socket.SCM_RIGHTS)
3772 self.assertGreaterEqual(len(cmsg_data), mindata)
3773 self.assertLessEqual(len(cmsg_data), maxdata)
3777 self.checkFDs(fds)
3779 def testCmsgTruncLen0(self):
3780 self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(0), maxdata=0)
3782 def _testCmsgTruncLen0(self):
3783 self.createAndSendFDs(1)
3785 def testCmsgTruncLen0Plus1(self):
3786 self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(0) + 1, maxdata=1)
3788 def _testCmsgTruncLen0Plus1(self):
3789 self.createAndSendFDs(2)
3791 def testCmsgTruncLen1(self):
3792 self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(SIZEOF_INT),
3795 def _testCmsgTruncLen1(self):
3796 self.createAndSendFDs(2)
3798 def testCmsgTruncLen2Minus1(self):
3799 self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(2 * SIZEOF_INT) - 1,
3802 def _testCmsgTruncLen2Minus1(self):
3803 self.createAndSendFDs(2)
3823 def ancillaryMapping(self, ancdata):
3829 self.assertNotIn((cmsg_level, cmsg_type), d)
3833 def checkHopLimit(self, ancbufsize, maxhop=255, ignoreflags=0):
3838 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
3840 self.misc_event.set()
3841 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3844 self.assertEqual(msg, MSG)
3845 self.checkRecvmsgAddress(addr, self.cli_addr)
3846 self.checkFlags(flags, eor=True, checkunset=socket.MSG_CTRUNC,
3849 self.assertEqual(len(ancdata), 1)
3850 self.assertIsInstance(ancdata[0], tuple)
3852 self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
3853 self.assertEqual(cmsg_type, socket.IPV6_HOPLIMIT)
3854 self.assertIsInstance(cmsg_data, bytes)
3855 self.assertEqual(len(cmsg_data), SIZEOF_INT)
3858 self.assertGreaterEqual(a[0], 0)
3859 self.assertLessEqual(a[0], maxhop)
3862 def testRecvHopLimit(self):
3864 self.checkHopLimit(ancbufsize=10240)
3867 def _testRecvHopLimit(self):
3871 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3872 self.sendToServer(MSG)
3875 def testRecvHopLimitCMSG_SPACE(self):
3877 self.checkHopLimit(ancbufsize=socket.CMSG_SPACE(SIZEOF_INT))
3880 def _testRecvHopLimitCMSG_SPACE(self):
3881 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3882 self.sendToServer(MSG)
3891 def testSetHopLimit(self):
3894 self.checkHopLimit(ancbufsize=10240, maxhop=self.hop_limit)
3897 def _testSetHopLimit(self):
3898 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3899 self.assertEqual(
3900 self.sendmsgToServer([MSG],
3902 array.array("i", [self.hop_limit]))]),
3905 def checkTrafficClassAndHopLimit(self, ancbufsize, maxhop=255,
3912 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
3914 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
3916 self.misc_event.set()
3917 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
3920 self.assertEqual(msg, MSG)
3921 self.checkRecvmsgAddress(addr, self.cli_addr)
3922 self.checkFlags(flags, eor=True, checkunset=socket.MSG_CTRUNC,
3924 self.assertEqual(len(ancdata), 2)
3925 ancmap = self.ancillaryMapping(ancdata)
3928 self.assertEqual(len(tcdata), SIZEOF_INT)
3931 self.assertGreaterEqual(a[0], 0)
3932 self.assertLessEqual(a[0], 255)
3935 self.assertEqual(len(hldata), SIZEOF_INT)
3938 self.assertGreaterEqual(a[0], 0)
3939 self.assertLessEqual(a[0], maxhop)
3943 def testRecvTrafficClassAndHopLimit(self):
3945 self.checkTrafficClassAndHopLimit(ancbufsize=10240)
3948 def _testRecvTrafficClassAndHopLimit(self):
3949 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3950 self.sendToServer(MSG)
3954 def testRecvTrafficClassAndHopLimitCMSG_SPACE(self):
3957 self.checkTrafficClassAndHopLimit(
3961 def _testRecvTrafficClassAndHopLimitCMSG_SPACE(self):
3962 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3963 self.sendToServer(MSG)
3968 def testSetTrafficClassAndHopLimit(self):
3971 self.checkTrafficClassAndHopLimit(ancbufsize=10240,
3972 maxhop=self.hop_limit)
3975 def _testSetTrafficClassAndHopLimit(self):
3976 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3977 self.assertEqual(
3978 self.sendmsgToServer([MSG],
3980 array.array("i", [self.traffic_class])),
3982 array.array("i", [self.hop_limit]))]),
3988 def testOddCmsgSize(self):
3992 self.checkTrafficClassAndHopLimit(ancbufsize=10240,
3993 maxhop=self.hop_limit)
3996 def _testOddCmsgSize(self):
3997 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
3999 nbytes = self.sendmsgToServer(
4002 array.array("i", [self.traffic_class]).tobytes() + b"\x00"),
4004 array.array("i", [self.hop_limit]))])
4006 self.assertIsInstance(e.errno, int)
4007 nbytes = self.sendmsgToServer(
4010 array.array("i", [self.traffic_class])),
4012 array.array("i", [self.hop_limit]))])
4013 self.assertEqual(nbytes, len(MSG))
4017 def checkHopLimitTruncatedHeader(self, ancbufsize, ignoreflags=0):
4024 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4026 self.misc_event.set()
4028 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
4031 self.assertEqual(msg, MSG)
4032 self.checkRecvmsgAddress(addr, self.cli_addr)
4033 self.assertEqual(ancdata, [])
4034 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC,
4038 def testCmsgTruncNoBufSize(self):
4041 self.checkHopLimitTruncatedHeader(ancbufsize=None,
4049 def _testCmsgTruncNoBufSize(self):
4050 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4051 self.sendToServer(MSG)
4054 def testSingleCmsgTrunc0(self):
4057 self.checkHopLimitTruncatedHeader(ancbufsize=0,
4061 def _testSingleCmsgTrunc0(self):
4062 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4063 self.sendToServer(MSG)
4069 def testSingleCmsgTrunc1(self):
4070 self.checkHopLimitTruncatedHeader(ancbufsize=1)
4073 def _testSingleCmsgTrunc1(self):
4074 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4075 self.sendToServer(MSG)
4078 def testSingleCmsgTrunc2Int(self):
4079 self.checkHopLimitTruncatedHeader(ancbufsize=2 * SIZEOF_INT)
4082 def _testSingleCmsgTrunc2Int(self):
4083 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4084 self.sendToServer(MSG)
4087 def testSingleCmsgTruncLen0Minus1(self):
4088 self.checkHopLimitTruncatedHeader(ancbufsize=socket.CMSG_LEN(0) - 1)
4091 def _testSingleCmsgTruncLen0Minus1(self):
4092 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4093 self.sendToServer(MSG)
4096 def testSingleCmsgTruncInData(self):
4100 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4102 self.misc_event.set()
4103 msg, ancdata, flags, addr = self.doRecvmsg(
4104 self.serv_sock, len(MSG), socket.CMSG_LEN(SIZEOF_INT) - 1)
4106 self.assertEqual(msg, MSG)
4107 self.checkRecvmsgAddress(addr, self.cli_addr)
4108 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC)
4110 self.assertLessEqual(len(ancdata), 1)
4113 self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
4114 self.assertEqual(cmsg_type, socket.IPV6_HOPLIMIT)
4115 self.assertLess(len(cmsg_data), SIZEOF_INT)
4118 def _testSingleCmsgTruncInData(self):
4119 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4120 self.sendToServer(MSG)
4122 def checkTruncatedSecondHeader(self, ancbufsize, ignoreflags=0):
4129 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4131 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4133 self.misc_event.set()
4134 msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock,
4137 self.assertEqual(msg, MSG)
4138 self.checkRecvmsgAddress(addr, self.cli_addr)
4139 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC,
4142 self.assertEqual(len(ancdata), 1)
4144 self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
4145 self.assertIn(cmsg_type, {socket.IPV6_TCLASS, socket.IPV6_HOPLIMIT})
4146 self.assertEqual(len(cmsg_data), SIZEOF_INT)
4149 self.assertGreaterEqual(a[0], 0)
4150 self.assertLessEqual(a[0], 255)
4156 def testSecondCmsgTrunc0(self):
4157 self.checkTruncatedSecondHeader(socket.CMSG_SPACE(SIZEOF_INT),
4161 def _testSecondCmsgTrunc0(self):
4162 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4163 self.sendToServer(MSG)
4167 def testSecondCmsgTrunc1(self):
4168 self.checkTruncatedSecondHeader(socket.CMSG_SPACE(SIZEOF_INT) + 1)
4171 def _testSecondCmsgTrunc1(self):
4172 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4173 self.sendToServer(MSG)
4177 def testSecondCmsgTrunc2Int(self):
4178 self.checkTruncatedSecondHeader(socket.CMSG_SPACE(SIZEOF_INT) +
4182 def _testSecondCmsgTrunc2Int(self):
4183 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4184 self.sendToServer(MSG)
4188 def testSecondCmsgTruncLen0Minus1(self):
4189 self.checkTruncatedSecondHeader(socket.CMSG_SPACE(SIZEOF_INT) +
4193 def _testSecondCmsgTruncLen0Minus1(self):
4194 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4195 self.sendToServer(MSG)
4199 def testSecondCmsgTruncInData(self):
4202 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4204 self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
4206 self.misc_event.set()
4207 msg, ancdata, flags, addr = self.doRecvmsg(
4208 self.serv_sock, len(MSG),
4211 self.assertEqual(msg, MSG)
4212 self.checkRecvmsgAddress(addr, self.cli_addr)
4213 self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC)
4218 self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
4220 self.assertEqual(len(cmsg_data), SIZEOF_INT)
4223 self.assertGreaterEqual(a[0], 0)
4224 self.assertLessEqual(a[0], 255)
4228 self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
4230 self.assertLess(len(cmsg_data), SIZEOF_INT)
4232 self.assertEqual(ancdata, [])
4235 def _testSecondCmsgTruncInData(self):
4236 self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
4237 self.sendToServer(MSG)
4264 def checkRecvmsgAddress(self, addr1, addr2):
4267 self.assertEqual(addr1[:-1], addr2[:-1])
4337 def checkRecvmsgAddress(self, addr1, addr2):
4340 self.assertEqual(addr1[:-1], addr2[:-1])
4424 def testRecvmsgEOF(self):
4426 super(RecvmsgSCTPStreamTest, self).testRecvmsgEOF()
4430 self.skipTest("sporadic ENOTCONN (kernel issue?) - see issue #13876")
4438 def testRecvmsgEOF(self):
4440 super(RecvmsgIntoSCTPStreamTest, self).testRecvmsgEOF()
4444 self.skipTest("sporadic ENOTCONN (kernel issue?) - see issue #13876")
4490 def setUp(self):
4494 self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler)
4505 def setAlarm(self, seconds):
4511 def setAlarm(self, seconds):
4524 def setUp(self):
4526 self.serv.settimeout(self.timeout)
4528 def checkInterruptedRecv(self, func, *args, **kwargs):
4532 self.setAlarm(self.alarm_time)
4533 with self.assertRaises(ZeroDivisionError) as cm:
4536 self.setAlarm(0)
4538 def testInterruptedRecvTimeout(self):
4539 self.checkInterruptedRecv(self.serv.recv, 1024)
4541 def testInterruptedRecvIntoTimeout(self):
4542 self.checkInterruptedRecv(self.serv.recv_into, bytearray(1024))
4544 def testInterruptedRecvfromTimeout(self):
4545 self.checkInterruptedRecv(self.serv.recvfrom, 1024)
4547 def testInterruptedRecvfromIntoTimeout(self):
4548 self.checkInterruptedRecv(self.serv.recvfrom_into, bytearray(1024))
4551 def testInterruptedRecvmsgTimeout(self):
4552 self.checkInterruptedRecv(self.serv.recvmsg, 1024)
4555 def testInterruptedRecvmsgIntoTimeout(self):
4556 self.checkInterruptedRecv(self.serv.recvmsg_into, [bytearray(1024)])
4570 def setUp(self):
4572 self.serv_conn = self.newSocket()
4573 self.addCleanup(self.serv_conn.close)
4577 cli_thread = threading.Thread(target=self.doConnect)
4579 self.cli_conn, addr = self.serv.accept()
4580 self.addCleanup(self.cli_conn.close)
4582 self.serv_conn.settimeout(self.timeout)
4584 def doConnect(self):
4585 self.serv_conn.connect(self.serv_addr)
4587 def checkInterruptedSend(self, func, *args, **kwargs):
4592 with self.assertRaises(ZeroDivisionError) as cm:
4594 self.setAlarm(self.alarm_time)
4597 self.setAlarm(0)
4601 def testInterruptedSendTimeout(self):
4602 self.checkInterruptedSend(self.serv_conn.send, b"a"*512)
4605 def testInterruptedSendtoTimeout(self):
4610 self.checkInterruptedSend(self.serv_conn.sendto, b"a"*512,
4611 self.serv_addr)
4615 def testInterruptedSendmsgTimeout(self):
4616 self.checkInterruptedSend(self.serv_conn.sendmsg, [b"a"*512])
4621 def testClose(self):
4622 conn, addr = self.serv.accept()
4625 sd = self.cli
4627 self.assertEqual(read, [sd])
4628 self.assertEqual(sd.recv(1), b'')
4634 def _testClose(self):
4635 self.cli.connect((HOST, self.port))
4641 def __init__(self, methodName='runTest'):
4642 SocketPairTest.__init__(self, methodName=methodName)
4644 def _check_defaults(self, sock):
4645 self.assertIsInstance(sock, socket.socket)
4647 self.assertEqual(sock.family, socket.AF_UNIX)
4649 self.assertEqual(sock.family, socket.AF_INET)
4650 self.assertEqual(sock.type, socket.SOCK_STREAM)
4651 self.assertEqual(sock.proto, 0)
4653 def _testDefaults(self):
4654 self._check_defaults(self.cli)
4656 def testDefaults(self):
4657 self._check_defaults(self.serv)
4659 def testRecv(self):
4660 msg = self.serv.recv(1024)
4661 self.assertEqual(msg, MSG)
4663 def _testRecv(self):
4664 self.cli.send(MSG)
4666 def testSend(self):
4667 self.serv.send(MSG)
4669 def _testSend(self):
4670 msg = self.cli.recv(1024)
4671 self.assertEqual(msg, MSG)
4676 def __init__(self, methodName='runTest'):
4677 self.event = threading.Event()
4678 ThreadedTCPSocketTest.__init__(self, methodName=methodName)
4680 def assert_sock_timeout(self, sock, timeout):
4681 self.assertEqual(self.serv.gettimeout(), timeout)
4684 self.assertEqual(sock.getblocking(), blocking)
4695 self.assertEqual(not bool(flag & os.O_NONBLOCK), fd_blocking)
4697 def testSetBlocking(self):
4699 self.serv.setblocking(True)
4700 self.assert_sock_timeout(self.serv, None)
4702 self.serv.setblocking(False)
4703 self.assert_sock_timeout(self.serv, 0.0)
4705 self.serv.settimeout(None)
4706 self.assert_sock_timeout(self.serv, None)
4708 self.serv.settimeout(0)
4709 self.assert_sock_timeout(self.serv, 0)
4711 self.serv.settimeout(10)
4712 self.assert_sock_timeout(self.serv, 10)
4714 self.serv.settimeout(0)
4715 self.assert_sock_timeout(self.serv, 0)
4717 def _testSetBlocking(self):
4721 def testSetBlocking_overflow(self):
4725 self.skipTest('needs UINT_MAX < ULONG_MAX')
4727 self.serv.setblocking(False)
4728 self.assertEqual(self.serv.gettimeout(), 0.0)
4730 self.serv.setblocking(_testcapi.UINT_MAX + 1)
4731 self.assertIsNone(self.serv.gettimeout())
4738 def testInitNonBlocking(self):
4740 self.serv.close()
4741 self.serv = socket.socket(socket.AF_INET,
4743 self.assert_sock_timeout(self.serv, 0)
4745 def _testInitNonBlocking(self):
4748 def testInheritFlagsBlocking(self):
4752 self.serv.settimeout(10)
4753 conn, addr = self.serv.accept()
4754 self.addCleanup(conn.close)
4755 self.assertIsNone(conn.gettimeout())
4757 def _testInheritFlagsBlocking(self):
4758 self.cli.connect((HOST, self.port))
4760 def testInheritFlagsTimeout(self):
4766 self.serv.settimeout(10)
4767 conn, addr = self.serv.accept()
4768 self.addCleanup(conn.close)
4769 self.assertEqual(conn.gettimeout(), default_timeout)
4771 def _testInheritFlagsTimeout(self):
4772 self.cli.connect((HOST, self.port))
4774 def testAccept(self):
4776 self.serv.setblocking(False)
4780 with self.assertRaises(BlockingIOError):
4781 conn, addr = self.serv.accept()
4783 self.assertLess(dt, 1.0)
4785 self.event.set()
4787 read, write, err = select.select([self.serv], [], [], support.LONG_TIMEOUT)
4788 if self.serv not in read:
4789 self.fail("Error trying to do accept after select.")
4792 conn, addr = self.serv.accept()
4793 self.addCleanup(conn.close)
4794 self.assertIsNone(conn.gettimeout())
4796 def _testAccept(self):
4799 self.event.wait()
4801 self.cli.connect((HOST, self.port))
4803 def testRecv(self):
4805 conn, addr = self.serv.accept()
4806 self.addCleanup(conn.close)
4810 with self.assertRaises(BlockingIOError):
4813 self.event.set()
4817 self.fail("Error during select call to non-blocking socket.")
4821 self.assertEqual(msg, MSG)
4823 def _testRecv(self):
4824 self.cli.connect((HOST, self.port))
4828 self.event.wait()
4831 self.cli.sendall(MSG)
4837 self.read_file is the io object returned by makefile() on
4841 self.write_file is the io object returned by makefile() on the
4856 def __init__(self, methodName='runTest'):
4857 SocketConnectedTest.__init__(self, methodName=methodName)
4859 def setUp(self):
4860 self.evt1, self.evt2, self.serv_finished, self.cli_finished = [
4862 SocketConnectedTest.setUp(self)
4863 self.read_file = self.cli_conn.makefile(
4864 self.read_mode, self.bufsize,
4865 encoding = self.encoding,
4866 errors = self.errors,
4867 newline = self.newline)
4869 def tearDown(self):
4870 self.serv_finished.set()
4871 self.read_file.close()
4872 self.assertTrue(self.read_file.closed)
4873 self.read_file = None
4874 SocketConnectedTest.tearDown(self)
4876 def clientSetUp(self):
4877 SocketConnectedTest.clientSetUp(self)
4878 self.write_file = self.serv_conn.makefile(
4879 self.write_mode, self.bufsize,
4880 encoding = self.encoding,
4881 errors = self.errors,
4882 newline = self.newline)
4884 def clientTearDown(self):
4885 self.cli_finished.set()
4886 self.write_file.close()
4887 self.assertTrue(self.write_file.closed)
4888 self.write_file = None
4889 SocketConnectedTest.clientTearDown(self)
4891 def testReadAfterTimeout(self):
4894 self.cli_conn.settimeout(1)
4895 self.read_file.read(3)
4897 self.assertRaises(TimeoutError, self.read_file.read, 1)
4899 with self.assertRaises(OSError) as ctx:
4900 self.read_file.read(1)
4901 self.assertIn("cannot read from timed out object", str(ctx.exception))
4903 def _testReadAfterTimeout(self):
4904 self.write_file.write(self.write_msg[0:3])
4905 self.write_file.flush()
4906 self.serv_finished.wait()
4908 def testSmallRead(self):
4910 first_seg = self.read_file.read(len(self.read_msg)-3)
4911 second_seg = self.read_file.read(3)
4913 self.assertEqual(msg, self.read_msg)
4915 def _testSmallRead(self):
4916 self.write_file.write(self.write_msg)
4917 self.write_file.flush()
4919 def testFullRead(self):
4921 msg = self.read_file.read()
4922 self.assertEqual(msg, self.read_msg)
4924 def _testFullRead(self):
4925 self.write_file.write(self.write_msg)
4926 self.write_file.close()
4928 def testUnbufferedRead(self):
4930 buf = type(self.read_msg)()
4932 char = self.read_file.read(1)
4936 self.assertEqual(buf, self.read_msg)
4938 def _testUnbufferedRead(self):
4939 self.write_file.write(self.write_msg)
4940 self.write_file.flush()
4942 def testReadline(self):
4944 line = self.read_file.readline()
4945 self.assertEqual(line, self.read_msg)
4947 def _testReadline(self):
4948 self.write_file.write(self.write_msg)
4949 self.write_file.flush()
4951 def testCloseAfterMakefile(self):
4953 self.cli_conn.close()
4955 msg = self.read_file.read()
4956 self.assertEqual(msg, self.read_msg)
4958 def _testCloseAfterMakefile(self):
4959 self.write_file.write(self.write_msg)
4960 self.write_file.flush()
4962 def testMakefileAfterMakefileClose(self):
4963 self.read_file.close()
4964 msg = self.cli_conn.recv(len(MSG))
4965 if isinstance(self.read_msg, str):
4967 self.assertEqual(msg, self.read_msg)
4969 def _testMakefileAfterMakefileClose(self):
4970 self.write_file.write(self.write_msg)
4971 self.write_file.flush()
4973 def testClosedAttr(self):
4974 self.assertTrue(not self.read_file.closed)
4976 def _testClosedAttr(self):
4977 self.assertTrue(not self.write_file.closed)
4979 def testAttributes(self):
4980 self.assertEqual(self.read_file.mode, self.read_mode)
4981 self.assertEqual(self.read_file.name, self.cli_conn.fileno())
4983 def _testAttributes(self):
4984 self.assertEqual(self.write_file.mode, self.write_mode)
4985 self.assertEqual(self.write_file.name, self.serv_conn.fileno())
4987 def testRealClose(self):
4988 self.read_file.close()
4989 self.assertRaises(ValueError, self.read_file.fileno)
4990 self.cli_conn.close()
4991 self.assertRaises(OSError, self.cli_conn.getsockname)
4993 def _testRealClose(self):
5009 def testUnbufferedReadline(self):
5011 line = self.read_file.readline() # first line
5012 self.assertEqual(line, b"A. " + self.write_msg) # first line
5013 self.read_file = self.cli_conn.makefile('rb', 0)
5014 line = self.read_file.readline() # second line
5015 self.assertEqual(line, b"B. " + self.write_msg) # second line
5017 def _testUnbufferedReadline(self):
5018 self.write_file.write(b"A. " + self.write_msg)
5019 self.write_file.write(b"B. " + self.write_msg)
5020 self.write_file.flush()
5022 def testMakefileClose(self):
5024 self.cli_conn.close()
5025 msg = self.cli_conn.recv(1024)
5026 self.assertEqual(msg, self.read_msg)
5028 self.read_file.close()
5029 self.assertRaises(OSError, self.cli_conn.recv, 1024)
5031 def _testMakefileClose(self):
5032 self.write_file.write(self.write_msg)
5033 self.write_file.flush()
5035 def testMakefileCloseSocketDestroy(self):
5036 refcount_before = sys.getrefcount(self.cli_conn)
5037 self.read_file.close()
5038 refcount_after = sys.getrefcount(self.cli_conn)
5039 self.assertEqual(refcount_before - 1, refcount_after)
5041 def _testMakefileCloseSocketDestroy(self):
5048 def testSmallReadNonBlocking(self):
5049 self.cli_conn.setblocking(False)
5050 self.assertEqual(self.read_file.readinto(bytearray(10)), None)
5051 self.assertEqual(self.read_file.read(len(self.read_msg) - 3), None)
5052 self.evt1.set()
5053 self.evt2.wait(1.0)
5054 first_seg = self.read_file.read(len(self.read_msg) - 3)
5058 first_seg = self.read_file.read(len(self.read_msg) - 3)
5060 n = self.read_file.readinto(buf)
5061 self.assertEqual(n, 3)
5063 self.assertEqual(msg, self.read_msg)
5064 self.assertEqual(self.read_file.readinto(bytearray(16)), None)
5065 self.assertEqual(self.read_file.read(1), None)
5067 def _testSmallReadNonBlocking(self):
5068 self.evt1.wait(1.0)
5069 self.write_file.write(self.write_msg)
5070 self.write_file.flush()
5071 self.evt2.set()
5074 self.serv_finished.wait(5.0)
5076 def testWriteNonBlocking(self):
5077 self.cli_finished.wait(5.0)
5080 if self.serv_skipped:
5081 self.skipTest(self.serv_skipped)
5083 def _testWriteNonBlocking(self):
5084 self.serv_skipped = None
5085 self.serv_conn.setblocking(False)
5090 n = self.write_file.write(BIG)
5091 self.assertGreater(n, 0)
5093 n = self.write_file.write(BIG)
5097 self.assertGreater(n, 0)
5102 self.serv_skipped = "failed to saturate the socket buffer"
5148 def clientSetUp(self):
5150 # BasicTCPTest, which defines self.port referenced below.
5151 self.cli = socket.create_connection((HOST, self.port))
5152 self.serv_conn = self.cli
5161 def connect(self, *args):
5165 def mocked_socket_module(self):
5168 socket.socket = self.MockSocket
5174 def test_connect(self):
5177 self.addCleanup(cli.close)
5178 with self.assertRaises(OSError) as cm:
5180 self.assertEqual(cm.exception.errno, errno.ECONNREFUSED)
5182 def test_create_connection(self):
5186 with self.assertRaises(OSError) as cm:
5204 self.assertIn(cm.exception.errno, expected_errnos)
5206 def test_create_connection_all_errors(self):
5213 self.fail('expected connection to fail')
5215 self.assertIsInstance(eg, ExceptionGroup)
5217 self.assertIsInstance(e, OSError)
5222 self.assertEqual(len(addresses), len(eg.exceptions))
5224 def test_create_connection_timeout(self):
5227 with self.mocked_socket_module():
5236 self.fail('TimeoutError not raised')
5241 def __init__(self, methodName='runTest'):
5242 SocketTCPTest.__init__(self, methodName=methodName)
5243 ThreadableTest.__init__(self)
5245 def clientSetUp(self):
5246 self.source_port = socket_helper.find_unused_port()
5248 def clientTearDown(self):
5249 self.cli.close()
5250 self.cli = None
5251 ThreadableTest.clientTearDown(self)
5253 def _justAccept(self):
5254 conn, addr = self.serv.accept()
5258 def _testFamily(self):
5259 self.cli = socket.create_connection((HOST, self.port),
5261 self.addCleanup(self.cli.close)
5262 self.assertEqual(self.cli.family, 2)
5265 def _testSourceAddress(self):
5266 self.cli = socket.create_connection((HOST, self.port),
5268 source_address=('', self.source_port))
5269 self.addCleanup(self.cli.close)
5270 self.assertEqual(self.cli.getsockname()[1], self.source_port)
5275 def _testTimeoutDefault(self):
5277 self.assertTrue(socket.getdefaulttimeout() is None)
5280 self.cli = socket.create_connection((HOST, self.port))
5281 self.addCleanup(self.cli.close)
5284 self.assertEqual(self.cli.gettimeout(), 42)
5287 def _testTimeoutNone(self):
5289 self.assertTrue(socket.getdefaulttimeout() is None)
5292 self.cli = socket.create_connection((HOST, self.port), timeout=None)
5293 self.addCleanup(self.cli.close)
5296 self.assertEqual(self.cli.gettimeout(), None)
5299 def _testTimeoutValueNamed(self):
5300 self.cli = socket.create_connection((HOST, self.port), timeout=30)
5301 self.assertEqual(self.cli.gettimeout(), 30)
5304 def _testTimeoutValueNonamed(self):
5305 self.cli = socket.create_connection((HOST, self.port), 30)
5306 self.addCleanup(self.cli.close)
5307 self.assertEqual(self.cli.gettimeout(), 30)
5312 def __init__(self, methodName='runTest'):
5313 SocketTCPTest.__init__(self, methodName=methodName)
5314 ThreadableTest.__init__(self)
5316 def clientSetUp(self):
5319 def clientTearDown(self):
5320 self.cli.close()
5321 self.cli = None
5322 ThreadableTest.clientTearDown(self)
5324 def testInsideTimeout(self):
5325 conn, addr = self.serv.accept()
5326 self.addCleanup(conn.close)
5331 def _testInsideTimeout(self):
5332 self.cli = sock = socket.create_connection((HOST, self.port))
5334 self.assertEqual(data, b"done!")
5336 def _testOutsideTimeout(self):
5337 self.cli = sock = socket.create_connection((HOST, self.port), timeout=1)
5338 self.assertRaises(TimeoutError, lambda: sock.recv(5))
5343 def testTCPTimeout(self):
5345 self.serv.settimeout(1.0)
5346 self.serv.accept()
5347 self.assertRaises(TimeoutError, raise_timeout,
5350 def testTimeoutZero(self):
5353 self.serv.settimeout(0.0)
5354 foo = self.serv.accept()
5356 self.fail("caught timeout instead of error (TCP)")
5360 self.fail("caught unexpected exception (TCP)")
5362 self.fail("accept() returned success when we did not expect it")
5366 def testInterruptedTimeout(self):
5370 self.serv.settimeout(5.0) # must be longer than alarm
5379 foo = self.serv.accept()
5381 self.fail("caught timeout instead of Alarm")
5385 self.fail("caught other exception instead of Alarm:"
5389 self.fail("nothing caught")
5393 self.fail("got Alarm in wrong place")
5400 def testUDPTimeout(self):
5402 self.serv.settimeout(1.0)
5403 self.serv.recv(1024)
5404 self.assertRaises(TimeoutError, raise_timeout,
5407 def testTimeoutZero(self):
5410 self.serv.settimeout(0.0)
5411 foo = self.serv.recv(1024)
5413 self.fail("caught timeout instead of error (UDP)")
5417 self.fail("caught unexpected exception (UDP)")
5419 self.fail("recv() returned success when we did not expect it")
5425 def testUDPLITETimeout(self):
5427 self.serv.settimeout(1.0)
5428 self.serv.recv(1024)
5429 self.assertRaises(TimeoutError, raise_timeout,
5432 def testTimeoutZero(self):
5435 self.serv.settimeout(0.0)
5436 foo = self.serv.recv(1024)
5438 self.fail("caught timeout instead of error (UDPLITE)")
5442 self.fail("caught unexpected exception (UDPLITE)")
5444 self.fail("recv() returned success when we did not expect it")
5448 def testExceptionTree(self):
5449 self.assertTrue(issubclass(OSError, Exception))
5450 self.assertTrue(issubclass(socket.herror, OSError))
5451 self.assertTrue(issubclass(socket.gaierror, OSError))
5452 self.assertTrue(issubclass(socket.timeout, OSError))
5453 self.assertIs(socket.error, OSError)
5454 self.assertIs(socket.timeout, TimeoutError)
5456 def test_setblocking_invalidfd(self):
5463 self.addCleanup(sock.detach)
5465 with self.assertRaises(OSError):
5474 def testLinuxAbstractNamespace(self):
5482 self.assertEqual(s1.getsockname(), address)
5483 self.assertEqual(s2.getpeername(), address)
5485 def testMaxName(self):
5486 address = b"\x00" + b"h" * (self.UNIX_PATH_MAX - 1)
5489 self.assertEqual(s.getsockname(), address)
5491 def testNameOverflow(self):
5492 address = "\x00" + "h" * self.UNIX_PATH_MAX
5494 self.assertRaises(OSError, s.bind, address)
5496 def testStrName(self):
5501 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00")
5505 def testBytearrayName(self):
5509 self.assertEqual(s.getsockname(), b"\x00python\x00test\x00")
5511 def testAutobind(self):
5517 self.assertRegex(s1.getsockname(), abstract_address)
5521 self.assertRegex(s2.getsockname(), abstract_address)
5522 self.assertNotEqual(s1.getsockname(), s2.getsockname())
5528 def setUp(self):
5529 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
5531 def tearDown(self):
5532 self.sock.close()
5534 def encoded(self, path):
5540 self.skipTest(
5545 def bind(self, sock, path):
5551 self.skipTest(
5557 def testUnbound(self):
5559 self.assertIn(self.sock.getsockname(), ('', None))
5561 def testStrAddr(self):
5564 self.bind(self.sock, path)
5565 self.addCleanup(os_helper.unlink, path)
5566 self.assertEqual(self.sock.getsockname(), path)
5568 def testBytesAddr(self):
5571 self.bind(self.sock, self.encoded(path))
5572 self.addCleanup(os_helper.unlink, path)
5573 self.assertEqual(self.sock.getsockname(), path)
5575 def testSurrogateescapeBind(self):
5579 b = self.encoded(path)
5580 self.bind(self.sock, b.decode("ascii", "surrogateescape"))
5581 self.addCleanup(os_helper.unlink, path)
5582 self.assertEqual(self.sock.getsockname(), path)
5584 def testUnencodableAddr(self):
5588 self.skipTest("No unencodable filename available")
5590 self.bind(self.sock, path)
5591 self.addCleanup(os_helper.unlink, path)
5592 self.assertEqual(self.sock.getsockname(), path)
5595 def testEmptyAddress(self):
5597 self.assertRaises(OSError, self.sock.bind, "")
5604 def __init__(self, methodName='runTest'):
5605 SocketConnectedTest.__init__(self, methodName=methodName)
5607 def testRecvIntoArray(self):
5609 nbytes = self.cli_conn.recv_into(buf)
5610 self.assertEqual(nbytes, len(MSG))
5613 self.assertEqual(msg, MSG)
5615 def _testRecvIntoArray(self):
5617 self.serv_conn.send(buf)
5619 def testRecvIntoBytearray(self):
5621 nbytes = self.cli_conn.recv_into(buf)
5622 self.assertEqual(nbytes, len(MSG))
5624 self.assertEqual(msg, MSG)
5628 def testRecvIntoMemoryview(self):
5630 nbytes = self.cli_conn.recv_into(memoryview(buf))
5631 self.assertEqual(nbytes, len(MSG))
5633 self.assertEqual(msg, MSG)
5637 def testRecvFromIntoArray(self):
5639 nbytes, addr = self.cli_conn.recvfrom_into(buf)
5640 self.assertEqual(nbytes, len(MSG))
5643 self.assertEqual(msg, MSG)
5645 def _testRecvFromIntoArray(self):
5647 self.serv_conn.send(buf)
5649 def testRecvFromIntoBytearray(self):
5651 nbytes, addr = self.cli_conn.recvfrom_into(buf)
5652 self.assertEqual(nbytes, len(MSG))
5654 self.assertEqual(msg, MSG)
5658 def testRecvFromIntoMemoryview(self):
5660 nbytes, addr = self.cli_conn.recvfrom_into(memoryview(buf))
5661 self.assertEqual(nbytes, len(MSG))
5663 self.assertEqual(msg, MSG)
5667 def testRecvFromIntoSmallBuffer(self):
5670 self.assertRaises(ValueError, self.cli_conn.recvfrom_into, buf, 1024)
5672 def _testRecvFromIntoSmallBuffer(self):
5673 self.serv_conn.send(MSG)
5675 def testRecvFromIntoEmptyBuffer(self):
5677 self.cli_conn.recvfrom_into(buf)
5678 self.cli_conn.recvfrom_into(buf, 0)
5710 def testRDM(self):
5713 self.addCleanup(srv.close)
5714 self.addCleanup(cli.close)
5727 self.assertEqual(cli.getsockname(), recvaddr)
5728 self.assertEqual(msg, MSG)
5734 def __init__(self, methodName = 'runTest'):
5735 unittest.TestCase.__init__(self, methodName = methodName)
5736 ThreadableTest.__init__(self)
5738 def setUp(self):
5739 self.srv = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM)
5740 self.addCleanup(self.srv.close)
5741 self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
5744 self.srv.bind(srvaddr)
5745 self.srv.listen()
5746 self.serverExplicitReady()
5747 self.conn, self.connaddr = self.srv.accept()
5748 self.addCleanup(self.conn.close)
5750 def clientSetUp(self):
5755 self.cli = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM)
5756 self.addCleanup(self.cli.close)
5759 self.cli.connect(addr)
5760 self.cliaddr = self.cli.getsockname()
5762 def testStream(self):
5763 msg = self.conn.recv(1024)
5764 self.assertEqual(msg, MSG)
5765 self.assertEqual(self.cliaddr, self.connaddr)
5767 def _testStream(self):
5768 self.cli.send(MSG)
5769 self.cli.close()
5774 def _testSocketClass(self):
5777 self.assertFalse(sock._closed)
5778 self.assertTrue(sock._closed)
5782 self.assertTrue(sock._closed)
5785 self.assertRaises(OSError, sock.sendall, b'foo')
5786 self.assertTrue(sock._closed)
5788 def testCreateConnectionBase(self):
5789 conn, addr = self.serv.accept()
5790 self.addCleanup(conn.close)
5794 def _testCreateConnectionBase(self):
5795 address = self.serv.getsockname()
5797 self.assertFalse(sock._closed)
5799 self.assertEqual(sock.recv(1024), b'foo')
5800 self.assertTrue(sock._closed)
5802 def testCreateConnectionClose(self):
5803 conn, addr = self.serv.accept()
5804 self.addCleanup(conn.close)
5808 def _testCreateConnectionClose(self):
5809 address = self.serv.getsockname()
5812 self.assertTrue(sock._closed)
5813 self.assertRaises(OSError, sock.sendall, b'foo')
5820 def test_SOCK_CLOEXEC(self):
5823 self.assertEqual(s.type, socket.SOCK_STREAM)
5824 self.assertFalse(s.get_inheritable())
5826 def test_default_inheritable(self):
5829 self.assertEqual(sock.get_inheritable(), False)
5831 def test_dup(self):
5837 self.assertEqual(newsock.get_inheritable(), False)
5839 def test_set_inheritable(self):
5843 self.assertEqual(sock.get_inheritable(), True)
5846 self.assertEqual(sock.get_inheritable(), False)
5849 def test_get_inheritable_cloexec(self):
5853 self.assertEqual(sock.get_inheritable(), False)
5860 self.assertEqual(sock.get_inheritable(), True)
5863 def test_set_inheritable_cloexec(self):
5867 self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
5871 self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
5875 def test_socketpair(self):
5877 self.addCleanup(s1.close)
5878 self.addCleanup(s2.close)
5879 self.assertEqual(s1.get_inheritable(), False)
5880 self.assertEqual(s2.get_inheritable(), False)
5886 def checkNonblock(self, s, nonblock=True, timeout=0.0):
5888 self.assertEqual(s.type, socket.SOCK_STREAM)
5889 self.assertEqual(s.gettimeout(), timeout)
5890 self.assertTrue(
5894 self.assertFalse(s.getblocking())
5903 self.assertTrue(s.getblocking())
5905 self.assertEqual(s.type, socket.SOCK_STREAM)
5906 self.assertEqual(s.gettimeout(), None)
5907 self.assertFalse(
5909 self.assertTrue(s.getblocking())
5912 def test_SOCK_NONBLOCK(self):
5917 self.checkNonblock(s)
5919 self.checkNonblock(s, nonblock=False)
5921 self.checkNonblock(s)
5923 self.checkNonblock(s, nonblock=False)
5925 self.checkNonblock(s, timeout=2.0)
5927 self.checkNonblock(s, nonblock=False)
5932 self.checkNonblock(s)
5935 self.checkNonblock(s, False)
5938 self.checkNonblock(s, timeout=2.0)
5941 self.checkNonblock(s, False)
5964 def testShare(self):
5970 p = multiprocessing.Process(target=self.remoteProcessServer, args=(q,))
5974 data = self.serv.share(p.pid)
5977 addr = self.serv.getsockname()
5978 self.serv.close()
5996 self.assertEqual(received, message)
5999 def testShareLength(self):
6000 data = self.serv.share(os.getpid())
6001 self.assertRaises(ValueError, socket.fromshare, data[:-1])
6002 self.assertRaises(ValueError, socket.fromshare, data+b"foo")
6004 def compareSockets(self, org, other):
6007 self.assertEqual(org.gettimeout(), None)
6008 self.assertEqual(org.gettimeout(), other.gettimeout())
6010 self.assertEqual(org.family, other.family)
6011 self.assertEqual(org.type, other.type)
6018 self.assertEqual(org.proto, other.proto)
6020 def testShareLocal(self):
6021 data = self.serv.share(os.getpid())
6024 self.compareSockets(self.serv, s)
6028 def testTypes(self):
6041 self.compareSockets(source, shared)
6081 def accept_conn(self):
6082 self.serv.settimeout(support.LONG_TIMEOUT)
6083 conn, addr = self.serv.accept()
6084 conn.settimeout(self.TIMEOUT)
6085 self.addCleanup(conn.close)
6088 def recv_data(self, conn):
6091 chunk = conn.recv(self.BUFSIZE)
6097 def meth_from_sock(self, sock):
6104 def _testRegularFile(self):
6105 address = self.serv.getsockname()
6108 meth = self.meth_from_sock(sock)
6110 self.assertEqual(sent, self.FILESIZE)
6111 self.assertEqual(file.tell(), self.FILESIZE)
6113 def testRegularFile(self):
6114 conn = self.accept_conn()
6115 data = self.recv_data(conn)
6116 self.assertEqual(len(data), self.FILESIZE)
6117 self.assertEqual(data, self.FILEDATA)
6121 def _testNonRegularFile(self):
6122 address = self.serv.getsockname()
6123 file = io.BytesIO(self.FILEDATA)
6126 self.assertEqual(sent, self.FILESIZE)
6127 self.assertEqual(file.tell(), self.FILESIZE)
6128 self.assertRaises(socket._GiveupOnSendfile,
6131 def testNonRegularFile(self):
6132 conn = self.accept_conn()
6133 data = self.recv_data(conn)
6134 self.assertEqual(len(data), self.FILESIZE)
6135 self.assertEqual(data, self.FILEDATA)
6139 def _testEmptyFileSend(self):
6140 address = self.serv.getsockname()
6143 self.addCleanup(os_helper.unlink, filename)
6146 meth = self.meth_from_sock(sock)
6148 self.assertEqual(sent, 0)
6149 self.assertEqual(file.tell(), 0)
6151 def testEmptyFileSend(self):
6152 conn = self.accept_conn()
6153 data = self.recv_data(conn)
6154 self.assertEqual(data, b"")
6158 def _testOffset(self):
6159 address = self.serv.getsockname()
6162 meth = self.meth_from_sock(sock)
6164 self.assertEqual(sent, self.FILESIZE - 5000)
6165 self.assertEqual(file.tell(), self.FILESIZE)
6167 def testOffset(self):
6168 conn = self.accept_conn()
6169 data = self.recv_data(conn)
6170 self.assertEqual(len(data), self.FILESIZE - 5000)
6171 self.assertEqual(data, self.FILEDATA[5000:])
6175 def _testCount(self):
6176 address = self.serv.getsockname()
6182 meth = self.meth_from_sock(sock)
6184 self.assertEqual(sent, count)
6185 self.assertEqual(file.tell(), count)
6187 def testCount(self):
6189 conn = self.accept_conn()
6190 data = self.recv_data(conn)
6191 self.assertEqual(len(data), count)
6192 self.assertEqual(data, self.FILEDATA[:count])
6196 def _testCountSmall(self):
6197 address = self.serv.getsockname()
6203 meth = self.meth_from_sock(sock)
6205 self.assertEqual(sent, count)
6206 self.assertEqual(file.tell(), count)
6208 def testCountSmall(self):
6210 conn = self.accept_conn()
6211 data = self.recv_data(conn)
6212 self.assertEqual(len(data), count)
6213 self.assertEqual(data, self.FILEDATA[:count])
6217 def _testCountWithOffset(self):
6218 address = self.serv.getsockname()
6222 meth = self.meth_from_sock(sock)
6224 self.assertEqual(sent, count)
6225 self.assertEqual(file.tell(), count + 2007)
6227 def testCountWithOffset(self):
6229 conn = self.accept_conn()
6230 data = self.recv_data(conn)
6231 self.assertEqual(len(data), count)
6232 self.assertEqual(data, self.FILEDATA[2007:count+2007])
6236 def _testNonBlocking(self):
6237 address = self.serv.getsockname()
6241 meth = self.meth_from_sock(sock)
6242 self.assertRaises(ValueError, meth, file)
6243 self.assertRaises(ValueError, sock.sendfile, file)
6245 def testNonBlocking(self):
6246 conn = self.accept_conn()
6248 self.fail('was not supposed to receive any data')
6252 def _testWithTimeout(self):
6253 address = self.serv.getsockname()
6258 meth = self.meth_from_sock(sock)
6260 self.assertEqual(sent, self.FILESIZE)
6262 def testWithTimeout(self):
6263 conn = self.accept_conn()
6264 data = self.recv_data(conn)
6265 self.assertEqual(len(data), self.FILESIZE)
6266 self.assertEqual(data, self.FILEDATA)
6270 def _testWithTimeoutTriggeredSend(self):
6271 address = self.serv.getsockname()
6275 meth = self.meth_from_sock(sock)
6276 self.assertRaises(TimeoutError, meth, file)
6278 def testWithTimeoutTriggeredSend(self):
6279 conn = self.accept_conn()
6286 def _test_errors(self):
6289 def test_errors(self):
6292 meth = self.meth_from_sock(s)
6293 self.assertRaisesRegex(
6297 meth = self.meth_from_sock(s)
6298 self.assertRaisesRegex(
6302 meth = self.meth_from_sock(s)
6303 self.assertRaisesRegex(TypeError, "positive integer",
6305 self.assertRaisesRegex(TypeError, "positive integer",
6307 self.assertRaisesRegex(ValueError, "positive integer",
6309 self.assertRaisesRegex(ValueError, "positive integer",
6319 def meth_from_sock(self, sock):
6326 def create_alg(self, typ, name):
6340 def test_sha256(self):
6343 with self.create_alg('hash', 'sha256') as algo:
6347 self.assertEqual(op.recv(512), expected)
6355 self.assertEqual(op.recv(512), expected)
6357 def test_hmac_sha1(self):
6359 with self.create_alg('hash', 'hmac(sha1)') as algo:
6364 self.assertEqual(op.recv(512), expected)
6369 def test_aes_cbc(self):
6375 with self.create_alg('skcipher', 'cbc(aes)') as algo:
6382 self.assertEqual(op.recv(msglen), ciphertext)
6388 self.assertEqual(op.recv(msglen), msg)
6398 self.assertEqual(len(enc), msglen * multiplier)
6399 self.assertEqual(enc[:msglen], ciphertext)
6406 self.assertEqual(len(dec), msglen * multiplier)
6407 self.assertEqual(dec, msg * multiplier)
6410 def test_aead_aes_gcm(self):
6421 with self.create_alg('aead', 'gcm(aes)') as algo:
6434 self.assertEqual(expected_ct, res[assoclen:-taglen])
6435 self.assertEqual(expected_tag, res[-taglen:])
6444 self.assertEqual(expected_ct, res[assoclen:-taglen])
6445 self.assertEqual(expected_tag, res[-taglen:])
6460 self.assertEqual(expected_ct, res[assoclen:-taglen])
6461 self.assertEqual(expected_tag, res[-taglen:])
6470 self.assertEqual(plain, res[assoclen:])
6473 def test_drbg_pr_sha256(self):
6475 with self.create_alg('rng', 'drbg_pr_sha256') as algo:
6481 self.assertEqual(len(rn), 32)
6483 def test_sendmsg_afalg_args(self):
6486 with self.assertRaises(TypeError):
6489 with self.assertRaises(TypeError):
6492 with self.assertRaises(TypeError):
6495 with self.assertRaises(TypeError):
6498 with self.assertRaises(TypeError):
6501 def test_length_restriction(self):
6504 self.addCleanup(sock.close)
6507 with self.assertRaises(FileNotFoundError):
6509 with self.assertRaisesRegex(ValueError, "type too long"):
6513 with self.assertRaises(FileNotFoundError):
6515 with self.assertRaisesRegex(ValueError, "name too long"):
6521 def test_tcp_keepalive(self):
6522 self.assertTrue(socket.TCP_KEEPALIVE)
6540 def test_new_tcp_flags(self):
6542 unknown = [s for s in provided if s not in self.knownTCPFlags]
6544 self.assertEqual([], unknown,
6550 def test_address(self):
6553 self.assertEqual(sock.getsockname()[0], "127.0.0.1")
6554 self.assertEqual(sock.getsockname()[1], port)
6558 self.assertEqual(sock.getsockname()[0], "::1")
6559 self.assertEqual(sock.getsockname()[1], port)
6561 def test_family_and_type(self):
6563 self.assertEqual(sock.family, socket.AF_INET)
6564 self.assertEqual(sock.type, socket.SOCK_STREAM)
6567 self.assertEqual(s.family, socket.AF_INET6)
6568 self.assertEqual(sock.type, socket.SOCK_STREAM)
6570 def test_reuse_port(self):
6572 with self.assertRaises(ValueError):
6577 self.assertEqual(opt, 0)
6580 self.assertNotEqual(opt, 0)
6586 def test_ipv6_only_default(self):
6593 def test_dualstack_ipv6_family(self):
6596 self.assertEqual(sock.family, socket.AF_INET6)
6602 def echo_server(self, sock):
6607 event.wait(self.timeout)
6614 sock.settimeout(self.timeout)
6617 self.addCleanup(thread.join, self.timeout)
6620 def echo_client(self, addr, family):
6622 sock.settimeout(self.timeout)
6625 self.assertEqual(sock.recv(1024), b'foo')
6627 def test_tcp4(self):
6630 self.echo_server(sock)
6631 self.echo_client(("127.0.0.1", port), socket.AF_INET)
6634 def test_tcp6(self):
6638 self.echo_server(sock)
6639 self.echo_client(("::1", port), socket.AF_INET6)
6646 def test_dual_stack_client_v4(self):
6650 self.echo_server(sock)
6651 self.echo_client(("127.0.0.1", port), socket.AF_INET)
6656 def test_dual_stack_client_v6(self):
6660 self.echo_server(sock)
6661 self.echo_client(("::1", port), socket.AF_INET6)
6667 def testSendAndRecvFds(self):
6679 self.addCleanup(close_pipes, pipes)
6688 self.addCleanup(close_fds, fds2)
6690 self.assertEqual(msg, MSG)
6691 self.assertEqual(len(fds2), len(fds))
6692 self.assertEqual(flags, 0)
6702 self.assertEqual(data, str(index).encode())