Tests: added Python test with threading.

This commit is contained in:
Andrei Zeliankou
2019-11-13 16:51:12 +03:00
parent 69ff7ce731
commit 45d75ee2cb
2 changed files with 46 additions and 0 deletions

View File

@@ -0,0 +1,33 @@
import sys
import time
import threading
class Foo(threading.Thread):
num = 10
def __init__(self, x):
self.__x = x
threading.Thread.__init__(self)
def log_index(self, index):
sys.stderr.write(
"(" + str(index) + ") Thread: " + str(self.__x) + "\n"
)
sys.stderr.flush()
def run(self):
i = 0
for _ in range(3):
self.log_index(i)
i += 1
time.sleep(1)
self.log_index(i)
i += 1
def application(environ, start_response):
Foo(Foo.num).start()
Foo.num += 10
start_response('200 OK', [('Content-Length', '0')])
return []

View File

@@ -507,6 +507,19 @@ Connection: close
self.assertEqual(self.get()['body'], '0123456789', 'write') self.assertEqual(self.get()['body'], '0123456789', 'write')
def test_python_application_threading(self):
"""wait_for_record() timeouts after 5s while every thread works at
least 3s. So without releasing GIL test should fail.
"""
self.load('threading')
for _ in range(10):
self.get(no_recv=True)
self.assertIsNotNone(
self.wait_for_record(r'\(5\) Thread: 100'), 'last thread finished'
)
if __name__ == '__main__': if __name__ == '__main__':
TestPythonApplication.main() TestPythonApplication.main()