cpython/Lib/test/test_asynchat.py

271 lines
8.7 KiB
Python
Raw Normal View History

# test asynchat
import asyncore, asynchat, socket, threading, time
import unittest
import sys
from test import test_support
# Skip tests if thread module does not exist.
test_support.import_module('thread')
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
HOST = test_support.HOST
SERVER_QUIT = 'QUIT\n'
class echo_server(threading.Thread):
# parameter to determine the number of bytes passed back to the
# client each send
chunk_size = 1
def __init__(self, event):
threading.Thread.__init__(self)
self.event = event
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.port = test_support.bind_port(self.sock)
# This will be set if the client wants us to wait before echoing data
# back.
self.start_resend_event = None
def run(self):
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
self.sock.listen(1)
self.event.set()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
conn, client = self.sock.accept()
self.buffer = ""
# collect data until quit message is seen
while SERVER_QUIT not in self.buffer:
data = conn.recv(1)
if not data:
break
self.buffer = self.buffer + data
# remove the SERVER_QUIT message
self.buffer = self.buffer.replace(SERVER_QUIT, '')
if self.start_resend_event:
self.start_resend_event.wait()
# re-send entire set of collected data
try:
# this may fail on some tests, such as test_close_when_done, since
# the client closes the channel when it's done sending
while self.buffer:
n = conn.send(self.buffer[:self.chunk_size])
time.sleep(0.001)
self.buffer = self.buffer[n:]
except:
pass
conn.close()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
self.sock.close()
class echo_client(asynchat.async_chat):
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
def __init__(self, terminator, server_port):
asynchat.async_chat.__init__(self)
self.contents = []
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
self.connect((HOST, server_port))
self.set_terminator(terminator)
self.buffer = ''
def handle_connect(self):
pass
if sys.platform == 'darwin':
# select.poll returns a select.POLLHUP at the end of the tests
# on darwin, so just ignore it
def handle_expt(self):
pass
def collect_incoming_data(self, data):
self.buffer += data
def found_terminator(self):
self.contents.append(self.buffer)
self.buffer = ""
def start_echo_server():
event = threading.Event()
s = echo_server(event)
s.start()
event.wait()
event.clear()
time.sleep(0.01) # Give server time to start accepting.
return s, event
class TestAsynchat(unittest.TestCase):
usepoll = False
def setUp (self):
self._threads = test_support.threading_setup()
def tearDown (self):
test_support.threading_cleanup(*self._threads)
def line_terminator_check(self, term, server_chunk):
event = threading.Event()
s = echo_server(event)
s.chunk_size = server_chunk
s.start()
event.wait()
event.clear()
time.sleep(0.01) # Give server time to start accepting.
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client(term, s.port)
c.push("hello ")
c.push("world%s" % term)
c.push("I'm not dead yet!%s" % term)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, ["hello world", "I'm not dead yet!"])
# the line terminator tests below check receiving variously-sized
# chunks back from the server in order to exercise all branches of
# async_chat.handle_read
def test_line_terminator1(self):
# test one-character terminator
for l in (1,2,3):
self.line_terminator_check('\n', l)
def test_line_terminator2(self):
# test two-character terminator
for l in (1,2,3):
self.line_terminator_check('\r\n', l)
def test_line_terminator3(self):
# test three-character terminator
for l in (1,2,3):
self.line_terminator_check('qqq', l)
def numeric_terminator_check(self, termlen):
# Try reading a fixed number of bytes
s, event = start_echo_server()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client(termlen, s.port)
data = "hello world, I'm not dead yet!\n"
c.push(data)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, [data[:termlen]])
def test_numeric_terminator1(self):
# check that ints & longs both work (since type is
# explicitly checked in async_chat.handle_read)
self.numeric_terminator_check(1)
self.numeric_terminator_check(1L)
def test_numeric_terminator2(self):
self.numeric_terminator_check(6L)
def test_none_terminator(self):
# Try reading a fixed number of bytes
s, event = start_echo_server()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client(None, s.port)
data = "hello world, I'm not dead yet!\n"
c.push(data)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, [])
self.assertEqual(c.buffer, data)
def test_simple_producer(self):
s, event = start_echo_server()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client('\n', s.port)
data = "hello world\nI'm not dead yet!\n"
p = asynchat.simple_producer(data+SERVER_QUIT, buffer_size=8)
c.push_with_producer(p)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, ["hello world", "I'm not dead yet!"])
def test_string_producer(self):
s, event = start_echo_server()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client('\n', s.port)
data = "hello world\nI'm not dead yet!\n"
c.push_with_producer(data+SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, ["hello world", "I'm not dead yet!"])
def test_empty_line(self):
# checks that empty lines are handled correctly
s, event = start_echo_server()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client('\n', s.port)
c.push("hello world\n\nI'm not dead yet!\n")
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
s.join()
self.assertEqual(c.contents, ["hello world", "", "I'm not dead yet!"])
def test_close_when_done(self):
s, event = start_echo_server()
s.start_resend_event = threading.Event()
- Issue #2550: The approach used by client/server code for obtaining ports to listen on in network-oriented tests has been refined in an effort to facilitate running multiple instances of the entire regression test suite in parallel without issue. test_support.bind_port() has been fixed such that it will always return a unique port -- which wasn't always the case with the previous implementation, especially if socket options had been set that affected address reuse (i.e. SO_REUSEADDR, SO_REUSEPORT). The new implementation of bind_port() will actually raise an exception if it is passed an AF_INET/SOCK_STREAM socket with either the SO_REUSEADDR or SO_REUSEPORT socket option set. Furthermore, if available, bind_port() will set the SO_EXCLUSIVEADDRUSE option on the socket it's been passed. This currently only applies to Windows. This option prevents any other sockets from binding to the host/port we've bound to, thus removing the possibility of the 'non-deterministic' behaviour, as Microsoft puts it, that occurs when a second SOCK_STREAM socket binds and accepts to a host/port that's already been bound by another socket. The optional preferred port parameter to bind_port() has been removed. Under no circumstances should tests be hard coding ports! test_support.find_unused_port() has also been introduced, which will pass a temporary socket object to bind_port() in order to obtain an unused port. The temporary socket object is then closed and deleted, and the port is returned. This method should only be used for obtaining an unused port in order to pass to an external program (i.e. the -accept [port] argument to openssl's s_server mode) or as a parameter to a server-oriented class that doesn't give you direct access to the underlying socket used. Finally, test_support.HOST has been introduced, which should be used for the host argument of any relevant socket calls (i.e. bind and connect). The following tests were updated to following the new conventions: test_socket, test_smtplib, test_asyncore, test_ssl, test_httplib, test_poplib, test_ftplib, test_telnetlib, test_socketserver, test_asynchat and test_socket_ssl. It is now possible for multiple instances of the regression test suite to run in parallel without issue.
2008-04-08 20:47:30 -03:00
c = echo_client('\n', s.port)
c.push("hello world\nI'm not dead yet!\n")
c.push(SERVER_QUIT)
c.close_when_done()
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
# Only allow the server to start echoing data back to the client after
# the client has closed its connection. This prevents a race condition
# where the server echoes all of its data before we can check that it
# got any down below.
s.start_resend_event.set()
s.join()
self.assertEqual(c.contents, [])
# the server might have been able to send a byte or two back, but this
# at least checks that it received something and didn't just fail
# (which could still result in the client not having received anything)
self.assertTrue(len(s.buffer) > 0)
class TestAsynchat_WithPoll(TestAsynchat):
usepoll = True
class TestHelperFunctions(unittest.TestCase):
def test_find_prefix_at_end(self):
self.assertEqual(asynchat.find_prefix_at_end("qwerty\r", "\r\n"), 1)
self.assertEqual(asynchat.find_prefix_at_end("qwertydkjf", "\r\n"), 0)
class TestFifo(unittest.TestCase):
def test_basic(self):
f = asynchat.fifo()
f.push(7)
f.push('a')
self.assertEqual(len(f), 2)
self.assertEqual(f.first(), 7)
self.assertEqual(f.pop(), (1, 7))
self.assertEqual(len(f), 1)
self.assertEqual(f.first(), 'a')
self.assertEqual(f.is_empty(), False)
self.assertEqual(f.pop(), (1, 'a'))
self.assertEqual(len(f), 0)
self.assertEqual(f.is_empty(), True)
self.assertEqual(f.pop(), (0, None))
def test_given_list(self):
f = asynchat.fifo(['x', 17, 3])
self.assertEqual(len(f), 3)
self.assertEqual(f.pop(), (1, 'x'))
self.assertEqual(f.pop(), (1, 17))
self.assertEqual(f.pop(), (1, 3))
self.assertEqual(f.pop(), (0, None))
def test_main(verbose=None):
test_support.run_unittest(TestAsynchat, TestAsynchat_WithPoll,
TestHelperFunctions, TestFifo)
if __name__ == "__main__":
test_main(verbose=True)