Tests: added Python test with iterator context.
This commit is contained in:
23
test/python/ctx_iter_atexit/wsgi.py
Normal file
23
test/python/ctx_iter_atexit/wsgi.py
Normal file
@@ -0,0 +1,23 @@
|
||||
import atexit
|
||||
|
||||
class application:
|
||||
def __init__(self, environ, start_response):
|
||||
self.environ = environ
|
||||
self.start = start_response
|
||||
|
||||
def __iter__(self):
|
||||
atexit.register(self._atexit)
|
||||
|
||||
content_length = int(self.environ.get('CONTENT_LENGTH', 0))
|
||||
body = bytes(self.environ['wsgi.input'].read(content_length))
|
||||
|
||||
self.start('200', [
|
||||
('Content-Type', self.environ.get('CONTENT_TYPE')),
|
||||
('Content-Length', str(len(body)))
|
||||
])
|
||||
yield body
|
||||
|
||||
def _atexit(self):
|
||||
self.start('200', [
|
||||
('Content-Length', '0')
|
||||
])
|
||||
@@ -60,5 +60,28 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython):
|
||||
self.assertNotIn('Transfer-Encoding', self.get()['headers'],
|
||||
'204 header transfer encoding')
|
||||
|
||||
def test_python_application_ctx_iter_atexit(self):
|
||||
self.skip_alerts.append(r'sendmsg.+failed')
|
||||
self.load('ctx_iter_atexit')
|
||||
|
||||
resp = self.post(headers={
|
||||
'Connection': 'close',
|
||||
'Content-Type': 'text/html',
|
||||
'Host': 'localhost'
|
||||
}, body='0123456789')
|
||||
|
||||
self.assertEqual(resp['status'], 200, 'ctx iter status')
|
||||
self.assertEqual(resp['body'], '0123456789', 'ctx iter body')
|
||||
|
||||
self.conf({
|
||||
"listeners": {},
|
||||
"applications": {}
|
||||
})
|
||||
|
||||
self.stop()
|
||||
|
||||
self.assertIsNotNone(self.search_in_log(r'RuntimeError'),
|
||||
'ctx iter atexit')
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user