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

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

Issue 16510: Using appropriate checks in tests
Patch Set: Created 5 years, 7 months ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
View unified diff | Download patch
« no previous file with comments | « Lib/test/test_slice.py ('k') | Lib/test/test_ssl.py » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 import unittest 1 import unittest
2 from test import support 2 from test import support
3 3
4 import errno 4 import errno
5 import io 5 import io
6 import socket 6 import socket
7 import select 7 import select
8 import tempfile 8 import tempfile
9 import time 9 import time
10 import traceback 10 import traceback
(...skipping 725 matching lines...) Expand 10 before | Expand all | Expand 10 after
736 socket.SO_REUSEADDR 736 socket.SO_REUSEADDR
737 737
738 def testHostnameRes(self): 738 def testHostnameRes(self):
739 # Testing hostname resolution mechanisms 739 # Testing hostname resolution mechanisms
740 hostname = socket.gethostname() 740 hostname = socket.gethostname()
741 try: 741 try:
742 ip = socket.gethostbyname(hostname) 742 ip = socket.gethostbyname(hostname)
743 except OSError: 743 except OSError:
744 # Probably name lookup wasn't set up right; skip this test 744 # Probably name lookup wasn't set up right; skip this test
745 self.skipTest('name lookup failure') 745 self.skipTest('name lookup failure')
746 self.assertTrue(ip.find('.') >= 0, "Error resolving host to ip.") 746 self.assertIn('.', ip, "Error resolving host to ip.")
747 try: 747 try:
748 hname, aliases, ipaddrs = socket.gethostbyaddr(ip) 748 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
749 except OSError: 749 except OSError:
750 # Probably a similar problem as above; skip this test 750 # Probably a similar problem as above; skip this test
751 self.skipTest('name lookup failure') 751 self.skipTest('name lookup failure')
752 all_host_names = [hostname, hname] + aliases 752 all_host_names = [hostname, hname] + aliases
753 fqhn = socket.getfqdn(ip) 753 fqhn = socket.getfqdn(ip)
754 if not fqhn in all_host_names: 754 if not fqhn in all_host_names:
755 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names))) 755 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names)))
756 756
(...skipping 349 matching lines...) Expand 10 before | Expand all | Expand 10 after
1106 self.skipTest('name lookup failure') 1106 self.skipTest('name lookup failure')
1107 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0]) 1107 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0])
1108 self.assertEqual(name[1], port) 1108 self.assertEqual(name[1], port)
1109 1109
1110 def testGetSockOpt(self): 1110 def testGetSockOpt(self):
1111 # Testing getsockopt() 1111 # Testing getsockopt()
1112 # We know a socket should start without reuse==0 1112 # We know a socket should start without reuse==0
1113 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1113 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1114 self.addCleanup(sock.close) 1114 self.addCleanup(sock.close)
1115 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1115 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1116 self.assertFalse(reuse != 0, "initial mode is reuse") 1116 self.assertEqual(reuse, 0, "initial mode is reuse")
1117 1117
1118 def testSetSockOpt(self): 1118 def testSetSockOpt(self):
1119 # Testing setsockopt() 1119 # Testing setsockopt()
1120 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1120 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1121 self.addCleanup(sock.close) 1121 self.addCleanup(sock.close)
1122 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 1122 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1123 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1123 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1124 self.assertFalse(reuse == 0, "failed to set reuse mode") 1124 self.assertNotEqual(reuse, 0, "failed to set reuse mode")
1125 1125
1126 def testSendAfterClose(self): 1126 def testSendAfterClose(self):
1127 # testing send() after close() with timeout 1127 # testing send() after close() with timeout
1128 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1128 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1129 sock.settimeout(1) 1129 sock.settimeout(1)
1130 sock.close() 1130 sock.close()
1131 self.assertRaises(OSError, sock.send, b"spam") 1131 self.assertRaises(OSError, sock.send, b"spam")
1132 1132
1133 def testNewAttributes(self): 1133 def testNewAttributes(self):
1134 # testing .family, .type and .protocol 1134 # testing .family, .type and .protocol
(...skipping 2603 matching lines...) Expand 10 before | Expand all | Expand 10 after
3738 self.serv.setblocking(True) 3738 self.serv.setblocking(True)
3739 self.assertIsNone(self.serv.gettimeout()) 3739 self.assertIsNone(self.serv.gettimeout())
3740 self.serv.setblocking(False) 3740 self.serv.setblocking(False)
3741 self.assertEqual(self.serv.gettimeout(), 0.0) 3741 self.assertEqual(self.serv.gettimeout(), 0.0)
3742 start = time.time() 3742 start = time.time()
3743 try: 3743 try:
3744 self.serv.accept() 3744 self.serv.accept()
3745 except OSError: 3745 except OSError:
3746 pass 3746 pass
3747 end = time.time() 3747 end = time.time()
3748 self.assertTrue((end - start) < 1.0, "Error setting non-blocking mode.") 3748 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.")
3749 3749
3750 def _testSetBlocking(self): 3750 def _testSetBlocking(self):
3751 pass 3751 pass
3752 3752
3753 @support.cpython_only 3753 @support.cpython_only
3754 def testSetBlocking_overflow(self): 3754 def testSetBlocking_overflow(self):
3755 # Issue 15989 3755 # Issue 15989
3756 import _testcapi 3756 import _testcapi
3757 if _testcapi.UINT_MAX >= _testcapi.ULONG_MAX: 3757 if _testcapi.UINT_MAX >= _testcapi.ULONG_MAX:
3758 self.skipTest('needs UINT_MAX < ULONG_MAX') 3758 self.skipTest('needs UINT_MAX < ULONG_MAX')
(...skipping 14 matching lines...) Expand all
3773 socket.SOCK_NONBLOCK) 3773 socket.SOCK_NONBLOCK)
3774 self.port = support.bind_port(self.serv) 3774 self.port = support.bind_port(self.serv)
3775 self.serv.listen(1) 3775 self.serv.listen(1)
3776 # actual testing 3776 # actual testing
3777 start = time.time() 3777 start = time.time()
3778 try: 3778 try:
3779 self.serv.accept() 3779 self.serv.accept()
3780 except OSError: 3780 except OSError:
3781 pass 3781 pass
3782 end = time.time() 3782 end = time.time()
3783 self.assertTrue((end - start) < 1.0, "Error creating with non-blocking m ode.") 3783 self.assertLess(end - start, 1.0, "Error creating with non-blocking mode .")
3784 3784
3785 def _testInitNonBlocking(self): 3785 def _testInitNonBlocking(self):
3786 pass 3786 pass
3787 3787
3788 def testInheritFlags(self): 3788 def testInheritFlags(self):
3789 # Issue #7995: when calling accept() on a listening socket with a 3789 # Issue #7995: when calling accept() on a listening socket with a
3790 # timeout, the resulting socket should not be non-blocking. 3790 # timeout, the resulting socket should not be non-blocking.
3791 self.serv.settimeout(10) 3791 self.serv.settimeout(10)
3792 try: 3792 try:
3793 conn, addr = self.serv.accept() 3793 conn, addr = self.serv.accept()
(...skipping 585 matching lines...) Expand 10 before | Expand all | Expand 10 after
4379 self.cli = socket.create_connection((HOST, self.port), timeout=30, 4379 self.cli = socket.create_connection((HOST, self.port), timeout=30,
4380 source_address=('', self.source_port)) 4380 source_address=('', self.source_port))
4381 self.addCleanup(self.cli.close) 4381 self.addCleanup(self.cli.close)
4382 self.assertEqual(self.cli.getsockname()[1], self.source_port) 4382 self.assertEqual(self.cli.getsockname()[1], self.source_port)
4383 # The port number being used is sufficient to show that the bind() 4383 # The port number being used is sufficient to show that the bind()
4384 # call happened. 4384 # call happened.
4385 4385
4386 testTimeoutDefault = _justAccept 4386 testTimeoutDefault = _justAccept
4387 def _testTimeoutDefault(self): 4387 def _testTimeoutDefault(self):
4388 # passing no explicit timeout uses socket's global default 4388 # passing no explicit timeout uses socket's global default
4389 self.assertTrue(socket.getdefaulttimeout() is None) 4389 self.assertIsNone(socket.getdefaulttimeout())
4390 socket.setdefaulttimeout(42) 4390 socket.setdefaulttimeout(42)
4391 try: 4391 try:
4392 self.cli = socket.create_connection((HOST, self.port)) 4392 self.cli = socket.create_connection((HOST, self.port))
4393 self.addCleanup(self.cli.close) 4393 self.addCleanup(self.cli.close)
4394 finally: 4394 finally:
4395 socket.setdefaulttimeout(None) 4395 socket.setdefaulttimeout(None)
4396 self.assertEqual(self.cli.gettimeout(), 42) 4396 self.assertEqual(self.cli.gettimeout(), 42)
4397 4397
4398 testTimeoutNone = _justAccept 4398 testTimeoutNone = _justAccept
4399 def _testTimeoutNone(self): 4399 def _testTimeoutNone(self):
4400 # None timeout means the same as sock.settimeout(None) 4400 # None timeout means the same as sock.settimeout(None)
4401 self.assertTrue(socket.getdefaulttimeout() is None) 4401 self.assertIsNone(socket.getdefaulttimeout())
4402 socket.setdefaulttimeout(30) 4402 socket.setdefaulttimeout(30)
4403 try: 4403 try:
4404 self.cli = socket.create_connection((HOST, self.port), timeout=None) 4404 self.cli = socket.create_connection((HOST, self.port), timeout=None)
4405 self.addCleanup(self.cli.close) 4405 self.addCleanup(self.cli.close)
4406 finally: 4406 finally:
4407 socket.setdefaulttimeout(None) 4407 socket.setdefaulttimeout(None)
4408 self.assertEqual(self.cli.gettimeout(), None) 4408 self.assertEqual(self.cli.gettimeout(), None)
4409 4409
4410 testTimeoutValueNamed = _justAccept 4410 testTimeoutValueNamed = _justAccept
4411 def _testTimeoutValueNamed(self): 4411 def _testTimeoutValueNamed(self):
(...skipping 709 matching lines...) Expand 10 before | Expand all | Expand 10 after
5121 InterruptedSendTimeoutTest, 5121 InterruptedSendTimeoutTest,
5122 TestSocketSharing, 5122 TestSocketSharing,
5123 ]) 5123 ])
5124 5124
5125 thread_info = support.threading_setup() 5125 thread_info = support.threading_setup()
5126 support.run_unittest(*tests) 5126 support.run_unittest(*tests)
5127 support.threading_cleanup(*thread_info) 5127 support.threading_cleanup(*thread_info)
5128 5128
5129 if __name__ == "__main__": 5129 if __name__ == "__main__":
5130 test_main() 5130 test_main()
OLDNEW
« no previous file with comments | « Lib/test/test_slice.py ('k') | Lib/test/test_ssl.py » ('j') | no next file with comments »

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