summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_socket.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_socket.py')
-rw-r--r--Lib/test/test_socket.py45
1 files changed, 44 insertions, 1 deletions
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 365da36..89cf1fa 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -20,6 +20,7 @@ import math
import pickle
import struct
import random
+import shutil
import string
import _thread as thread
import threading
@@ -1284,6 +1285,7 @@ class GeneralModuleTests(unittest.TestCase):
def testCloseException(self):
sock = socket.socket()
+ sock.bind((socket._LOCALHOST, 0))
socket.socket(fileno=sock.fileno()).close()
try:
sock.close()
@@ -1636,9 +1638,11 @@ class GeneralModuleTests(unittest.TestCase):
) + 1
with socket.socket(
- family=unknown_family, type=unknown_type, fileno=fd) as s:
+ family=unknown_family, type=unknown_type, proto=23,
+ fileno=fd) as s:
self.assertEqual(s.family, unknown_family)
self.assertEqual(s.type, unknown_type)
+ self.assertEqual(s.proto, 23)
@unittest.skipUnless(hasattr(os, 'sendfile'), 'test needs os.sendfile()')
def test__sendfile_use_sendfile(self):
@@ -1658,6 +1662,45 @@ class GeneralModuleTests(unittest.TestCase):
with self.assertRaises(TypeError):
sock._sendfile_use_sendfile(File(None))
+ def _test_socket_fileno(self, s, family, stype):
+ self.assertEqual(s.family, family)
+ self.assertEqual(s.type, stype)
+
+ fd = s.fileno()
+ s2 = socket.socket(fileno=fd)
+ self.addCleanup(s2.close)
+ # detach old fd to avoid double close
+ s.detach()
+ self.assertEqual(s2.family, family)
+ self.assertEqual(s2.type, stype)
+ self.assertEqual(s2.fileno(), fd)
+
+ def test_socket_fileno(self):
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.addCleanup(s.close)
+ s.bind((support.HOST, 0))
+ self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_STREAM)
+
+ if hasattr(socket, "SOCK_DGRAM"):
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ self.addCleanup(s.close)
+ s.bind((support.HOST, 0))
+ self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_DGRAM)
+
+ if support.IPV6_ENABLED:
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+ self.addCleanup(s.close)
+ s.bind((support.HOSTv6, 0, 0, 0))
+ self._test_socket_fileno(s, socket.AF_INET6, socket.SOCK_STREAM)
+
+ if hasattr(socket, "AF_UNIX"):
+ tmpdir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, tmpdir)
+ s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ self.addCleanup(s.close)
+ s.bind(os.path.join(tmpdir, 'socket'))
+ self._test_socket_fileno(s, socket.AF_UNIX, socket.SOCK_STREAM)
+
@unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.')
class BasicCANTest(unittest.TestCase):