Commit f369f011 authored by Denis Bilenko's avatar Denis Bilenko

Fix tests on Python2.5

- make test__threading_vs_settrace.py compatible with 2.5 (still fails though)
- fix test__socket.py
- fix test__ssl.py
- 2.5/test_thread.py: reduce all delays 10 times
parent 4ca65560
......@@ -18,7 +18,7 @@ numtasks = 10
def task(ident):
global running
rmutex.acquire()
delay = random.random() * numtasks
delay = random.random() * numtasks / 10.
rmutex.release()
if verbose:
print 'task', ident, 'will run for', round(delay, 1), 'sec'
......@@ -86,7 +86,7 @@ def task2(ident):
delay = 0.001
else:
rmutex.acquire()
delay = random.random() * numtasks
delay = random.random() * numtasks / 10.
rmutex.release()
if verbose:
print 'task', ident, 'will run for', round(delay, 1), 'sec'
......
......@@ -51,7 +51,9 @@ class TestTCP(greentest.TestCase):
del self.listener
def create_connection(self):
return socket.create_connection(('127.0.0.1', self.port))
sock = socket.socket()
sock.connect(('127.0.0.1', self.port))
return sock
def _test_sendall(self, data):
......@@ -86,22 +88,14 @@ class TestTCP(greentest.TestCase):
def server():
(client, addr) = self.listener.accept()
# start reading, then, while reading, start writing. the reader should not hang forever
alive = [1]
N = 100000
def sendloop():
try:
while alive:
client.sendall('t' * 100000)
except socket.error:
return
raise AssertionError('expected socket error')
def sendall():
client.sendall('t' * N)
sender = Thread(target=sendloop)
try:
result = client.recv(1000)
self.assertEqual(result, 'hello world')
finally:
del alive[0]
sender = Thread(target=sendall)
result = client.recv(1000)
self.assertEqual(result, 'hello world')
sender.join()
server_thread = Thread(target=server)
......
......@@ -20,7 +20,8 @@ class TestSSL(TestTCP):
def create_connection(self):
return ssl.wrap_socket(super(TestSSL, self).create_connection())
test_sendall_timeout = None
def test_sendall_timeout(self):
pass
del TestTCP
......
......@@ -25,7 +25,7 @@ t.start()
def trace(frame, event, arg):
if threading is not None:
threading.current_thread()
threading.currentThread()
return trace
......@@ -49,7 +49,10 @@ sys.stdout.write("..finishing..")
class ThreadTrace(unittest.TestCase):
def test_untraceable_lock(self):
old = sys.gettrace()
if hasattr(sys, 'gettrace'):
old = sys.gettrace()
else:
old = None
lst = []
try:
def trace(frame, ev, arg):
......
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