<commit_before># Test suite for SocketServer.py
from greentest import test_support
from greentest.test_support import (verbose, verify, TESTFN, TestSkipped,
                               reap_children)
test_support.requires('network')

import errno
import os

NREQ = 3
DELAY = 0.1

class MyMixinHandler:
    def handle(self):
        time.sleep(DELAY)
        line = self.rfile.readline()
        time.sleep(DELAY)
        self.wfile.write(line)

class MyStreamHandler(MyMixinHandler, StreamRequestHandler):
    pass

class MyDatagramHandler(MyMixinHandler, DatagramRequestHandler):
    pass

class MyMixinServer:
    def serve_a_few(self):
        for i in range(NREQ):
            self.handle_request()
    def handle_error(self, request, client_address):
        self.close_request(request)
        self.server_close()
        raise

teststring = "hello world\r\n"

def receive(sock, n, timeout=20):
    #print 'select args: ', [sock], [], [], timeout
    r, w, x = select.select([sock], [], [], timeout)
    #print 'select results: ', r, w, x
    if sock in r:
        return sock.recv(n)
    else:
        raise RuntimeError, "timed out on %r" % (sock,)

def testdgram(proto, addr):
    s = socket.socket(proto, socket.SOCK_DGRAM)
    s.sendto(teststring, addr)
    buf = data = receive(s, 100)
    while data and '\n' not in buf:
        data = receive(s, 100)
        buf += data
    verify(buf == teststring)
    s.close()

def teststream(proto, addr):
    s = socket.socket(proto, socket.SOCK_STREAM)
    s.connect(addr)
    s.sendall(teststring)
    buf = data = receive(s, 100)
    while data and '\n' not in buf:
        data = receive(s, 100)
        buf += data
    verify(buf == teststring)
    s.close()

class ServerThread(threading.Thread):
    def __init__(self, addr, svrcls, hdlrcls):
        threading.Thread.__init__(self)
        self.__addr = addr
        self.__svrcls = svrcls
        self.__hdlrcls = hdlrcls
    def run(self):
        class svrcls(MyMixinServer, self.__svrcls):
            pass
        if verbose: print "thread: creating server"
        svr = svrcls(self.__addr, self.__hdlrcls)
        # pull the address out of the server in case it changed
        # this can happen if another process is using the port
        addr = svr.server_address
        if addr:
            self.__addr = addr
            if self.__addr != svr.socket.getsockname():
                raise RuntimeError('server_address was %s, expected %s' %
                                       (self.__addr, svr.socket.getsockname()))
        if verbose: print "thread: serving three times"
        svr.serve_a_few()
        if verbose: print "thread: done"

seed = 0
def pickport():
    global seed
    seed += 1
    return 10000 + (os.getpid() % 1000)*10 + seed

host = "localhost"
testfiles = []
def pickaddr(proto):
    if proto == socket.AF_INET:
        return (host, pickport())
    else:
        fn = TESTFN + str(pickport())
        if os.name == 'os2':
            # AF_UNIX socket names on OS/2 require a specific prefix
            # which can't include a drive letter and must also use
            # backslashes as directory separators
            if fn[1] == ':':
                fn = fn[2:]
            if fn[0] in (os.sep, os.altsep):
                fn = fn[1:]
            fn = os.path.join('\socket', fn)
            if os.sep == '/':
                fn = fn.replace(os.sep, os.altsep)
            else:
                fn = fn.replace(os.altsep, os.sep)
        testfiles.append(fn)
        return fn

def cleanup():
    for fn in testfiles:
        try:
            os.remove(fn)
        except os.error:
            pass
    testfiles[:] = []

def testloop(proto, servers, hdlrcls, testfunc):
    for svrcls in servers:
        addr = pickaddr(proto)
        if verbose:
            print "ADDR =", addr
            print "CLASS =", svrcls
        t = ServerThread(addr, svrcls, hdlrcls)
        if verbose: print "server created"
        t.start()
        if verbose: print "server running"
        for i in range(NREQ):
            time.sleep(DELAY)
            if verbose: print "test client", i
            testfunc(proto, addr)
        if verbose: print "waiting for server"
        t.join()
        if verbose: print "done"

class ForgivingTCPServer(TCPServer):
    # prevent errors if another process is using the port we want
    def server_bind(self):
        host, default_port = self.server_address
        # this code shamelessly stolen from test.test_support
        # the ports were changed to protect the innocent
        import sys
        for port in [default_port, 3434, 8798, 23833]:
            try:
                self.server_address = host, port
                TCPServer.server_bind(self)
                break
            except socket.error, (err, msg):
                if err != errno.EADDRINUSE:
                    raise
                print >>sys.__stderr__, \
                    '  WARNING: failed to listen on port %d, trying another' % port

tcpservers = [ForgivingTCPServer, ThreadingTCPServer]
if hasattr(os, 'fork') and os.name not in ('os2',):
    tcpservers.append(ForkingTCPServer)
udpservers = [UDPServer, ThreadingUDPServer]

print 'WARNING: ThreadingUDPServer is disabled'
udpservers = [UDPServer]

if hasattr(os, 'fork') and os.name not in ('os2',):
    udpservers.append(ForkingUDPServer)<commit_msg>test_socketserver.py: reverted disabling of a test<commit_after># Test suite for SocketServer.py
