File: test__http.py

package info (click to toggle)
python-gevent 0.13.6-1%2Bnmu3
  • links: PTS
  • area: main
  • in suites: wheezy
  • size: 2,324 kB
  • sloc: python: 13,296; makefile: 95; ansic: 37
file content (222 lines) | stat: -rw-r--r-- 6,246 bytes parent folder | download
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
from gevent import monkey; monkey.patch_socket()
import gevent
from gevent import http
import greentest
import os
import socket
import errno
from test__pywsgi import read_http

# add test for "chunked POST input -> chunked output"


class BoundTestCase(greentest.TestCase):

    address = ('127.0.0.1', 0)

    def setUp(self):
        greentest.TestCase.setUp(self)
        self.server = http.HTTPServer(self.address, self.handle)
        self.server.start()

    def tearDown(self):
        #self.print_netstat('before stop')
        timeout = gevent.Timeout.start_new(0.1)
        try:
            self.server.stop()
        finally:
            timeout.cancel()
        self.check_refused()
        greentest.TestCase.tearDown(self)

    def print_netstat(self, comment=''):
        cmd = 'echo "%s" && netstat -anp 2>&1 | grep %s' % (comment, self.server.server_port)
        os.system(cmd)

    def connect(self):
        s = socket.socket()
        s.connect((self.server.server_host, self.server.server_port))
        return s

    def urlopen(self, *args, **kwargs):
        fd = self.connect().makefile(bufsize=1)
        fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        return read_http(fd, *args, **kwargs)

    def check_refused(self):
        try:
            self.connect()
        except socket.error, ex:
            if ex[0] != errno.ECONNREFUSED:
                raise
        except IOError, e:
            print 'WARNING: instead of ECONNREFUSED got IOError: %s' % e


class TestNoop(BoundTestCase):

    def handle(self, r):
        pass

    def test(self):
        self.urlopen(code=500)


class TestClientCloses(BoundTestCase):

    # this test is useless. currently there's no way to know that the client closed the connection,
    # because libevent calls close_cb callback after you've tried to push something to the client

    def handle(self, r):
        self.log.append('reply')
        gevent.sleep(0.1)
        r.send_reply(200, 'OK', 'hello world')
        # QQQ should I get an exception here because the connection is closed?
        self.log.append('reply_done')

    def test(self):
        self.log = ['hey']
        s = self.connect()
        s.sendall('GET / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 100\r\n\r\n')
        s.close()
        gevent.sleep(0.2)
        self.assertEqual(self.log, ['hey', 'reply', 'reply_done'])


class TestStop(BoundTestCase):

    # this triggers if connection_closed is not handled properly
    #p: http.c:1921: evhttp_send: Assertion `((&evcon->requests)->tqh_first) == req' failed.

    def reply(self, r):
        # at this point object that was wrapped by r no longer exists
        r.send_reply(200, 'OK', 'hello world')

    def handle(self, r):
        # gonna reply later, when the connection is closed by the client
        return gevent.spawn_later(0.01, self.reply, r)

    def test(self):
        server = http.HTTPServer(self.address, self.handle)
        server.start()
        s = self.connect()
        s.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        s.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        s.close()
        server.stop()
        gevent.sleep(0.02)
        # stopping what already stopped is OK
        server.stop()


class TestSendReply(BoundTestCase):

    def handle(self, r):
        r.send_reply(200, 'OK', 'hello world')

    def test(self):
        self.urlopen(body='hello world')

    def test_keepalive(self):
        s = self.connect()
        s.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        s.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')


class TestException(BoundTestCase):

    def handle(self, r):
        raise greentest.ExpectedException('TestException.handle')

    def test(self):
        self.urlopen(code=500)


class TestSendReplyLater(BoundTestCase):

    def handle(self, r):
        gevent.sleep(0.01)
        r.send_reply(200, 'OK', 'hello world')

    def test(self):
        self.urlopen(body='hello world')

    def test_client_closes_10(self):
        s = self.connect()
        s.sendall('GET / HTTP/1.0\r\n\r\n')
        s.close()
        gevent.sleep(0.02)

    def test_client_closes_11(self):
        s = self.connect()
        s.sendall('GET / HTTP/1.1\r\n\r\n')
        s.close()
        gevent.sleep(0.02)


class TestDetach(BoundTestCase):

    def handle(self, r):
        input = r.input_buffer
        output = r.output_buffer
        assert r.input_buffer is input
        assert r.output_buffer is output
        assert input._obj
        assert output._obj
        r.detach()
        assert not input._obj
        assert not output._obj
        assert input.read() == ''
        assert output.read() == ''
        self.handled = True
        gevent.kill(self.current, Exception('test done'))

    def test(self):
        self.current = gevent.getcurrent()
        try:
            try:
                self.urlopen()
            except Exception, ex:
                assert str(ex) == 'test done', ex
        finally:
            self.current = None
        assert self.handled


# class TestSendReplyStartChunk(BoundTestCase):
#     spawn = True
#
#     def handle(self, r):
#         r.send_reply_start(200, 'OK')
#         gevent.sleep(0.2)
#         print 'handler sending chunk'
#         r.send_reply_chunk('hi')
#         print 'handler done'
#
#     def test(self):
#         response = urlopen(self.url)
#         print 'connected to %s' % self.url
#         assert response.code == 200, response
#         assert response.msg == 'OK', response
#         with gevent.Timeout(0.1, False):
#             data = response.read()
#             assert 'should not read anything', repr(data)
#         self.print_netstat('before response.close')
#         response.close()
#         self.print_netstat('after response.close')
#         print 123
#         gevent.sleep(0.5)
#         print 1234
#
#     def test_client_closes(self):
#         s = self.connect()
#         s.sendall('GET / HTTP/1.0\r\n\r\n')
#         gevent.sleep(0.1)
#         #self.print_netstat('before close')
#         s.close()
#         #self.print_netstat('after close')
#         gevent.sleep(0.5)


if __name__ == '__main__':
    greentest.main()