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

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

Issue 16510: Using appropriate checks in tests
Patch Set: Created 6 years, 9 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_smtplib.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 #!/usr/bin/env python3 1 #!/usr/bin/env python3
2 2
3 import unittest 3 import unittest
4 from test import support 4 from test import support
5 from unittest.case import _ExpectedFailure 5 from unittest.case import _ExpectedFailure
6 6
7 import errno 7 import errno
8 import io 8 import io
9 import socket 9 import socket
10 import select 10 import select
(...skipping 697 matching lines...) Expand 10 before | Expand all | Expand 10 after
708 socket.SO_REUSEADDR 708 socket.SO_REUSEADDR
709 709
710 def testHostnameRes(self): 710 def testHostnameRes(self):
711 # Testing hostname resolution mechanisms 711 # Testing hostname resolution mechanisms
712 hostname = socket.gethostname() 712 hostname = socket.gethostname()
713 try: 713 try:
714 ip = socket.gethostbyname(hostname) 714 ip = socket.gethostbyname(hostname)
715 except OSError: 715 except OSError:
716 # Probably name lookup wasn't set up right; skip this test 716 # Probably name lookup wasn't set up right; skip this test
717 return 717 return
718 self.assertTrue(ip.find('.') >= 0, "Error resolving host to ip.") 718 self.assertGreaterEqual(ip.find('.'), 0, "Error resolving host to ip.")
719 try: 719 try:
720 hname, aliases, ipaddrs = socket.gethostbyaddr(ip) 720 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
721 except OSError: 721 except OSError:
722 # Probably a similar problem as above; skip this test 722 # Probably a similar problem as above; skip this test
723 return 723 return
724 all_host_names = [hostname, hname] + aliases 724 all_host_names = [hostname, hname] + aliases
725 fqhn = socket.getfqdn(ip) 725 fqhn = socket.getfqdn(ip)
726 if not fqhn in all_host_names: 726 if not fqhn in all_host_names:
727 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names))) 727 self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names)))
728 728
(...skipping 318 matching lines...) Expand 10 before | Expand all | Expand 10 after
1047 return 1047 return
1048 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0]) 1048 self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0])
1049 self.assertEqual(name[1], port) 1049 self.assertEqual(name[1], port)
1050 1050
1051 def testGetSockOpt(self): 1051 def testGetSockOpt(self):
1052 # Testing getsockopt() 1052 # Testing getsockopt()
1053 # We know a socket should start without reuse==0 1053 # We know a socket should start without reuse==0
1054 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1054 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1055 self.addCleanup(sock.close) 1055 self.addCleanup(sock.close)
1056 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1056 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1057 self.assertFalse(reuse != 0, "initial mode is reuse") 1057 self.assertEqual(reuse, 0, "initial mode is reuse")
1058 1058
1059 def testSetSockOpt(self): 1059 def testSetSockOpt(self):
1060 # Testing setsockopt() 1060 # Testing setsockopt()
1061 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1061 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1062 self.addCleanup(sock.close) 1062 self.addCleanup(sock.close)
1063 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 1063 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1064 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 1064 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
1065 self.assertFalse(reuse == 0, "failed to set reuse mode") 1065 self.assertNotEqual(reuse, 0, "failed to set reuse mode")
1066 1066
1067 def testSendAfterClose(self): 1067 def testSendAfterClose(self):
1068 # testing send() after close() with timeout 1068 # testing send() after close() with timeout
1069 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1069 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1070 sock.settimeout(1) 1070 sock.settimeout(1)
1071 sock.close() 1071 sock.close()
1072 self.assertRaises(OSError, sock.send, b"spam") 1072 self.assertRaises(OSError, sock.send, b"spam")
1073 1073
1074 def testNewAttributes(self): 1074 def testNewAttributes(self):
1075 # testing .family, .type and .protocol 1075 # testing .family, .type and .protocol
(...skipping 2493 matching lines...) Expand 10 before | Expand all | Expand 10 after
3569 self.serv.setblocking(True) 3569 self.serv.setblocking(True)
3570 self.assertIsNone(self.serv.gettimeout()) 3570 self.assertIsNone(self.serv.gettimeout())
3571 self.serv.setblocking(False) 3571 self.serv.setblocking(False)
3572 self.assertEqual(self.serv.gettimeout(), 0.0) 3572 self.assertEqual(self.serv.gettimeout(), 0.0)
3573 start = time.time() 3573 start = time.time()
3574 try: 3574 try:
3575 self.serv.accept() 3575 self.serv.accept()
3576 except OSError: 3576 except OSError:
3577 pass 3577 pass
3578 end = time.time() 3578 end = time.time()
3579 self.assertTrue((end - start) < 1.0, "Error setting non-blocking mode.") 3579 self.assertLess(end - start, 1.0, "Error setting non-blocking mode.")
3580 # Issue 15989 3580 # Issue 15989
3581 if _testcapi.UINT_MAX < _testcapi.ULONG_MAX: 3581 if _testcapi.UINT_MAX < _testcapi.ULONG_MAX:
3582 self.serv.setblocking(_testcapi.UINT_MAX + 1) 3582 self.serv.setblocking(_testcapi.UINT_MAX + 1)
3583 self.assertIsNone(self.serv.gettimeout()) 3583 self.assertIsNone(self.serv.gettimeout())
3584 3584
3585 def _testSetBlocking(self): 3585 def _testSetBlocking(self):
3586 pass 3586 pass
3587 3587
3588 if hasattr(socket, "SOCK_NONBLOCK"): 3588 if hasattr(socket, "SOCK_NONBLOCK"):
3589 @support.requires_linux_version(2, 6, 28) 3589 @support.requires_linux_version(2, 6, 28)
3590 def testInitNonBlocking(self): 3590 def testInitNonBlocking(self):
3591 # reinit server socket 3591 # reinit server socket
3592 self.serv.close() 3592 self.serv.close()
3593 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM | 3593 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM |
3594 socket.SOCK_NONBLOCK) 3594 socket.SOCK_NONBLOCK)
3595 self.port = support.bind_port(self.serv) 3595 self.port = support.bind_port(self.serv)
3596 self.serv.listen(1) 3596 self.serv.listen(1)
3597 # actual testing 3597 # actual testing
3598 start = time.time() 3598 start = time.time()
3599 try: 3599 try:
3600 self.serv.accept() 3600 self.serv.accept()
3601 except OSError: 3601 except OSError:
3602 pass 3602 pass
3603 end = time.time() 3603 end = time.time()
3604 self.assertTrue((end - start) < 1.0, "Error creating with non-blocki ng mode.") 3604 self.assertLess(end - start, 1.0, "Error creating with non-blocking mode.")
3605 3605
3606 def _testInitNonBlocking(self): 3606 def _testInitNonBlocking(self):
3607 pass 3607 pass
3608 3608
3609 def testInheritFlags(self): 3609 def testInheritFlags(self):
3610 # Issue #7995: when calling accept() on a listening socket with a 3610 # Issue #7995: when calling accept() on a listening socket with a
3611 # timeout, the resulting socket should not be non-blocking. 3611 # timeout, the resulting socket should not be non-blocking.
3612 self.serv.settimeout(10) 3612 self.serv.settimeout(10)
3613 try: 3613 try:
3614 conn, addr = self.serv.accept() 3614 conn, addr = self.serv.accept()
(...skipping 194 matching lines...) Expand 10 before | Expand all | Expand 10 after
3809 msg = self.cli_conn.recv(len(MSG)) 3809 msg = self.cli_conn.recv(len(MSG))
3810 if isinstance(self.read_msg, str): 3810 if isinstance(self.read_msg, str):
3811 msg = msg.decode() 3811 msg = msg.decode()
3812 self.assertEqual(msg, self.read_msg) 3812 self.assertEqual(msg, self.read_msg)
3813 3813
3814 def _testMakefileAfterMakefileClose(self): 3814 def _testMakefileAfterMakefileClose(self):
3815 self.write_file.write(self.write_msg) 3815 self.write_file.write(self.write_msg)
3816 self.write_file.flush() 3816 self.write_file.flush()
3817 3817
3818 def testClosedAttr(self): 3818 def testClosedAttr(self):
3819 self.assertTrue(not self.read_file.closed) 3819 self.assertFalse(self.read_file.closed)
3820 3820
3821 def _testClosedAttr(self): 3821 def _testClosedAttr(self):
3822 self.assertTrue(not self.write_file.closed) 3822 self.assertFalse(self.write_file.closed)
3823 3823
3824 def testAttributes(self): 3824 def testAttributes(self):
3825 self.assertEqual(self.read_file.mode, self.read_mode) 3825 self.assertEqual(self.read_file.mode, self.read_mode)
3826 self.assertEqual(self.read_file.name, self.cli_conn.fileno()) 3826 self.assertEqual(self.read_file.name, self.cli_conn.fileno())
3827 3827
3828 def _testAttributes(self): 3828 def _testAttributes(self):
3829 self.assertEqual(self.write_file.mode, self.write_mode) 3829 self.assertEqual(self.write_file.mode, self.write_mode)
3830 self.assertEqual(self.write_file.name, self.serv_conn.fileno()) 3830 self.assertEqual(self.write_file.name, self.serv_conn.fileno())
3831 3831
3832 def testRealClose(self): 3832 def testRealClose(self):
(...skipping 367 matching lines...) Expand 10 before | Expand all | Expand 10 after
4200 self.cli = socket.create_connection((HOST, self.port), timeout=30, 4200 self.cli = socket.create_connection((HOST, self.port), timeout=30,
4201 source_address=('', self.source_port)) 4201 source_address=('', self.source_port))
4202 self.addCleanup(self.cli.close) 4202 self.addCleanup(self.cli.close)
4203 self.assertEqual(self.cli.getsockname()[1], self.source_port) 4203 self.assertEqual(self.cli.getsockname()[1], self.source_port)
4204 # The port number being used is sufficient to show that the bind() 4204 # The port number being used is sufficient to show that the bind()
4205 # call happened. 4205 # call happened.
4206 4206
4207 testTimeoutDefault = _justAccept 4207 testTimeoutDefault = _justAccept
4208 def _testTimeoutDefault(self): 4208 def _testTimeoutDefault(self):
4209 # passing no explicit timeout uses socket's global default 4209 # passing no explicit timeout uses socket's global default
4210 self.assertTrue(socket.getdefaulttimeout() is None) 4210 self.assertIsNone(socket.getdefaulttimeout())
4211 socket.setdefaulttimeout(42) 4211 socket.setdefaulttimeout(42)
4212 try: 4212 try:
4213 self.cli = socket.create_connection((HOST, self.port)) 4213 self.cli = socket.create_connection((HOST, self.port))
4214 self.addCleanup(self.cli.close) 4214 self.addCleanup(self.cli.close)
4215 finally: 4215 finally:
4216 socket.setdefaulttimeout(None) 4216 socket.setdefaulttimeout(None)
4217 self.assertEqual(self.cli.gettimeout(), 42) 4217 self.assertEqual(self.cli.gettimeout(), 42)
4218 4218
4219 testTimeoutNone = _justAccept 4219 testTimeoutNone = _justAccept
4220 def _testTimeoutNone(self): 4220 def _testTimeoutNone(self):
4221 # None timeout means the same as sock.settimeout(None) 4221 # None timeout means the same as sock.settimeout(None)
4222 self.assertTrue(socket.getdefaulttimeout() is None) 4222 self.assertIsNone(socket.getdefaulttimeout())
4223 socket.setdefaulttimeout(30) 4223 socket.setdefaulttimeout(30)
4224 try: 4224 try:
4225 self.cli = socket.create_connection((HOST, self.port), timeout=None) 4225 self.cli = socket.create_connection((HOST, self.port), timeout=None)
4226 self.addCleanup(self.cli.close) 4226 self.addCleanup(self.cli.close)
4227 finally: 4227 finally:
4228 socket.setdefaulttimeout(None) 4228 socket.setdefaulttimeout(None)
4229 self.assertEqual(self.cli.gettimeout(), None) 4229 self.assertEqual(self.cli.gettimeout(), None)
4230 4230
4231 testTimeoutValueNamed = _justAccept 4231 testTimeoutValueNamed = _justAccept
4232 def _testTimeoutValueNamed(self): 4232 def _testTimeoutValueNamed(self):
(...skipping 638 matching lines...) Expand 10 before | Expand all | Expand 10 after
4871 InterruptedSendTimeoutTest, 4871 InterruptedSendTimeoutTest,
4872 TestSocketSharing, 4872 TestSocketSharing,
4873 ]) 4873 ])
4874 4874
4875 thread_info = support.threading_setup() 4875 thread_info = support.threading_setup()
4876 support.run_unittest(*tests) 4876 support.run_unittest(*tests)
4877 support.threading_cleanup(*thread_info) 4877 support.threading_cleanup(*thread_info)
4878 4878
4879 if __name__ == "__main__": 4879 if __name__ == "__main__":
4880 test_main() 4880 test_main()
OLDNEW
« no previous file with comments | « Lib/test/test_smtplib.py ('k') | Lib/test/test_ssl.py » ('j') | no next file with comments »

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