diff options
Diffstat (limited to 'Lib/test/test_socket.py')
| -rw-r--r-- | Lib/test/test_socket.py | 81 |
1 files changed, 50 insertions, 31 deletions
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index baca4c1..ffa2587 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -18,23 +18,12 @@ import contextlib from weakref import proxy import signal import math +import pickle try: import fcntl except ImportError: fcntl = False -def try_address(host, port=0, family=socket.AF_INET): - """Try to bind a socket on the given host:port and return True - if that has been possible.""" - try: - sock = socket.socket(family, socket.SOCK_STREAM) - sock.bind((host, port)) - except (socket.error, socket.gaierror): - return False - else: - sock.close() - return True - def linux_version(): try: # platform.release() is something like '2.6.33.7-desktop-2mnb' @@ -44,8 +33,7 @@ def linux_version(): return 0, 0, 0 HOST = support.HOST -MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf8') ## test unicode string and carriage return -SUPPORTS_IPV6 = socket.has_ipv6 and try_address('::1', family=socket.AF_INET6) +MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8') ## test unicode string and carriage return try: import _thread as thread @@ -282,18 +270,13 @@ class GeneralModuleTests(unittest.TestCase): def testSocketError(self): # Testing socket module exceptions - def raise_error(*args, **kwargs): + msg = "Error raising socket exception (%s)." + with self.assertRaises(socket.error, msg=msg % 'socket.error'): raise socket.error - def raise_herror(*args, **kwargs): + with self.assertRaises(socket.error, msg=msg % 'socket.herror'): raise socket.herror - def raise_gaierror(*args, **kwargs): + with self.assertRaises(socket.error, msg=msg % 'socket.gaierror'): raise socket.gaierror - self.assertRaises(socket.error, raise_error, - "Error raising socket exception.") - self.assertRaises(socket.error, raise_herror, - "Error raising socket exception.") - self.assertRaises(socket.error, raise_gaierror, - "Error raising socket exception.") def testSendtoErrors(self): # Testing that sendto doens't masks failures. See #10169. @@ -369,6 +352,36 @@ class GeneralModuleTests(unittest.TestCase): if not fqhn in all_host_names: self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names))) + @unittest.skipUnless(hasattr(socket, 'sethostname'), "test needs socket.sethostname()") + @unittest.skipUnless(hasattr(socket, 'gethostname'), "test needs socket.gethostname()") + def test_sethostname(self): + oldhn = socket.gethostname() + try: + socket.sethostname('new') + except socket.error as e: + if e.errno == errno.EPERM: + self.skipTest("test should be run as root") + else: + raise + try: + # running test as root! + self.assertEqual(socket.gethostname(), 'new') + # Should work with bytes objects too + socket.sethostname(b'bar') + self.assertEqual(socket.gethostname(), 'bar') + finally: + socket.sethostname(oldhn) + + @unittest.skipUnless(hasattr(socket, 'if_nameindex'), + 'socket.if_nameindex() not available.') + def testInterfaceNameIndex(self): + interfaces = socket.if_nameindex() + for index, name in interfaces: + # interface indices are non-zero integers + self.assertGreater(index, 0) + self.assertEqual(index, socket.if_nametoindex(name)) + self.assertEqual(name, socket.if_indextoname(index)) + def testRefCountGetNameInfo(self): # Testing reference count for getnameinfo if hasattr(sys, "getrefcount"): @@ -654,7 +667,7 @@ class GeneralModuleTests(unittest.TestCase): socket.getaddrinfo('localhost', 80) socket.getaddrinfo('127.0.0.1', 80) socket.getaddrinfo(None, 80) - if SUPPORTS_IPV6: + if support.IPV6_ENABLED: socket.getaddrinfo('::1', 80) # port can be a string service name such as "http", a numeric # port number or None @@ -774,7 +787,13 @@ class GeneralModuleTests(unittest.TestCase): fp.close() self.assertEqual(repr(fp), "<_io.BufferedReader name=-1>") - def testListenBacklog0(self): + def test_pickle(self): + sock = socket.socket() + with sock: + for protocol in range(pickle.HIGHEST_PROTOCOL + 1): + self.assertRaises(TypeError, pickle.dumps, sock, protocol) + + def test_listen_backlog0(self): srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.bind((HOST, 0)) # backlog = 0 @@ -1102,7 +1121,7 @@ class FileObjectClassTestCase(SocketConnectedTest): """ bufsize = -1 # Use default buffer size - encoding = 'utf8' + encoding = 'utf-8' errors = 'strict' newline = None @@ -1323,7 +1342,7 @@ class FileObjectInterruptedTestCase(unittest.TestCase): data = b'' else: data = '' - expecting = expecting.decode('utf8') + expecting = expecting.decode('utf-8') while len(data) != len(expecting): part = fo.read(size) if not part: @@ -1485,7 +1504,7 @@ class UnicodeReadFileObjectClassTestCase(FileObjectClassTestCase): """Tests for socket.makefile() in text mode (rather than binary)""" read_mode = 'r' - read_msg = MSG.decode('utf8') + read_msg = MSG.decode('utf-8') write_mode = 'wb' write_msg = MSG newline = '' @@ -1497,7 +1516,7 @@ class UnicodeWriteFileObjectClassTestCase(FileObjectClassTestCase): read_mode = 'rb' read_msg = MSG write_mode = 'w' - write_msg = MSG.decode('utf8') + write_msg = MSG.decode('utf-8') newline = '' @@ -1505,9 +1524,9 @@ class UnicodeReadWriteFileObjectClassTestCase(FileObjectClassTestCase): """Tests for socket.makefile() in text mode (rather than binary)""" read_mode = 'r' - read_msg = MSG.decode('utf8') + read_msg = MSG.decode('utf-8') write_mode = 'w' - write_msg = MSG.decode('utf8') + write_msg = MSG.decode('utf-8') newline = '' |