from greentest import test_support
from greentest.test_support import (verbose, verify, TESTFN, TestSkipped,
                               reap_children)
test_support.requires('network')

import errno
import os

NREQ = 3
DELAY = 0.1

class MyMixinHandler:
    def handle(self):
        time.sleep(DELAY)
        line = self.rfile.readline()
        time.sleep(DELAY)
        self.wfile.write(line)

class MyStreamHandler(MyMixinHandler, StreamRequestHandler):
    pass

class MyDatagramHandler(MyMixinHandler, DatagramRequestHandler):
    pass

class MyMixinServer:
    def serve_a_few(self):
        for i in range(NREQ):
            self.handle_request()
    def handle_error(self, request, client_address):
        self.close_request(request)
        self.server_close()
        raise

teststring = "hello world\r\n"

def receive(sock, n, timeout=20):
    #print 'select args: ', [sock], [], [], timeout
    r, w, x = select.select([sock], [], [], timeout)
    #print 'select results: ', r, w, x
    if sock in r:
        return sock.recv(n)
    else:
        raise RuntimeError, "timed out on %r" % (sock,)

def testdgram(proto, addr):
    s = socket.socket(proto, socket.SOCK_DGRAM)
    s.sendto(teststring, addr)
    buf = data = receive(s, 100)
    while data and '\n' not in buf:
        data = receive(s, 100)
        buf += data
    verify(buf == teststring)
    s.close()

def teststream(proto, addr):
    s = socket.socket(proto, socket.SOCK_STREAM)
    s.connect(addr)
    s.sendall(teststring)
    buf = data = receive(s, 100)
    while data and '\n' not in buf:
        data = receive(s, 100)
        buf += data
    verify(buf == teststring)
    s.close()

class ServerThread(threading.Thread):
    def __init__(self, addr, svrcls, hdlrcls):
        threading.Thread.__init__(self)
        self.__addr = addr
        self.__svrcls = svrcls
        self.__hdlrcls = hdlrcls
    def run(self):
        class svrcls(MyMixinServer, self.__svrcls):
            pass
        if verbose: print "thread: creating server"
        svr = svrcls(self.__addr, self.__hdlrcls)
        # pull the address out of the server in case it changed
        # this can happen if another process is using the port
        addr = svr.server_address
        if addr:
            self.__addr = addr
            if self.__addr != svr.socket.getsockname():
                raise RuntimeError('server_address was %s, expected %s' %
                                       (self.__addr, svr.socket.getsockname()))
        if verbose: print "thread: serving three times"
        svr.serve_a_few()
        if verbose: print "thread: done"

seed = 0
def pickport():
    global seed
    seed += 1
    return 10000 + (os.getpid() % 1000)*10 + seed

host = "localhost"
testfiles = []
def pickaddr(proto):
    if proto == socket.AF_INET:
        return (host, pickport())
    else:
        fn = TESTFN + str(pickport())
        if os.name == 'os2':
            # AF_UNIX socket names on OS/2 require a specific prefix
            # which can't include a drive letter and must also use
            # backslashes as directory separators
            if fn[1] == ':':
                fn = fn[2:]
            if fn[0] in (os.sep, os.altsep):
                fn = fn[1:]
            fn = os.path.join('\socket', fn)
            if os.sep == '/':
                fn = fn.replace(os.sep, os.altsep)
            else:
                fn = fn.replace(os.altsep, os.sep)
        testfiles.append(fn)
        return fn

def cleanup():
    for fn in testfiles:
        try:
            os.remove(fn)
        except os.error:
            pass
    testfiles[:] = []

def testloop(proto, servers, hdlrcls, testfunc):
    for svrcls in servers:
        addr = pickaddr(proto)
        if verbose:
            print "ADDR =", addr
            print "CLASS =", svrcls
        t = ServerThread(addr, svrcls, hdlrcls)
        if verbose: print "server created"
        t.start()
        if verbose: print "server running"
        for i in range(NREQ):
            time.sleep(DELAY)
            if verbose: print "test client", i
            testfunc(proto, addr)
        if verbose: print "waiting for server"
        t.join()
        if verbose: print "done"

class ForgivingTCPServer(TCPServer):
    # prevent errors if another process is using the port we want
    def server_bind(self):
        host, default_port = self.server_address
        # this code shamelessly stolen from test.test_support
        # the ports were changed to protect the innocent
        import sys
        for port in [default_port, 3434, 8798, 23833]:
            try:
                self.server_address = host, port
                TCPServer.server_bind(self)
                break
            except socket.error, (err, msg):
                if err != errno.EADDRINUSE:
                    raise
                print >>sys.__stderr__, \
                    '  WARNING: failed to listen on port %d, trying another' % port

tcpservers = [ForgivingTCPServer, ThreadingTCPServer]
if hasattr(os, 'fork') and os.name not in ('os2',):
    tcpservers.append(ForkingTCPServer)
udpservers = [UDPServer, ThreadingUDPServer]

if hasattr(os, 'fork') and os.name not in ('os2',):
    udpservers.append(ForkingUDPServer)
{'fraction_non_alphanumeric': 0.10276974727221859, 'fraction_numerical': 0.009512263359134571, 'mean_word_length': 2.1120139291932674, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3805058', 'n_tokens_mistral': 3352, 'n_tokens_neox': 3227, 'n_words': 953}