summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_asynchat.py
blob: 99261679673117934fd6397a18f644b8cdedb235 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
# 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)