← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-eventlet/trusty-proposed

« back to all changes in this revision

Viewing changes to tests/patcher_test.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-03-02 08:57:57 UTC
  • Revision ID: package-import@ubuntu.com-20120302085757-kpasohz8nme655o1
Tags: 0.9.16-1ubuntu2
debian/patches/eventlet-leak.patch: Fixes memory leak in nova. 

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/patcher_test.py
293
293
        self.assertEqual(output, "done\n", output)
294
294
 
295
295
 
 
296
class Threading(ProcessBase):
 
297
    def test_orig_thread(self):
 
298
        new_mod = """import eventlet
 
299
eventlet.monkey_patch()
 
300
from eventlet import patcher
 
301
import threading
 
302
_threading = patcher.original('threading')
 
303
def test():
 
304
    print repr(threading.current_thread())
 
305
t = _threading.Thread(target=test)
 
306
t.start()
 
307
t.join()
 
308
print len(threading._active)
 
309
print len(_threading._active)
 
310
"""
 
311
        self.write_to_tempfile("newmod", new_mod)
 
312
        output, lines = self.launch_subprocess('newmod')
 
313
        self.assertEqual(len(lines), 4, "\n".join(lines))
 
314
        self.assert_(lines[0].startswith('<Thread'), lines[0])
 
315
        self.assertEqual(lines[1], "1", lines[1])
 
316
        self.assertEqual(lines[2], "1", lines[2])
 
317
 
 
318
    def test_threading(self):
 
319
        new_mod = """import eventlet
 
320
eventlet.monkey_patch()
 
321
import threading
 
322
def test():
 
323
    print repr(threading.current_thread())
 
324
t = threading.Thread(target=test)
 
325
t.start()
 
326
t.join()
 
327
print len(threading._active)
 
328
"""
 
329
        self.write_to_tempfile("newmod", new_mod)
 
330
        output, lines = self.launch_subprocess('newmod')
 
331
        self.assertEqual(len(lines), 3, "\n".join(lines))
 
332
        self.assert_(lines[0].startswith('<_MainThread'), lines[0])
 
333
        self.assertEqual(lines[1], "1", lines[1])
 
334
 
 
335
    def test_tpool(self):
 
336
        new_mod = """import eventlet
 
337
eventlet.monkey_patch()
 
338
from eventlet import tpool
 
339
import threading
 
340
def test():
 
341
    print repr(threading.current_thread())
 
342
tpool.execute(test)
 
343
print len(threading._active)
 
344
"""
 
345
        self.write_to_tempfile("newmod", new_mod)
 
346
        output, lines = self.launch_subprocess('newmod')
 
347
        self.assertEqual(len(lines), 3, "\n".join(lines))
 
348
        self.assert_(lines[0].startswith('<Thread'), lines[0])
 
349
        self.assertEqual(lines[1], "1", lines[1])
 
350
 
 
351
    def test_greenlet(self):
 
352
        new_mod = """import eventlet
 
353
eventlet.monkey_patch()
 
354
from eventlet import event
 
355
import threading
 
356
evt = event.Event()
 
357
def test():
 
358
    print repr(threading.current_thread())
 
359
    evt.send()
 
360
eventlet.spawn_n(test)
 
361
evt.wait()
 
362
print len(threading._active)
 
363
"""
 
364
        self.write_to_tempfile("newmod", new_mod)
 
365
        output, lines = self.launch_subprocess('newmod')
 
366
        self.assertEqual(len(lines), 3, "\n".join(lines))
 
367
        self.assert_(lines[0].startswith('<_MainThread'), lines[0])
 
368
        self.assertEqual(lines[1], "1", lines[1])
 
369
 
 
370
    def test_greenthread(self):
 
371
        new_mod = """import eventlet
 
372
eventlet.monkey_patch()
 
373
import threading
 
374
def test():
 
375
    print repr(threading.current_thread())
 
376
t = eventlet.spawn(test)
 
377
t.wait()
 
378
print len(threading._active)
 
379
"""
 
380
        self.write_to_tempfile("newmod", new_mod)
 
381
        output, lines = self.launch_subprocess('newmod')
 
382
        self.assertEqual(len(lines), 3, "\n".join(lines))
 
383
        self.assert_(lines[0].startswith('<_GreenThread'), lines[0])
 
384
        self.assertEqual(lines[1], "1", lines[1])
 
385
 
296
386
if __name__ == '__main__':
297
387
    main()