← Back to branch summary

~ubuntu-branches/debian/jessie/python-eventlet/jessie

« back to all changes in this revision

Viewing changes to tests/wsgi_test.py

  • Committer: Bazaar Package Importer
  • Author(s): Stefano Rivera
  • Date: 2011-06-02 16:18:16 UTC
  • mfrom: (1.1.4 upstream)
  • Revision ID: james.westby@ubuntu.com-20110602161816-c888ncsqx70pfvfu
Tags: 0.9.15-1
https://launchpad.net/bugs/771512
* New upstream release.
  - Drop all patches, accepted upstream.
* Correct DEP3 headers (first line of Description is the subject)
* Bump Standards-Version to 3.9.2, no changes needed.
* Drop Breaks: ${python:Breaks}, no longer used by dh_python2.
* debian/copyright: Update to DEP5 Format r174.
* Restore doc/modules/zmq.rst and BD on Sphinx 1.0.
* reuseaddr.patch: The logic for deciding whether to use SO_REUSEADDR was
  inverted.
* retry-on-timeout.patch: If an operation times out, try one last time.
  (LP: #771512)

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/wsgi_test.py
124
124
            raise ConnectionClosed
125
125
        raise
126
126
    if not response_line:
127
 
        raise ConnectionClosed
 
127
        raise ConnectionClosed(response_line)
128
128
    
129
129
    header_lines = []
130
130
    while True:
174
174
        eventlet.sleep(0) # give previous server a chance to start
175
175
        if self.killer:
176
176
            greenthread.kill(self.killer)
177
 
            eventlet.sleep(0) # give killer a chance to kill
178
177
 
179
178
        new_kwargs = dict(max_size=128,
180
179
                          log=self.logfile,
952
951
        self.assertEqual(headers['connection'], 'close')
953
952
        self.assert_('unicode' in body)
954
953
 
 
954
    def test_ipv6(self):
 
955
        try:
 
956
            sock = eventlet.listen(('::1', 0), family=socket.AF_INET6)
 
957
        except (socket.gaierror, socket.error):  # probably no ipv6
 
958
            return
 
959
        log = StringIO()
 
960
        # first thing the server does is try to log the IP it's bound to
 
961
        def run_server():
 
962
            try:
 
963
                server = wsgi.server(sock=sock, log=log, site=Site())
 
964
            except ValueError:
 
965
                log.write('broked')
 
966
        eventlet.spawn_n(run_server)
 
967
        logval = log.getvalue()
 
968
        while not logval:
 
969
            eventlet.sleep(0.0)
 
970
            logval = log.getvalue()
 
971
        if 'broked' in logval:
 
972
            self.fail('WSGI server raised exception with ipv6 socket')
 
973
 
 
974
    def test_debug(self):
 
975
        self.spawn_server(debug=False)
 
976
        def crasher(env, start_response):
 
977
            raise RuntimeError("intentional crash")
 
978
        self.site.application = crasher
 
979
 
 
980
        sock = eventlet.connect(('localhost', self.port))
 
981
        fd = sock.makefile('w')
 
982
        fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
 
983
        fd.flush()
 
984
        response_line, headers, body = read_http(sock)
 
985
        self.assert_(response_line.startswith('HTTP/1.1 500 Internal Server Error'))
 
986
        self.assertEqual(body, '')
 
987
        self.assertEqual(headers['connection'], 'close')
 
988
        self.assert_('transfer-encoding' not in headers)
 
989
 
 
990
        # verify traceback when debugging enabled
 
991
        self.spawn_server(debug=True)
 
992
        self.site.application = crasher
 
993
        sock = eventlet.connect(('localhost', self.port))
 
994
        fd = sock.makefile('w')
 
995
        fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
 
996
        fd.flush()
 
997
        response_line, headers, body = read_http(sock)
 
998
        self.assert_(response_line.startswith('HTTP/1.1 500 Internal Server Error'))
 
999
        self.assert_('intentional crash' in body)
 
1000
        self.assert_('RuntimeError' in body)
 
1001
        self.assert_('Traceback' in body)
 
1002
        self.assertEqual(headers['connection'], 'close')
 
1003
        self.assert_('transfer-encoding' not in headers)
 
1004
 
 
1005
 
955
1006
def read_headers(sock):
956
1007
    fd = sock.makefile()
957
1008
    try:
1136
1187
            signal.alarm(0)
1137
1188
            signal.signal(signal.SIGALRM, signal.SIG_DFL)
1138
1189
 
1139
 
        assert not got_signal, "caught alarm signal. infinite loop detected."
1140
 
 
1141
 
    def test_ipv6(self):
1142
 
        try:
1143
 
            sock = eventlet.listen(('::1', 0), family=socket.AF_INET6)
1144
 
        except (socket.gaierror, socket.error):  # probably no ipv6
1145
 
            return
1146
 
        log = StringIO()
1147
 
        # first thing the server does is try to log the IP it's bound to
1148
 
        def run_server():
1149
 
            try:
1150
 
                server = wsgi.server(sock=sock, log=log, site=Site())
1151
 
            except ValueError:
1152
 
                log.write('broked')
1153
 
        eventlet.spawn_n(run_server)
1154
 
        logval = log.getvalue()
1155
 
        while not logval:
1156
 
            eventlet.sleep(0.0)
1157
 
            logval = log.getvalue()
1158
 
        if 'broked' in logval:
1159
 
            self.fail('WSGI server raised exception with ipv6 socket')
 
1190
        assert not got_signal, "caught alarm signal. infinite loop detected."        
1160
1191
 
1161
1192
 
1162
1193
if __name__ == '__main__':