Skip to content

G
gevent
Commit 1cc7e059 authored by Denis Bilenko's avatar Denis Bilenko
Browse files
Options

remove tests related to deleted modules

parent 6359107e
Hide whitespace changes
Inline Side-by-side
Showing with 2 additions and 374 deletions
greentest/test__all__.py
View file @ 1cc7e059
......@@ -17,8 +17,7 @@ ANY = ANY()
NOT_IMPLEMENTED = {
'socket': ['CAPI', 'gethostbyaddr', 'gethostbyname_ex', 'getnameinfo'],
'thread': ['allocate', 'exit_thread', 'interrupt_main', 'start_new'],
'select': ANY,
'httplib': ANY}
'select': ANY}
COULD_BE_MISSING = {
'socket': ['create_connection', 'RAND_add', 'RAND_egd', 'RAND_status']}
......@@ -165,8 +164,7 @@ are missing from %r:
for path, modname in walk_modules(include_so=True):
modname = modname.replace('gevent.', '').split('.')[0]
if modname not in ('http', 'httplib', 'wsgi'):
exec ('''def test_%s(self): self._test("gevent.%s")''' % (modname, modname))
exec ('''def test_%s(self): self._test("gevent.%s")''' % (modname, modname))
del path, modname
......
greentest/test__http.py deleted 100644 → 0
View file @ 6359107e
import sys; sys.exit(0) # gevent.http is currently not working
from gevent import monkey; monkey.patch_socket()
import gevent
from gevent import http
import greentest
import os
import sys
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 = sys.exc_info()[1]
if ex.args[0] != errno.ECONNREFUSED:
raise
except IOError:
e = sys.exc_info()[1]
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 = sys.exc_info()[1]
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()
greentest/test__issue58.py deleted 100644 → 0
View file @ 6359107e
import sys; sys.exit(0) # gevent.http is currently not working
import gevent
from gevent import wsgi
from gevent import socket
def error(env, start_response):
try:
raise ValueError('hello')
except Exception:
exc = sys.exc_info()
raise
server = wsgi.WSGIServer(('', 0), error)
server.start()
conn = socket.create_connection(('127.0.0.1', server.server_port))
conn.sendall('GET / HTTP/1.1\r\nConnection: close\r\n\r\n')
timeout = gevent.Timeout.start_new(0.1)
try:
conn.makefile(bufsize=1).read()
finally:
timeout.cancel()
greentest/test__server_http.py deleted 100644 → 0
View file @ 6359107e
import sys; sys.exit(0) # gevent.http is currently not working
import unittest
import gevent
from gevent import http
import test__server
from test__server import *
internal_error_start = 'HTTP/1.0 500 Internal Server Error\n'.replace('\n', '\r\n')
internal_error_end = '\n\nInternal Server Error'.replace('\n', '\r\n')
internal_error503 = '''HTTP/1.0 503 Service Unavailable
Connection: close
Content-type: text/plain
Content-length: 31
Service Temporarily Unavailable'''.replace('\n', '\r\n')
class SimpleHTTPServer(http.HTTPServer):
def handle(self, request):
if request.uri == '/ping':
request.send_reply(200, "OK", "PONG")
elif request.uri == '/short':
gevent.sleep(0.1)
request.send_reply(200, "OK", 'hello')
elif request.uri == '/long':
gevent.sleep(10)
request.send_reply(200, "OK", 'hello')
else:
request.send_reply(404, "gevent.http", "")
class Settings:
ServerClass = http.HTTPServer
ServerSubClass = SimpleHTTPServer
restartable = False
close_socket_detected = False
@staticmethod
def assert500(self):
conn = self.makefile()
conn.write('GET / HTTP/1.0\r\n\r\n')
result = conn.read()
assert result.startswith(internal_error_start), (result, internal_error_start)
assert result.endswith(internal_error_end), (result, internal_error_end)
assertAcceptedConnectionError = assert500
@staticmethod
def assert503(self):
conn = self.makefile()
conn.write('GET / HTTP/1.0\r\n\r\n')
result = conn.read()
assert result == internal_error503, (result, internal_error503)
assertPoolFull = assert503
test__server.Settings = Settings
if __name__ == '__main__':
unittest.main()
greentest/test__server_wsgi.py deleted 100644 → 0
View file @ 6359107e
import sys; sys.exit(0) # gevent.http is currently not working
import unittest
import gevent
from gevent import wsgi
import test__server
from test__server import *
from test__server_http import Settings as http_Settings
import gevent.pywsgi; gevent.pywsgi.WSGIServer.__init__ = None
def application(self, environ, start_response):
if environ['PATH_INFO'] == '/ping':
start_response("200 OK", [])
return ["PONG"]
elif environ['PATH_INFO'] == '/short':
gevent.sleep(0.1)
start_response("200 after 0.1 seconds", [])
return ["hello"]
elif environ['PATH_INFO'] == '/long':
gevent.sleep(10)
start_response("200 after 10 seconds", [])
return ["hello"]
else:
start_response("404 wsgi WTF?", [])
return []
class SimpleWSGIServer(wsgi.WSGIServer):
application = application
class Settings(http_Settings):
ServerClass = wsgi.WSGIServer
ServerSubClass = SimpleWSGIServer
test__server.Settings = Settings
if __name__ == '__main__':
unittest.main()
greentest/test__wsgi.py deleted 100644 → 0
View file @ 6359107e
import sys; sys.exit(0) # gevent.http is currently not working
import gevent
from gevent import wsgi
import test__pywsgi
from test__pywsgi import *
del TestHttps
test__pywsgi.server_implements_chunked = False
test__pywsgi.server_implements_pipeline = False
test__pywsgi.server_implements_100continue = False
TestCase.get_wsgi_module = lambda *args: wsgi
if __name__ == '__main__':
greentest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7