From 94c6ff5c174f15a73e938c44c8abb170b0aaed69 Mon Sep 17 00:00:00 2001
From: Ashwin Naren <arihant2math@gmail.com>
Date: Thu, 10 Apr 2025 18:03:50 -0700
Subject: [PATCH 1/3] try to remove asynchat and asyncore
---
Lib/asynchat.py | 307 -----------------------
Lib/asyncore.py | 642 ------------------------------------------------
2 files changed, 949 deletions(-)
delete mode 100644 Lib/asynchat.py
delete mode 100644 Lib/asyncore.py
diff --git a/Lib/asynchat.py b/Lib/asynchat.py
deleted file mode 100644
index fc1146adbb..0000000000
--- a/Lib/asynchat.py
+++ /dev/null
@@ -1,307 +0,0 @@
-# -*- Mode: Python; tab-width: 4 -*-
-# Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp
-# Author: Sam Rushing <rushing@nightmare.com>
-
-# ======================================================================
-# Copyright 1996 by Sam Rushing
-#
-# All Rights Reserved
-#
-# Permission to use, copy, modify, and distribute this software and
-# its documentation for any purpose and without fee is hereby
-# granted, provided that the above copyright notice appear in all
-# copies and that both that copyright notice and this permission
-# notice appear in supporting documentation, and that the name of Sam
-# Rushing not be used in advertising or publicity pertaining to
-# distribution of the software without specific, written prior
-# permission.
-#
-# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
-# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
-# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
-# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
-# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
-# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
-# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-# ======================================================================
-
-r"""A class supporting chat-style (command/response) protocols.
-
-This class adds support for 'chat' style protocols - where one side
-sends a 'command', and the other sends a response (examples would be
-the common internet protocols - smtp, nntp, ftp, etc..).
-
-The handle_read() method looks at the input stream for the current
-'terminator' (usually '\r\n' for single-line responses, '\r\n.\r\n'
-for multi-line output), calling self.found_terminator() on its
-receipt.
-
-for example:
-Say you build an async nntp client using this class. At the start
-of the connection, you'll have self.terminator set to '\r\n', in
-order to process the single-line greeting. Just before issuing a
-'LIST' command you'll set it to '\r\n.\r\n'. The output of the LIST
-command will be accumulated (using your own 'collect_incoming_data'
-method) up to the terminator, and then control will be returned to
-you - by calling your self.found_terminator() method.
-"""
-import asyncore
-from collections import deque
-
-
-class async_chat(asyncore.dispatcher):
- """This is an abstract class. You must derive from this class, and add
- the two methods collect_incoming_data() and found_terminator()"""
-
- # these are overridable defaults
-
- ac_in_buffer_size = 65536
- ac_out_buffer_size = 65536
-
- # we don't want to enable the use of encoding by default, because that is a
- # sign of an application bug that we don't want to pass silently
-
- use_encoding = 0
- encoding = 'latin-1'
-
- def __init__(self, sock=None, map=None):
- # for string terminator matching
- self.ac_in_buffer = b''
-
- # we use a list here rather than io.BytesIO for a few reasons...
- # del lst[:] is faster than bio.truncate(0)
- # lst = [] is faster than bio.truncate(0)
- self.incoming = []
-
- # we toss the use of the "simple producer" and replace it with
- # a pure deque, which the original fifo was a wrapping of
- self.producer_fifo = deque()
- asyncore.dispatcher.__init__(self, sock, map)
-
- def collect_incoming_data(self, data):
- raise NotImplementedError("must be implemented in subclass")
-
- def _collect_incoming_data(self, data):
- self.incoming.append(data)
-
- def _get_data(self):
- d = b''.join(self.incoming)
- del self.incoming[:]
- return d
-
- def found_terminator(self):
- raise NotImplementedError("must be implemented in subclass")
-
- def set_terminator(self, term):
- """Set the input delimiter.
-
- Can be a fixed string of any length, an integer, or None.
- """
- if isinstance(term, str) and self.use_encoding:
- term = bytes(term, self.encoding)
- elif isinstance(term, int) and term < 0:
- raise ValueError('the number of received bytes must be positive')
- self.terminator = term
-
- def get_terminator(self):
- return self.terminator
-
- # grab some more data from the socket,
- # throw it to the collector method,
- # check for the terminator,
- # if found, transition to the next state.
-
- def handle_read(self):
-
- try:
- data = self.recv(self.ac_in_buffer_size)
- except BlockingIOError:
- return
- except OSError as why:
- self.handle_error()
- return
-
- if isinstance(data, str) and self.use_encoding:
- data = bytes(str, self.encoding)
- self.ac_in_buffer = self.ac_in_buffer + data
-
- # Continue to search for self.terminator in self.ac_in_buffer,
- # while calling self.collect_incoming_data. The while loop
- # is necessary because we might read several data+terminator
- # combos with a single recv(4096).
-
- while self.ac_in_buffer:
- lb = len(self.ac_in_buffer)
- terminator = self.get_terminator()
- if not terminator:
- # no terminator, collect it all
- self.collect_incoming_data(self.ac_in_buffer)
- self.ac_in_buffer = b''
- elif isinstance(terminator, int):
- # numeric terminator
- n = terminator
- if lb < n:
- self.collect_incoming_data(self.ac_in_buffer)
- self.ac_in_buffer = b''
- self.terminator = self.terminator - lb
- else:
- self.collect_incoming_data(self.ac_in_buffer[:n])
- self.ac_in_buffer = self.ac_in_buffer[n:]
- self.terminator = 0
- self.found_terminator()
- else:
- # 3 cases:
- # 1) end of buffer matches terminator exactly:
- # collect data, transition
- # 2) end of buffer matches some prefix:
- # collect data to the prefix
- # 3) end of buffer does not match any prefix:
- # collect data
- terminator_len = len(terminator)
- index = self.ac_in_buffer.find(terminator)
- if index != -1:
- # we found the terminator
- if index > 0:
- # don't bother reporting the empty string
- # (source of subtle bugs)
- self.collect_incoming_data(self.ac_in_buffer[:index])
- self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:]
- # This does the Right Thing if the terminator
- # is changed here.
- self.found_terminator()
- else:
- # check for a prefix of the terminator
- index = find_prefix_at_end(self.ac_in_buffer, terminator)
- if index:
- if index != lb:
- # we found a prefix, collect up to the prefix
- self.collect_incoming_data(self.ac_in_buffer[:-index])
- self.ac_in_buffer = self.ac_in_buffer[-index:]
- break
- else:
- # no prefix, collect it all
- self.collect_incoming_data(self.ac_in_buffer)
- self.ac_in_buffer = b''
-
- def handle_write(self):
- self.initiate_send()
-
- def handle_close(self):
- self.close()
-
- def push(self, data):
- if not isinstance(data, (bytes, bytearray, memoryview)):
- raise TypeError('data argument must be byte-ish (%r)',
- type(data))
- sabs = self.ac_out_buffer_size
- if len(data) > sabs:
- for i in range(0, len(data), sabs):
- self.producer_fifo.append(data[i:i+sabs])
- else:
- self.producer_fifo.append(data)
- self.initiate_send()
-
- def push_with_producer(self, producer):
- self.producer_fifo.append(producer)
- self.initiate_send()
-
- def readable(self):
- "predicate for inclusion in the readable for select()"
- # cannot use the old predicate, it violates the claim of the
- # set_terminator method.
-
- # return (len(self.ac_in_buffer) <= self.ac_in_buffer_size)
- return 1
-
- def writable(self):
- "predicate for inclusion in the writable for select()"
- return self.producer_fifo or (not self.connected)
-
- def close_when_done(self):
- "automatically close this channel once the outgoing queue is empty"
- self.producer_fifo.append(None)
-
- def initiate_send(self):
- while self.producer_fifo and self.connected:
- first = self.producer_fifo[0]
- # handle empty string/buffer or None entry
- if not first:
- del self.producer_fifo[0]
- if first is None:
- self.handle_close()
- return
-
- # handle classic producer behavior
- obs = self.ac_out_buffer_size
- try:
- data = first[:obs]
- except TypeError:
- data = first.more()
- if data:
- self.producer_fifo.appendleft(data)
- else:
- del self.producer_fifo[0]
- continue
-
- if isinstance(data, str) and self.use_encoding:
- data = bytes(data, self.encoding)
-
- # send the data
- try:
- num_sent = self.send(data)
- except OSError:
- self.handle_error()
- return
-
- if num_sent:
- if num_sent < len(data) or obs < len(first):
- self.producer_fifo[0] = first[num_sent:]
- else:
- del self.producer_fifo[0]
- # we tried to send some actual data
- return
-
- def discard_buffers(self):
- # Emergencies only!
- self.ac_in_buffer = b''
- del self.incoming[:]
- self.producer_fifo.clear()
-
-
-class simple_producer:
-
- def __init__(self, data, buffer_size=512):
- self.data = data
- self.buffer_size = buffer_size
-
- def more(self):
- if len(self.data) > self.buffer_size:
- result = self.data[:self.buffer_size]
- self.data = self.data[self.buffer_size:]
- return result
- else:
- result = self.data
- self.data = b''
- return result
-
-
-# Given 'haystack', see if any prefix of 'needle' is at its end. This
-# assumes an exact match has already been checked. Return the number of
-# characters matched.
-# for example:
-# f_p_a_e("qwerty\r", "\r\n") => 1
-# f_p_a_e("qwertydkjf", "\r\n") => 0
-# f_p_a_e("qwerty\r\n", "\r\n") => <undefined>
-
-# this could maybe be made faster with a computed regex?
-# [answer: no; circa Python-2.0, Jan 2001]
-# new python: 28961/s
-# old python: 18307/s
-# re: 12820/s
-# regex: 14035/s
-
-def find_prefix_at_end(haystack, needle):
- l = len(needle) - 1
- while l and not haystack.endswith(needle[:l]):
- l -= 1
- return l
diff --git a/Lib/asyncore.py b/Lib/asyncore.py
deleted file mode 100644
index 0e92be3ad1..0000000000
--- a/Lib/asyncore.py
+++ /dev/null
@@ -1,642 +0,0 @@
-# -*- Mode: Python -*-
-# Id: asyncore.py,v 2.51 2000/09/07 22:29:26 rushing Exp
-# Author: Sam Rushing <rushing@nightmare.com>
-
-# ======================================================================
-# Copyright 1996 by Sam Rushing
-#
-# All Rights Reserved
-#
-# Permission to use, copy, modify, and distribute this software and
-# its documentation for any purpose and without fee is hereby
-# granted, provided that the above copyright notice appear in all
-# copies and that both that copyright notice and this permission
-# notice appear in supporting documentation, and that the name of Sam
-# Rushing not be used in advertising or publicity pertaining to
-# distribution of the software without specific, written prior
-# permission.
-#
-# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
-# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
-# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
-# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
-# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
-# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
-# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-# ======================================================================
-
-"""Basic infrastructure for asynchronous socket service clients and servers.
-
-There are only two ways to have a program on a single processor do "more
-than one thing at a time". Multi-threaded programming is the simplest and
-most popular way to do it, but there is another very different technique,
-that lets you have nearly all the advantages of multi-threading, without
-actually using multiple threads. it's really only practical if your program
-is largely I/O bound. If your program is CPU bound, then pre-emptive
-scheduled threads are probably what you really need. Network servers are
-rarely CPU-bound, however.
-
-If your operating system supports the select() system call in its I/O
-library (and nearly all do), then you can use it to juggle multiple
-communication channels at once; doing other work while your I/O is taking
-place in the "background." Although this strategy can seem strange and
-complex, especially at first, it is in many ways easier to understand and
-control than multi-threaded programming. The module documented here solves
-many of the difficult problems for you, making the task of building
-sophisticated high-performance network servers and clients a snap.
-"""
-
-import select
-import socket
-import sys
-import time
-import warnings
-
-import os
-from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
- ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
- errorcode
-
-_DISCONNECTED = frozenset({ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE,
- EBADF})
-
-try:
- socket_map
-except NameError:
- socket_map = {}
-
-def _strerror(err):
- try:
- return os.strerror(err)
- except (ValueError, OverflowError, NameError):
- if err in errorcode:
- return errorcode[err]
- return "Unknown error %s" %err
-
-class ExitNow(Exception):
- pass
-
-_reraised_exceptions = (ExitNow, KeyboardInterrupt, SystemExit)
-
-def read(obj):
- try:
- obj.handle_read_event()
- except _reraised_exceptions:
- raise
- except:
- obj.handle_error()
-
-def write(obj):
- try:
- obj.handle_write_event()
- except _reraised_exceptions:
- raise
- except:
- obj.handle_error()
-
-def _exception(obj):
- try:
- obj.handle_expt_event()
- except _reraised_exceptions:
- raise
- except:
- obj.handle_error()
-
-def readwrite(obj, flags):
- try:
- if flags & select.POLLIN:
- obj.handle_read_event()
- if flags & select.POLLOUT:
- obj.handle_write_event()
- if flags & select.POLLPRI:
- obj.handle_expt_event()
- if flags & (select.POLLHUP | select.POLLERR | select.POLLNVAL):
- obj.handle_close()
- except OSError as e:
- if e.args[0] not in _DISCONNECTED:
- obj.handle_error()
- else:
- obj.handle_close()
- except _reraised_exceptions:
- raise
- except:
- obj.handle_error()
-
-def poll(timeout=0.0, map=None):
- if map is None:
- map = socket_map
- if map:
- r = []; w = []; e = []
- for fd, obj in list(map.items()):
- is_r = obj.readable()
- is_w = obj.writable()
- if is_r:
- r.append(fd)
- # accepting sockets should not be writable
- if is_w and not obj.accepting:
- w.append(fd)
- if is_r or is_w:
- e.append(fd)
- if [] == r == w == e:
- time.sleep(timeout)
- return
-
- r, w, e = select.select(r, w, e, timeout)
-
- for fd in r:
- obj = map.get(fd)
- if obj is None:
- continue
- read(obj)
-
- for fd in w:
- obj = map.get(fd)
- if obj is None:
- continue
- write(obj)
-
- for fd in e:
- obj = map.get(fd)
- if obj is None:
- continue
- _exception(obj)
-
-def poll2(timeout=0.0, map=None):
- # Use the poll() support added to the select module in Python 2.0
- if map is None:
- map = socket_map
- if timeout is not None:
- # timeout is in milliseconds
- timeout = int(timeout*1000)
- pollster = select.poll()
- if map:
- for fd, obj in list(map.items()):
- flags = 0
- if obj.readable():
- flags |= select.POLLIN | select.POLLPRI
- # accepting sockets should not be writable
- if obj.writable() and not obj.accepting:
- flags |= select.POLLOUT
- if flags:
- pollster.register(fd, flags)
-
- r = pollster.poll(timeout)
- for fd, flags in r:
- obj = map.get(fd)
- if obj is None:
- continue
- readwrite(obj, flags)
-
-poll3 = poll2 # Alias for backward compatibility
-
-def loop(timeout=30.0, use_poll=False, map=None, count=None):
- if map is None:
- map = socket_map
-
- if use_poll and hasattr(select, 'poll'):
- poll_fun = poll2
- else:
- poll_fun = poll
-
- if count is None:
- while map:
- poll_fun(timeout, map)
-
- else:
- while map and count > 0:
- poll_fun(timeout, map)
- count = count - 1
-
-class dispatcher:
-
- debug = False
- connected = False
- accepting = False
- connecting = False
- closing = False
- addr = None
- ignore_log_types = frozenset({'warning'})
-
- def __init__(self, sock=None, map=None):
- if map is None:
- self._map = socket_map
- else:
- self._map = map
-
- self._fileno = None
-
- if sock:
- # Set to nonblocking just to make sure for cases where we
- # get a socket from a blocking source.
- sock.setblocking(0)
- self.set_socket(sock, map)
- self.connected = True
- # The constructor no longer requires that the socket
- # passed be connected.
- try:
- self.addr = sock.getpeername()
- except OSError as err:
- if err.args[0] in (ENOTCONN, EINVAL):
- # To handle the case where we got an unconnected
- # socket.
- self.connected = False
- else:
- # The socket is broken in some unknown way, alert
- # the user and remove it from the map (to prevent
- # polling of broken sockets).
- self.del_channel(map)
- raise
- else:
- self.socket = None
-
- def __repr__(self):
- status = [self.__class__.__module__+"."+self.__class__.__qualname__]
- if self.accepting and self.addr:
- status.append('listening')
- elif self.connected:
- status.append('connected')
- if self.addr is not None:
- try:
- status.append('%s:%d' % self.addr)
- except TypeError:
- status.append(repr(self.addr))
- return '<%s at %#x>' % (' '.join(status), id(self))
-
- def add_channel(self, map=None):
- #self.log_info('adding channel %s' % self)
- if map is None:
- map = self._map
- map[self._fileno] = self
-
- def del_channel(self, map=None):
- fd = self._fileno
- if map is None:
- map = self._map
- if fd in map:
- #self.log_info('closing channel %d:%s' % (fd, self))
- del map[fd]
- self._fileno = None
-
- def create_socket(self, family=socket.AF_INET, type=socket.SOCK_STREAM):
- self.family_and_type = family, type
- sock = socket.socket(family, type)
- sock.setblocking(0)
- self.set_socket(sock)
-
- def set_socket(self, sock, map=None):
- self.socket = sock
- self._fileno = sock.fileno()
- self.add_channel(map)
-
- def set_reuse_addr(self):
- # try to re-use a server port if possible
- try:
- self.socket.setsockopt(
- socket.SOL_SOCKET, socket.SO_REUSEADDR,
- self.socket.getsockopt(socket.SOL_SOCKET,
- socket.SO_REUSEADDR) | 1
- )
- except OSError:
- pass
-
- # ==================================================
- # predicates for select()
- # these are used as filters for the lists of sockets
- # to pass to select().
- # ==================================================
-
- def readable(self):
- return True
-
- def writable(self):
- return True
-
- # ==================================================
- # socket object methods.
- # ==================================================
-
- def listen(self, num):
- self.accepting = True
- if os.name == 'nt' and num > 5:
- num = 5
- return self.socket.listen(num)
-
- def bind(self, addr):
- self.addr = addr
- return self.socket.bind(addr)
-
- def connect(self, address):
- self.connected = False
- self.connecting = True
- err = self.socket.connect_ex(address)
- if err in (EINPROGRESS, EALREADY, EWOULDBLOCK) \
- or err == EINVAL and os.name == 'nt':
- self.addr = address
- return
- if err in (0, EISCONN):
- self.addr = address
- self.handle_connect_event()
- else:
- raise OSError(err, errorcode[err])
-
- def accept(self):
- # XXX can return either an address pair or None
- try:
- conn, addr = self.socket.accept()
- except TypeError:
- return None
- except OSError as why:
- if why.args[0] in (EWOULDBLOCK, ECONNABORTED, EAGAIN):
- return None
- else:
- raise
- else:
- return conn, addr
-
- def send(self, data):
- try:
- result = self.socket.send(data)
- return result
- except OSError as why:
- if why.args[0] == EWOULDBLOCK:
- return 0
- elif why.args[0] in _DISCONNECTED:
- self.handle_close()
- return 0
- else:
- raise
-
- def recv(self, buffer_size):
- try:
- data = self.socket.recv(buffer_size)
- if not data:
- # a closed connection is indicated by signaling
- # a read condition, and having recv() return 0.
- self.handle_close()
- return b''
- else:
- return data
- except OSError as why:
- # winsock sometimes raises ENOTCONN
- if why.args[0] in _DISCONNECTED:
- self.handle_close()
- return b''
- else:
- raise
-
- def close(self):
- self.connected = False
- self.accepting = False
- self.connecting = False
- self.del_channel()
- if self.socket is not None:
- try:
- self.socket.close()
- except OSError as why:
- if why.args[0] not in (ENOTCONN, EBADF):
- raise
-
- # log and log_info may be overridden to provide more sophisticated
- # logging and warning methods. In general, log is for 'hit' logging
- # and 'log_info' is for informational, warning and error logging.
-
- def log(self, message):
- sys.stderr.write('log: %s\n' % str(message))
-
- def log_info(self, message, type='info'):
- if type not in self.ignore_log_types:
- print('%s: %s' % (type, message))
-
- def handle_read_event(self):
- if self.accepting:
- # accepting sockets are never connected, they "spawn" new
- # sockets that are connected
- self.handle_accept()
- elif not self.connected:
- if self.connecting:
- self.handle_connect_event()
- self.handle_read()
- else:
- self.handle_read()
-
- def handle_connect_event(self):
- err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
- if err != 0:
- raise OSError(err, _strerror(err))
- self.handle_connect()
- self.connected = True
- self.connecting = False
-
- def handle_write_event(self):
- if self.accepting:
- # Accepting sockets shouldn't get a write event.
- # We will pretend it didn't happen.
- return
-
- if not self.connected:
- if self.connecting:
- self.handle_connect_event()
- self.handle_write()
-
- def handle_expt_event(self):
- # handle_expt_event() is called if there might be an error on the
- # socket, or if there is OOB data
- # check for the error condition first
- err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
- if err != 0:
- # we can get here when select.select() says that there is an
- # exceptional condition on the socket
- # since there is an error, we'll go ahead and close the socket
- # like we would in a subclassed handle_read() that received no
- # data
- self.handle_close()
- else:
- self.handle_expt()
-
- def handle_error(self):
- nil, t, v, tbinfo = compact_traceback()
-
- # sometimes a user repr method will crash.
- try:
- self_repr = repr(self)
- except:
- self_repr = '<__repr__(self) failed for object at %0x>' % id(self)
-
- self.log_info(
- 'uncaptured python exception, closing channel %s (%s:%s %s)' % (
- self_repr,
- t,
- v,
- tbinfo
- ),
- 'error'
- )
- self.handle_close()
-
- def handle_expt(self):
- self.log_info('unhandled incoming priority event', 'warning')
-
- def handle_read(self):
- self.log_info('unhandled read event', 'warning')
-
- def handle_write(self):
- self.log_info('unhandled write event', 'warning')
-
- def handle_connect(self):
- self.log_info('unhandled connect event', 'warning')
-
- def handle_accept(self):
- pair = self.accept()
- if pair is not None:
- self.handle_accepted(*pair)
-
- def handle_accepted(self, sock, addr):
- sock.close()
- self.log_info('unhandled accepted event', 'warning')
-
- def handle_close(self):
- self.log_info('unhandled close event', 'warning')
- self.close()
-
-# ---------------------------------------------------------------------------
-# adds simple buffered output capability, useful for simple clients.
-# [for more sophisticated usage use asynchat.async_chat]
-# ---------------------------------------------------------------------------
-
-class dispatcher_with_send(dispatcher):
-
- def __init__(self, sock=None, map=None):
- dispatcher.__init__(self, sock, map)
- self.out_buffer = b''
-
- def initiate_send(self):
- num_sent = 0
- num_sent = dispatcher.send(self, self.out_buffer[:65536])
- self.out_buffer = self.out_buffer[num_sent:]
-
- def handle_write(self):
- self.initiate_send()
-
- def writable(self):
- return (not self.connected) or len(self.out_buffer)
-
- def send(self, data):
- if self.debug:
- self.log_info('sending %s' % repr(data))
- self.out_buffer = self.out_buffer + data
- self.initiate_send()
-
-# ---------------------------------------------------------------------------
-# used for debugging.
-# ---------------------------------------------------------------------------
-
-def compact_traceback():
- t, v, tb = sys.exc_info()
- tbinfo = []
- if not tb: # Must have a traceback
- raise AssertionError("traceback does not exist")
- while tb:
- tbinfo.append((
- tb.tb_frame.f_code.co_filename,
- tb.tb_frame.f_code.co_name,
- str(tb.tb_lineno)
- ))
- tb = tb.tb_next
-
- # just to be safe
- del tb
-
- file, function, line = tbinfo[-1]
- info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])
- return (file, function, line), t, v, info
-
-def close_all(map=None, ignore_all=False):
- if map is None:
- map = socket_map
- for x in list(map.values()):
- try:
- x.close()
- except OSError as x:
- if x.args[0] == EBADF:
- pass
- elif not ignore_all:
- raise
- except _reraised_exceptions:
- raise
- except:
- if not ignore_all:
- raise
- map.clear()
-
-# Asynchronous File I/O:
-#
-# After a little research (reading man pages on various unixen, and
-# digging through the linux kernel), I've determined that select()
-# isn't meant for doing asynchronous file i/o.
-# Heartening, though - reading linux/mm/filemap.c shows that linux
-# supports asynchronous read-ahead. So _MOST_ of the time, the data
-# will be sitting in memory for us already when we go to read it.
-#
-# What other OS's (besides NT) support async file i/o? [VMS?]
-#
-# Regardless, this is useful for pipes, and stdin/stdout...
-
-if os.name == 'posix':
- class file_wrapper:
- # Here we override just enough to make a file
- # look like a socket for the purposes of asyncore.
- # The passed fd is automatically os.dup()'d
-
- def __init__(self, fd):
- self.fd = os.dup(fd)
-
- def __del__(self):
- if self.fd >= 0:
- warnings.warn("unclosed file %r" % self, ResourceWarning,
- source=self)
- self.close()
-
- def recv(self, *args):
- return os.read(self.fd, *args)
-
- def send(self, *args):
- return os.write(self.fd, *args)
-
- def getsockopt(self, level, optname, buflen=None):
- if (level == socket.SOL_SOCKET and
- optname == socket.SO_ERROR and
- not buflen):
- return 0
- raise NotImplementedError("Only asyncore specific behaviour "
- "implemented.")
-
- read = recv
- write = send
-
- def close(self):
- if self.fd < 0:
- return
- fd = self.fd
- self.fd = -1
- os.close(fd)
-
- def fileno(self):
- return self.fd
-
- class file_dispatcher(dispatcher):
-
- def __init__(self, fd, map=None):
- dispatcher.__init__(self, None, map)
- self.connected = True
- try:
- fd = fd.fileno()
- except AttributeError:
- pass
- self.set_file(fd)
- # set it to non-blocking mode
- os.set_blocking(fd, False)
-
- def set_file(self, fd):
- self.socket = file_wrapper(fd)
- self._fileno = self.socket.fileno()
- self.add_channel()
From 4c352979bcf80164b9ab39abac3f5693491ba7bd Mon Sep 17 00:00:00 2001
From: Ashwin Naren <arihant2math@gmail.com>
Date: Thu, 10 Apr 2025 18:28:05 -0700
Subject: [PATCH 2/3] remove associated tests
---
Lib/test/test_asynchat.py | 290 -------------
Lib/test/test_asyncore.py | 838 --------------------------------------
2 files changed, 1128 deletions(-)
delete mode 100644 Lib/test/test_asynchat.py
delete mode 100644 Lib/test/test_asyncore.py
diff --git a/Lib/test/test_asynchat.py b/Lib/test/test_asynchat.py
deleted file mode 100644
index 1fcc882ce6..0000000000
--- a/Lib/test/test_asynchat.py
+++ /dev/null
@@ -1,290 +0,0 @@
-# test asynchat
-
-from test import support
-from test.support import socket_helper
-from test.support import threading_helper
-
-
-import asynchat
-import asyncore
-import errno
-import socket
-import sys
-import threading
-import time
-import unittest
-import unittest.mock
-
-HOST = socket_helper.HOST
-SERVER_QUIT = b'QUIT\n'
-TIMEOUT = 3.0
-
-
-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
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.port = socket_helper.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):
- self.sock.listen()
- self.event.set()
- conn, client = self.sock.accept()
- self.buffer = b""
- # 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, b'')
-
- 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()
- self.sock.close()
-
-class echo_client(asynchat.async_chat):
-
- def __init__(self, terminator, server_port):
- asynchat.async_chat.__init__(self)
- self.contents = []
- self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
- self.connect((HOST, server_port))
- self.set_terminator(terminator)
- self.buffer = b""
-
- 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 = b""
-
-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 = threading_helper.threading_setup()
-
- def tearDown(self):
- threading_helper.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.
- c = echo_client(term, s.port)
- c.push(b"hello ")
- c.push(b"world" + term)
- c.push(b"I'm not dead yet!" + term)
- c.push(SERVER_QUIT)
- asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- threading_helper.join_thread(s)
-
- self.assertEqual(c.contents, [b"hello world", b"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(b'\n', l)
-
- def test_line_terminator2(self):
- # test two-character terminator
- for l in (1, 2, 3):
- self.line_terminator_check(b'\r\n', l)
-
- def test_line_terminator3(self):
- # test three-character terminator
- for l in (1, 2, 3):
- self.line_terminator_check(b'qqq', l)
-
- def numeric_terminator_check(self, termlen):
- # Try reading a fixed number of bytes
- s, event = start_echo_server()
- c = echo_client(termlen, s.port)
- data = b"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)
- threading_helper.join_thread(s)
-
- 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)
-
- def test_numeric_terminator2(self):
- self.numeric_terminator_check(6)
-
- def test_none_terminator(self):
- # Try reading a fixed number of bytes
- s, event = start_echo_server()
- c = echo_client(None, s.port)
- data = b"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)
- threading_helper.join_thread(s)
-
- self.assertEqual(c.contents, [])
- self.assertEqual(c.buffer, data)
-
- def test_simple_producer(self):
- s, event = start_echo_server()
- c = echo_client(b'\n', s.port)
- data = b"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)
- threading_helper.join_thread(s)
-
- self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
-
- def test_string_producer(self):
- s, event = start_echo_server()
- c = echo_client(b'\n', s.port)
- data = b"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)
- threading_helper.join_thread(s)
-
- self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
-
- def test_empty_line(self):
- # checks that empty lines are handled correctly
- s, event = start_echo_server()
- c = echo_client(b'\n', s.port)
- c.push(b"hello world\n\nI'm not dead yet!\n")
- c.push(SERVER_QUIT)
- asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- threading_helper.join_thread(s)
-
- self.assertEqual(c.contents,
- [b"hello world", b"", b"I'm not dead yet!"])
-
- def test_close_when_done(self):
- s, event = start_echo_server()
- s.start_resend_event = threading.Event()
- c = echo_client(b'\n', s.port)
- c.push(b"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()
- threading_helper.join_thread(s)
-
- 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.assertGreater(len(s.buffer), 0)
-
- def test_push(self):
- # Issue #12523: push() should raise a TypeError if it doesn't get
- # a bytes string
- s, event = start_echo_server()
- c = echo_client(b'\n', s.port)
- data = b'bytes\n'
- c.push(data)
- c.push(bytearray(data))
- c.push(memoryview(data))
- self.assertRaises(TypeError, c.push, 10)
- self.assertRaises(TypeError, c.push, 'unicode')
- c.push(SERVER_QUIT)
- asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- threading_helper.join_thread(s)
- self.assertEqual(c.contents, [b'bytes', b'bytes', b'bytes'])
-
-
-class TestAsynchat_WithPoll(TestAsynchat):
- usepoll = True
-
-
-class TestAsynchatMocked(unittest.TestCase):
- def test_blockingioerror(self):
- # Issue #16133: handle_read() must ignore BlockingIOError
- sock = unittest.mock.Mock()
- sock.recv.side_effect = BlockingIOError(errno.EAGAIN)
-
- dispatcher = asynchat.async_chat()
- dispatcher.set_socket(sock)
- self.addCleanup(dispatcher.del_channel)
-
- with unittest.mock.patch.object(dispatcher, 'handle_error') as error:
- dispatcher.handle_read()
- self.assertFalse(error.called)
-
-
-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 TestNotConnected(unittest.TestCase):
- def test_disallow_negative_terminator(self):
- # Issue #11259
- client = asynchat.async_chat()
- self.assertRaises(ValueError, client.set_terminator, -1)
-
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/test/test_asyncore.py b/Lib/test/test_asyncore.py
deleted file mode 100644
index bd43463da3..0000000000
--- a/Lib/test/test_asyncore.py
+++ /dev/null
@@ -1,838 +0,0 @@
-import asyncore
-import unittest
-import select
-import os
-import socket
-import sys
-import time
-import errno
-import struct
-import threading
-
-from test import support
-from test.support import os_helper
-from test.support import socket_helper
-from test.support import threading_helper
-from test.support import warnings_helper
-from io import BytesIO
-
-if support.PGO:
- raise unittest.SkipTest("test is not helpful for PGO")
-
-
-TIMEOUT = 3
-HAS_UNIX_SOCKETS = hasattr(socket, 'AF_UNIX')
-
-class dummysocket:
- def __init__(self):
- self.closed = False
-
- def close(self):
- self.closed = True
-
- def fileno(self):
- return 42
-
-class dummychannel:
- def __init__(self):
- self.socket = dummysocket()
-
- def close(self):
- self.socket.close()
-
-class exitingdummy:
- def __init__(self):
- pass
-
- def handle_read_event(self):
- raise asyncore.ExitNow()
-
- handle_write_event = handle_read_event
- handle_close = handle_read_event
- handle_expt_event = handle_read_event
-
-class crashingdummy:
- def __init__(self):
- self.error_handled = False
-
- def handle_read_event(self):
- raise Exception()
-
- handle_write_event = handle_read_event
- handle_close = handle_read_event
- handle_expt_event = handle_read_event
-
- def handle_error(self):
- self.error_handled = True
-
-# used when testing senders; just collects what it gets until newline is sent
-def capture_server(evt, buf, serv):
- try:
- serv.listen()
- conn, addr = serv.accept()
- except socket.timeout:
- pass
- else:
- n = 200
- start = time.monotonic()
- while n > 0 and time.monotonic() - start < 3.0:
- r, w, e = select.select([conn], [], [], 0.1)
- if r:
- n -= 1
- data = conn.recv(10)
- # keep everything except for the newline terminator
- buf.write(data.replace(b'\n', b''))
- if b'\n' in data:
- break
- time.sleep(0.01)
-
- conn.close()
- finally:
- serv.close()
- evt.set()
-
-def bind_af_aware(sock, addr):
- """Helper function to bind a socket according to its family."""
- if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
- # Make sure the path doesn't exist.
- os_helper.unlink(addr)
- socket_helper.bind_unix_socket(sock, addr)
- else:
- sock.bind(addr)
-
-
-class HelperFunctionTests(unittest.TestCase):
- def test_readwriteexc(self):
- # Check exception handling behavior of read, write and _exception
-
- # check that ExitNow exceptions in the object handler method
- # bubbles all the way up through asyncore read/write/_exception calls
- tr1 = exitingdummy()
- self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
- self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
- self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)
-
- # check that an exception other than ExitNow in the object handler
- # method causes the handle_error method to get called
- tr2 = crashingdummy()
- asyncore.read(tr2)
- self.assertEqual(tr2.error_handled, True)
-
- tr2 = crashingdummy()
- asyncore.write(tr2)
- self.assertEqual(tr2.error_handled, True)
-
- tr2 = crashingdummy()
- asyncore._exception(tr2)
- self.assertEqual(tr2.error_handled, True)
-
- # asyncore.readwrite uses constants in the select module that
- # are not present in Windows systems (see this thread:
- # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
- # These constants should be present as long as poll is available
-
- @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
- def test_readwrite(self):
- # Check that correct methods are called by readwrite()
-
- attributes = ('read', 'expt', 'write', 'closed', 'error_handled')
-
- expected = (
- (select.POLLIN, 'read'),
- (select.POLLPRI, 'expt'),
- (select.POLLOUT, 'write'),
- (select.POLLERR, 'closed'),
- (select.POLLHUP, 'closed'),
- (select.POLLNVAL, 'closed'),
- )
-
- class testobj:
- def __init__(self):
- self.read = False
- self.write = False
- self.closed = False
- self.expt = False
- self.error_handled = False
-
- def handle_read_event(self):
- self.read = True
-
- def handle_write_event(self):
- self.write = True
-
- def handle_close(self):
- self.closed = True
-
- def handle_expt_event(self):
- self.expt = True
-
- def handle_error(self):
- self.error_handled = True
-
- for flag, expectedattr in expected:
- tobj = testobj()
- self.assertEqual(getattr(tobj, expectedattr), False)
- asyncore.readwrite(tobj, flag)
-
- # Only the attribute modified by the routine we expect to be
- # called should be True.
- for attr in attributes:
- self.assertEqual(getattr(tobj, attr), attr==expectedattr)
-
- # check that ExitNow exceptions in the object handler method
- # bubbles all the way up through asyncore readwrite call
- tr1 = exitingdummy()
- self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
-
- # check that an exception other than ExitNow in the object handler
- # method causes the handle_error method to get called
- tr2 = crashingdummy()
- self.assertEqual(tr2.error_handled, False)
- asyncore.readwrite(tr2, flag)
- self.assertEqual(tr2.error_handled, True)
-
- def test_closeall(self):
- self.closeall_check(False)
-
- def test_closeall_default(self):
- self.closeall_check(True)
-
- def closeall_check(self, usedefault):
- # Check that close_all() closes everything in a given map
-
- l = []
- testmap = {}
- for i in range(10):
- c = dummychannel()
- l.append(c)
- self.assertEqual(c.socket.closed, False)
- testmap[i] = c
-
- if usedefault:
- socketmap = asyncore.socket_map
- try:
- asyncore.socket_map = testmap
- asyncore.close_all()
- finally:
- testmap, asyncore.socket_map = asyncore.socket_map, socketmap
- else:
- asyncore.close_all(testmap)
-
- self.assertEqual(len(testmap), 0)
-
- for c in l:
- self.assertEqual(c.socket.closed, True)
-
- def test_compact_traceback(self):
- try:
- raise Exception("I don't like spam!")
- except:
- real_t, real_v, real_tb = sys.exc_info()
- r = asyncore.compact_traceback()
- else:
- self.fail("Expected exception")
-
- (f, function, line), t, v, info = r
- self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
- self.assertEqual(function, 'test_compact_traceback')
- self.assertEqual(t, real_t)
- self.assertEqual(v, real_v)
- self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
-
-
-class DispatcherTests(unittest.TestCase):
- def setUp(self):
- pass
-
- def tearDown(self):
- asyncore.close_all()
-
- def test_basic(self):
- d = asyncore.dispatcher()
- self.assertEqual(d.readable(), True)
- self.assertEqual(d.writable(), True)
-
- def test_repr(self):
- d = asyncore.dispatcher()
- self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
-
- def test_log(self):
- d = asyncore.dispatcher()
-
- # capture output of dispatcher.log() (to stderr)
- l1 = "Lovely spam! Wonderful spam!"
- l2 = "I don't like spam!"
- with support.captured_stderr() as stderr:
- d.log(l1)
- d.log(l2)
-
- lines = stderr.getvalue().splitlines()
- self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2])
-
- def test_log_info(self):
- d = asyncore.dispatcher()
-
- # capture output of dispatcher.log_info() (to stdout via print)
- l1 = "Have you got anything without spam?"
- l2 = "Why can't she have egg bacon spam and sausage?"
- l3 = "THAT'S got spam in it!"
- with support.captured_stdout() as stdout:
- d.log_info(l1, 'EGGS')
- d.log_info(l2)
- d.log_info(l3, 'SPAM')
-
- lines = stdout.getvalue().splitlines()
- expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
- self.assertEqual(lines, expected)
-
- def test_unhandled(self):
- d = asyncore.dispatcher()
- d.ignore_log_types = ()
-
- # capture output of dispatcher.log_info() (to stdout via print)
- with support.captured_stdout() as stdout:
- d.handle_expt()
- d.handle_read()
- d.handle_write()
- d.handle_connect()
-
- lines = stdout.getvalue().splitlines()
- expected = ['warning: unhandled incoming priority event',
- 'warning: unhandled read event',
- 'warning: unhandled write event',
- 'warning: unhandled connect event']
- self.assertEqual(lines, expected)
-
- def test_strerror(self):
- # refers to bug #8573
- err = asyncore._strerror(errno.EPERM)
- if hasattr(os, 'strerror'):
- self.assertEqual(err, os.strerror(errno.EPERM))
- err = asyncore._strerror(-1)
- self.assertTrue(err != "")
-
-
-class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
- def readable(self):
- return False
-
- def handle_connect(self):
- pass
-
-
-class DispatcherWithSendTests(unittest.TestCase):
- def setUp(self):
- pass
-
- def tearDown(self):
- asyncore.close_all()
-
- @threading_helper.reap_threads
- def test_send(self):
- evt = threading.Event()
- sock = socket.socket()
- sock.settimeout(3)
- port = socket_helper.bind_port(sock)
-
- cap = BytesIO()
- args = (evt, cap, sock)
- t = threading.Thread(target=capture_server, args=args)
- t.start()
- try:
- # wait a little longer for the server to initialize (it sometimes
- # refuses connections on slow machines without this wait)
- time.sleep(0.2)
-
- data = b"Suppose there isn't a 16-ton weight?"
- d = dispatcherwithsend_noread()
- d.create_socket()
- d.connect((socket_helper.HOST, port))
-
- # give time for socket to connect
- time.sleep(0.1)
-
- d.send(data)
- d.send(data)
- d.send(b'\n')
-
- n = 1000
- while d.out_buffer and n > 0:
- asyncore.poll()
- n -= 1
-
- evt.wait()
-
- self.assertEqual(cap.getvalue(), data*2)
- finally:
- threading_helper.join_thread(t)
-
-
-@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),
- 'asyncore.file_wrapper required')
-class FileWrapperTest(unittest.TestCase):
- def setUp(self):
- self.d = b"It's not dead, it's sleeping!"
- with open(os_helper.TESTFN, 'wb') as file:
- file.write(self.d)
-
- def tearDown(self):
- os_helper.unlink(os_helper.TESTFN)
-
- def test_recv(self):
- fd = os.open(os_helper.TESTFN, os.O_RDONLY)
- w = asyncore.file_wrapper(fd)
- os.close(fd)
-
- self.assertNotEqual(w.fd, fd)
- self.assertNotEqual(w.fileno(), fd)
- self.assertEqual(w.recv(13), b"It's not dead")
- self.assertEqual(w.read(6), b", it's")
- w.close()
- self.assertRaises(OSError, w.read, 1)
-
- def test_send(self):
- d1 = b"Come again?"
- d2 = b"I want to buy some cheese."
- fd = os.open(os_helper.TESTFN, os.O_WRONLY | os.O_APPEND)
- w = asyncore.file_wrapper(fd)
- os.close(fd)
-
- w.write(d1)
- w.send(d2)
- w.close()
- with open(os_helper.TESTFN, 'rb') as file:
- self.assertEqual(file.read(), self.d + d1 + d2)
-
- @unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
- 'asyncore.file_dispatcher required')
- def test_dispatcher(self):
- fd = os.open(os_helper.TESTFN, os.O_RDONLY)
- data = []
- class FileDispatcher(asyncore.file_dispatcher):
- def handle_read(self):
- data.append(self.recv(29))
- s = FileDispatcher(fd)
- os.close(fd)
- asyncore.loop(timeout=0.01, use_poll=True, count=2)
- self.assertEqual(b"".join(data), self.d)
-
- def test_resource_warning(self):
- # Issue #11453
- fd = os.open(os_helper.TESTFN, os.O_RDONLY)
- f = asyncore.file_wrapper(fd)
-
- os.close(fd)
- with warnings_helper.check_warnings(('', ResourceWarning)):
- f = None
- support.gc_collect()
-
- def test_close_twice(self):
- fd = os.open(os_helper.TESTFN, os.O_RDONLY)
- f = asyncore.file_wrapper(fd)
- os.close(fd)
-
- os.close(f.fd) # file_wrapper dupped fd
- with self.assertRaises(OSError):
- f.close()
-
- self.assertEqual(f.fd, -1)
- # calling close twice should not fail
- f.close()
-
-
-class BaseTestHandler(asyncore.dispatcher):
-
- def __init__(self, sock=None):
- asyncore.dispatcher.__init__(self, sock)
- self.flag = False
-
- def handle_accept(self):
- raise Exception("handle_accept not supposed to be called")
-
- def handle_accepted(self):
- raise Exception("handle_accepted not supposed to be called")
-
- def handle_connect(self):
- raise Exception("handle_connect not supposed to be called")
-
- def handle_expt(self):
- raise Exception("handle_expt not supposed to be called")
-
- def handle_close(self):
- raise Exception("handle_close not supposed to be called")
-
- def handle_error(self):
- raise
-
-
-class BaseServer(asyncore.dispatcher):
- """A server which listens on an address and dispatches the
- connection to a handler.
- """
-
- def __init__(self, family, addr, handler=BaseTestHandler):
- asyncore.dispatcher.__init__(self)
- self.create_socket(family)
- self.set_reuse_addr()
- bind_af_aware(self.socket, addr)
- self.listen(5)
- self.handler = handler
-
- @property
- def address(self):
- return self.socket.getsockname()
-
- def handle_accepted(self, sock, addr):
- self.handler(sock)
-
- def handle_error(self):
- raise
-
-
-class BaseClient(BaseTestHandler):
-
- def __init__(self, family, address):
- BaseTestHandler.__init__(self)
- self.create_socket(family)
- self.connect(address)
-
- def handle_connect(self):
- pass
-
-
-class BaseTestAPI:
-
- def tearDown(self):
- asyncore.close_all(ignore_all=True)
-
- def loop_waiting_for_flag(self, instance, timeout=5):
- timeout = float(timeout) / 100
- count = 100
- while asyncore.socket_map and count > 0:
- asyncore.loop(timeout=0.01, count=1, use_poll=self.use_poll)
- if instance.flag:
- return
- count -= 1
- time.sleep(timeout)
- self.fail("flag not set")
-
- def test_handle_connect(self):
- # make sure handle_connect is called on connect()
-
- class TestClient(BaseClient):
- def handle_connect(self):
- self.flag = True
-
- server = BaseServer(self.family, self.addr)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_handle_accept(self):
- # make sure handle_accept() is called when a client connects
-
- class TestListener(BaseTestHandler):
-
- def __init__(self, family, addr):
- BaseTestHandler.__init__(self)
- self.create_socket(family)
- bind_af_aware(self.socket, addr)
- self.listen(5)
- self.address = self.socket.getsockname()
-
- def handle_accept(self):
- self.flag = True
-
- server = TestListener(self.family, self.addr)
- client = BaseClient(self.family, server.address)
- self.loop_waiting_for_flag(server)
-
- def test_handle_accepted(self):
- # make sure handle_accepted() is called when a client connects
-
- class TestListener(BaseTestHandler):
-
- def __init__(self, family, addr):
- BaseTestHandler.__init__(self)
- self.create_socket(family)
- bind_af_aware(self.socket, addr)
- self.listen(5)
- self.address = self.socket.getsockname()
-
- def handle_accept(self):
- asyncore.dispatcher.handle_accept(self)
-
- def handle_accepted(self, sock, addr):
- sock.close()
- self.flag = True
-
- server = TestListener(self.family, self.addr)
- client = BaseClient(self.family, server.address)
- self.loop_waiting_for_flag(server)
-
-
- def test_handle_read(self):
- # make sure handle_read is called on data received
-
- class TestClient(BaseClient):
- def handle_read(self):
- self.flag = True
-
- class TestHandler(BaseTestHandler):
- def __init__(self, conn):
- BaseTestHandler.__init__(self, conn)
- self.send(b'x' * 1024)
-
- server = BaseServer(self.family, self.addr, TestHandler)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_handle_write(self):
- # make sure handle_write is called
-
- class TestClient(BaseClient):
- def handle_write(self):
- self.flag = True
-
- server = BaseServer(self.family, self.addr)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_handle_close(self):
- # make sure handle_close is called when the other end closes
- # the connection
-
- class TestClient(BaseClient):
-
- def handle_read(self):
- # in order to make handle_close be called we are supposed
- # to make at least one recv() call
- self.recv(1024)
-
- def handle_close(self):
- self.flag = True
- self.close()
-
- class TestHandler(BaseTestHandler):
- def __init__(self, conn):
- BaseTestHandler.__init__(self, conn)
- self.close()
-
- server = BaseServer(self.family, self.addr, TestHandler)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_handle_close_after_conn_broken(self):
- # Check that ECONNRESET/EPIPE is correctly handled (issues #5661 and
- # #11265).
-
- data = b'\0' * 128
-
- class TestClient(BaseClient):
-
- def handle_write(self):
- self.send(data)
-
- def handle_close(self):
- self.flag = True
- self.close()
-
- def handle_expt(self):
- self.flag = True
- self.close()
-
- class TestHandler(BaseTestHandler):
-
- def handle_read(self):
- self.recv(len(data))
- self.close()
-
- def writable(self):
- return False
-
- server = BaseServer(self.family, self.addr, TestHandler)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- @unittest.skipIf(sys.platform.startswith("sunos"),
- "OOB support is broken on Solaris")
- def test_handle_expt(self):
- # Make sure handle_expt is called on OOB data received.
- # Note: this might fail on some platforms as OOB data is
- # tenuously supported and rarely used.
- if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
- self.skipTest("Not applicable to AF_UNIX sockets.")
-
- if sys.platform == "darwin" and self.use_poll:
- self.skipTest("poll may fail on macOS; see issue #28087")
-
- class TestClient(BaseClient):
- def handle_expt(self):
- self.socket.recv(1024, socket.MSG_OOB)
- self.flag = True
-
- class TestHandler(BaseTestHandler):
- def __init__(self, conn):
- BaseTestHandler.__init__(self, conn)
- self.socket.send(bytes(chr(244), 'latin-1'), socket.MSG_OOB)
-
- server = BaseServer(self.family, self.addr, TestHandler)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_handle_error(self):
-
- class TestClient(BaseClient):
- def handle_write(self):
- 1.0 / 0
- def handle_error(self):
- self.flag = True
- try:
- raise
- except ZeroDivisionError:
- pass
- else:
- raise Exception("exception not raised")
-
- server = BaseServer(self.family, self.addr)
- client = TestClient(self.family, server.address)
- self.loop_waiting_for_flag(client)
-
- def test_connection_attributes(self):
- server = BaseServer(self.family, self.addr)
- client = BaseClient(self.family, server.address)
-
- # we start disconnected
- self.assertFalse(server.connected)
- self.assertTrue(server.accepting)
- # this can't be taken for granted across all platforms
- #self.assertFalse(client.connected)
- self.assertFalse(client.accepting)
-
- # execute some loops so that client connects to server
- asyncore.loop(timeout=0.01, use_poll=self.use_poll, count=100)
- self.assertFalse(server.connected)
- self.assertTrue(server.accepting)
- self.assertTrue(client.connected)
- self.assertFalse(client.accepting)
-
- # disconnect the client
- client.close()
- self.assertFalse(server.connected)
- self.assertTrue(server.accepting)
- self.assertFalse(client.connected)
- self.assertFalse(client.accepting)
-
- # stop serving
- server.close()
- self.assertFalse(server.connected)
- self.assertFalse(server.accepting)
-
- def test_create_socket(self):
- s = asyncore.dispatcher()
- s.create_socket(self.family)
- self.assertEqual(s.socket.type, socket.SOCK_STREAM)
- self.assertEqual(s.socket.family, self.family)
- self.assertEqual(s.socket.gettimeout(), 0)
- self.assertFalse(s.socket.get_inheritable())
-
- def test_bind(self):
- if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
- self.skipTest("Not applicable to AF_UNIX sockets.")
- s1 = asyncore.dispatcher()
- s1.create_socket(self.family)
- s1.bind(self.addr)
- s1.listen(5)
- port = s1.socket.getsockname()[1]
-
- s2 = asyncore.dispatcher()
- s2.create_socket(self.family)
- # EADDRINUSE indicates the socket was correctly bound
- self.assertRaises(OSError, s2.bind, (self.addr[0], port))
-
- def test_set_reuse_addr(self):
- if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
- self.skipTest("Not applicable to AF_UNIX sockets.")
-
- with socket.socket(self.family) as sock:
- try:
- sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- except OSError:
- unittest.skip("SO_REUSEADDR not supported on this platform")
- else:
- # if SO_REUSEADDR succeeded for sock we expect asyncore
- # to do the same
- s = asyncore.dispatcher(socket.socket(self.family))
- self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,
- socket.SO_REUSEADDR))
- s.socket.close()
- s.create_socket(self.family)
- s.set_reuse_addr()
- self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,
- socket.SO_REUSEADDR))
-
- @threading_helper.reap_threads
- def test_quick_connect(self):
- # see: http://bugs.python.org/issue10340
- if self.family not in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
- self.skipTest("test specific to AF_INET and AF_INET6")
-
- server = BaseServer(self.family, self.addr)
- # run the thread 500 ms: the socket should be connected in 200 ms
- t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
- count=5))
- t.start()
- try:
- with socket.socket(self.family, socket.SOCK_STREAM) as s:
- s.settimeout(.2)
- s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
- struct.pack('ii', 1, 0))
-
- try:
- s.connect(server.address)
- except OSError:
- pass
- finally:
- threading_helper.join_thread(t)
-
-class TestAPI_UseIPv4Sockets(BaseTestAPI):
- family = socket.AF_INET
- addr = (socket_helper.HOST, 0)
-
-@unittest.skipUnless(socket_helper.IPV6_ENABLED, 'IPv6 support required')
-class TestAPI_UseIPv6Sockets(BaseTestAPI):
- family = socket.AF_INET6
- addr = (socket_helper.HOSTv6, 0)
-
-@unittest.skipUnless(HAS_UNIX_SOCKETS, 'Unix sockets required')
-class TestAPI_UseUnixSockets(BaseTestAPI):
- if HAS_UNIX_SOCKETS:
- family = socket.AF_UNIX
- addr = os_helper.TESTFN
-
- def tearDown(self):
- os_helper.unlink(self.addr)
- BaseTestAPI.tearDown(self)
-
-class TestAPI_UseIPv4Select(TestAPI_UseIPv4Sockets, unittest.TestCase):
- use_poll = False
-
-@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
-class TestAPI_UseIPv4Poll(TestAPI_UseIPv4Sockets, unittest.TestCase):
- use_poll = True
-
-class TestAPI_UseIPv6Select(TestAPI_UseIPv6Sockets, unittest.TestCase):
- use_poll = False
-
-@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
-class TestAPI_UseIPv6Poll(TestAPI_UseIPv6Sockets, unittest.TestCase):
- use_poll = True
-
-class TestAPI_UseUnixSocketsSelect(TestAPI_UseUnixSockets, unittest.TestCase):
- use_poll = False
-
-@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
-class TestAPI_UseUnixSocketsPoll(TestAPI_UseUnixSockets, unittest.TestCase):
- use_poll = True
-
-if __name__ == "__main__":
- unittest.main()
From 44c0f5ffd328e622840ce5447121d108f44d230a Mon Sep 17 00:00:00 2001
From: Ashwin Naren <arihant2math@gmail.com>
Date: Thu, 10 Apr 2025 18:32:53 -0700
Subject: [PATCH 3/3] temp patch of test_ftplib
---
Lib/test/test_ftplib.py | 9 ---------
1 file changed, 9 deletions(-)
diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py
index 7e632efa4c..b3e4e776a4 100644
--- a/Lib/test/test_ftplib.py
+++ b/Lib/test/test_ftplib.py
@@ -25,15 +25,6 @@
from test.support import asyncore
from test.support.socket_helper import HOST, HOSTv6
-import sys
-if sys.platform == 'win32':
- raise unittest.SkipTest("test_ftplib not working on windows")
-if getattr(sys, '_rustpython_debugbuild', False):
- raise unittest.SkipTest("something's weird on debug builds")
-
-asynchat = warnings_helper.import_deprecated('asynchat')
-asyncore = warnings_helper.import_deprecated('asyncore')
-
support.requires_working_socket(module=True)