Permalink
Browse files

adds a little unit tests

  • Loading branch information...
1 parent f3ac4e7 commit fbfa9ddd1efbe7511b707c530314b87fda6bb419 @anrs anrs committed Mar 7, 2012
Showing with 10 additions and 93 deletions.
  1. +10 −6 tests/test_caoe.py
  2. +0 −87 tests/test_nonfork_caoe.py
View
@@ -29,8 +29,8 @@ def is_process_alive(pid):
else:
return True
-def parent(path, pause=False):
- caoe.install()
+def parent(path, pause=False, fork=True):
+ caoe.install(fork=fork)
open(os.path.join(path, 'parent-%d' % os.getpid()), 'w').close()
for i in range(3):
p = Process(target=child, args=(path,))
@@ -46,19 +46,21 @@ def child(path):
open(os.path.join(path, 'child-%d' % pid), 'w').close()
time.sleep(100)
-def test_all_child_processes_should_be_killed_if_parent_quit_normally():
+def test_all_child_processes_should_be_killed_if_parent_quit_normally(fork=True):
with mkdtmp() as tmpdir:
- p = Process(target=parent, args=(tmpdir,))
+ p = Process(target=parent, args=(tmpdir,), kwargs={'fork': fork})
p.start()
p.join(1)
cpids = [int(x.split('-')[1]) for x in os.listdir(tmpdir) if x.startswith('child-')]
eq_(len(cpids), 3)
ok_(all(not is_process_alive(pid) for pid in cpids))
+def test_nonfork_all_child_processes_should_be_killed_if_parent_quit_normally():
+ test_all_child_processes_should_be_killed_if_parent_quit_normally(fork=False)
-def test_all_child_processes_should_be_killed_if_parent_is_killed():
+def test_all_child_processes_should_be_killed_if_parent_is_killed(fork=True):
with mkdtmp() as tmpdir:
- p = Process(target=parent, args=(tmpdir,), kwargs={'pause': True})
+ p = Process(target=parent, args=(tmpdir,), kwargs={'pause': True, 'fork': fork})
p.start()
time.sleep(1) # wait for child processes spawned
p.terminate()
@@ -68,6 +70,8 @@ def test_all_child_processes_should_be_killed_if_parent_is_killed():
time.sleep(1) # wait for killing children
ok_(all(not is_process_alive(pid) for pid in cpids))
+def test_nonfork_all_child_processes_should_be_killed_if_parent_is_killed():
+ test_all_child_processes_should_be_killed_if_parent_is_killed(fork=False)
def test_all_child_processes_should_be_killed_if_parent_is_killed_by_SIGKILL():
with mkdtmp() as tmpdir:
@@ -1,87 +0,0 @@
-import os
-import tempfile
-from contextlib import contextmanager
-import shutil
-from multiprocessing import Process
-import time
-from signal import SIGKILL
-
-from nose.tools import ok_, eq_, set_trace
-
-import caoe
-
-@contextmanager
-def mkdtmp():
- tmpdir = tempfile.mkdtemp()
- yield tmpdir
- shutil.rmtree(tmpdir)
-
-def is_process_alive(pid):
- try:
- os.kill(pid, 0)
- except OSError, e:
- if e.errno == 3: # process is dead
- return False
- elif e.errno == 1: # no permission
- return True
- else:
- raise
- else:
- return True
-
-def parent(path, pause=False):
- caoe.install(fork=False)
- open(os.path.join(path, 'parent-%d' % os.getpid()), 'w').close()
- for i in range(3):
- print 'parent', os.getpid()
- p = Process(target=child, args=(path,))
- p.daemon = True
- p.start()
- if pause:
- time.sleep(10)
- else:
- time.sleep(0.1)
-
-def child(path):
- pid = os.getpid()
- print 'child', pid
- open(os.path.join(path, 'child-%d' % pid), 'w').close()
- time.sleep(100)
-
-def test_all_child_processes_should_be_killed_if_parent_quit_normally():
- with mkdtmp() as tmpdir:
- p = Process(target=parent, args=(tmpdir,))
- p.start()
- p.join(1)
- cpids = [int(x.split('-')[1]) for x in os.listdir(tmpdir) if x.startswith('child-')]
- eq_(len(cpids), 3)
- ok_(all(not is_process_alive(pid) for pid in cpids))
-
-test_all_child_processes_should_be_killed_if_parent_quit_normally()
-
-def test_all_child_processes_should_be_killed_if_parent_is_killed():
- with mkdtmp() as tmpdir:
- p = Process(target=parent, args=(tmpdir,), kwargs={'pause': True})
- p.start()
- time.sleep(1) # wait for child processes spawned
- p.terminate()
- p.join()
- cpids = [int(x.split('-')[1]) for x in os.listdir(tmpdir) if x.startswith('child-')]
- eq_(len(cpids), 3)
- time.sleep(1) # wait for killing children
- ok_(all(not is_process_alive(pid) for pid in cpids))
-
-
-def test_all_child_processes_could_not_be_killed_if_parent_is_killed_by_SIGKILL():
- with mkdtmp() as tmpdir:
- p = Process(target=parent, args=(tmpdir,), kwargs={'pause': True})
- p.start()
- time.sleep(1) # wait for child processes spawned
- os.kill(p.pid, SIGKILL)
- p.join()
- time.sleep(5) # wait for the parent status checking interval
- cpids = [int(x.split('-')[1]) for x in os.listdir(tmpdir) if x.startswith('child-')]
- eq_(len(cpids), 3)
- ok_(all(is_process_alive(pid) for pid in cpids))
-
-

0 comments on commit fbfa9dd

Please sign in to comment.