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

Unified Diff: Lib/test/test_socket.py

Issue 17561: Add socket.create_server_sock() convenience function
Patch Set: Created 6 years, 10 months ago
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 side-by-side diff with in-line comments
Download patch
« Lib/socket.py ('K') | « Lib/socket.py ('k') | no next file » | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -4250,6 +4250,63 @@
socket.create_connection((HOST, 1234))
+class CreateServerSockTest(unittest.TestCase):
+
+ def echo_server(self, sock, port=None):
+ def run():
+ with sock:
+ conn, _ = sock.accept()
+ with conn:
+ msg = conn.recv(1024)
+ if not msg:
+ return
+ conn.sendall(msg)
+
+ sock.settimeout(2)
+ t = threading.Thread(target=run)
+ t.start()
+ time.sleep(.1)
+
+ def test_base(self):
+ port = support.find_unused_port()
+ with socket.create_server_sock((None, port)) as sock:
+ self.assertEqual(sock.getsockname()[1], port)
+ self.assertEqual(sock.type, socket.SOCK_STREAM)
+ if socket.has_dual_stack():
+ self.assertEqual(sock.family, socket.AF_INET6)
+ else:
+ self.assertEqual(sock.family, socket.AF_INET)
+ self.echo_server(sock)
+ with socket.create_connection(('localhost', port), timeout=2) as cl:
+ cl.sendall(b'foo')
+ self.assertEqual(cl.recv(1024), b'foo')
+
+ def test_dual_stack(self):
+ with socket.create_server_sock((None, 0)) as sock:
+ self.echo_server(sock)
+ port = sock.getsockname()[1]
+ with socket.create_connection(("127.0.0.1", port), timeout=2) as cl:
+ cl.sendall(b'foo')
+ self.assertEqual(cl.recv(1024), b'foo')
+
+ with socket.create_server_sock((None, 0)) as sock:
+ self.echo_server(sock)
+ port = sock.getsockname()[1]
+ if socket.has_dual_stack():
+ with socket.create_connection(("::1", port), timeout=2) as cl:
+ cl.sendall(b'foo')
+ self.assertEqual(cl.recv(1024), b'foo')
+ else:
+ self.assertRaises(ConnectionRefusedError,
+ socket.create_connection, ("::1", port))
+ # just stop server
+ with socket.create_connection(("127.0.0.1", port), timeout=2) \
+ as cl:
+ cl.sendall(b'foo')
+ cl.recv(1024)
+ unittest.skip('dual stack cannot be tested as not supported')
+
+
@unittest.skipUnless(thread, 'Threading required for this test.')
class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):
« Lib/socket.py ('K') | « Lib/socket.py ('k') | no next file » | no next file with comments »

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