# test asynchat -- requires threading

import thread # If this fails, we can't test this module

import asyncore, asynchat, socket, threading, time

import unittest

from test import test_support

HOST = "127.0.0.1"

PORT = 54322

class echo_server(threading.Thread):

    def run(self):

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        global PORT

        PORT = test_support.bind_port(sock, HOST, PORT)

        sock.listen(1)

        conn, client = sock.accept()

        buffer = ""

        while "\n" not in buffer:

            data = conn.recv(1)

            if not data:

                break

            buffer = buffer + data

        while buffer:

            n = conn.send(buffer)

            buffer = buffer[n:]

        conn.close()

        sock.close()

class echo_client(asynchat.async_chat):

    def __init__(self, terminator):

        asynchat.async_chat.__init__(self)

        self.contents = None

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)

        self.connect((HOST, PORT))

        self.set_terminator(terminator)

        self.buffer = ""

    def handle_connect(self):

        pass

        ##print "Connected"

    def collect_incoming_data(self, data):

        self.buffer = self.buffer + data

    def found_terminator(self):

        #print "Received:", repr(self.buffer)

        self.contents = self.buffer

        self.buffer = ""

        self.close()

class TestAsynchat(unittest.TestCase):

    def setUp (self):

        pass

    def tearDown (self):

        pass

    def test_line_terminator(self):

        s = echo_server()

        s.start()

        time.sleep(1) # Give server time to initialize

        c = echo_client('\n')

        c.push("hello ")

        c.push("world\n")

        asyncore.loop()

        s.join()

        self.assertEqual(c.contents, 'hello world')

    def test_numeric_terminator(self):

        # Try reading a fixed number of bytes

        s = echo_server()

        s.start()

        time.sleep(1) # Give server time to initialize

        c = echo_client(6L)

        c.push("hello ")

        c.push("world\n")

        asyncore.loop()

        s.join()

        self.assertEqual(c.contents, 'hello ')

def test_main(verbose=None):

    test_support.run_unittest(TestAsynchat)

if __name__ == "__main__":

    test_main(verbose=True)