saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Fix pickling error by decorating

Browse source
This commit is contained in:
twangboy 2017-11-30 08:59:31 -07:00
parent f8b8a8966d
commit be96de09cc
No known key found for this signature in database
GPG key ID: 93FF3BDEB278C9EB
1 changed files with 62 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

84
tests/unit/utils/test_process.py
View file

@ -7,6 +7,7 @@ import sys
import time
import signal
import multiprocessing
import functools
# Import Salt Testing libs
from tests.support.unit import TestCase, skipIf
@ -20,19 +21,64 @@ import salt.ext.six as six
from salt.ext.six.moves import range # pylint: disable=import-error,redefined-builtin
def die(func):
'''
Add proc title
'''
@functools.wraps(func)
def wrapper(self):
# Strip off the "test_" from the function name
name = func.__name__[5:]
def _die():
salt.utils.appendproctitle('test_{0}'.format(name))
setattr(self, 'die_' + name, _die)
return wrapper
def incr(func):
'''
Increment counter
'''
@functools.wraps(func)
def wrapper(self):
# Strip off the "test_" from the function name
name = func.__name__[5:]
def _incr(counter, num):
salt.utils.appendproctitle('test_{0}'.format(name))
for _ in range(0, num):
counter.value += 1
setattr(self, 'incr_' + name, _incr)
return wrapper
def spin(func):
'''
Spin indefinitely
'''
@functools.wraps(func)
def wrapper(self):
# Strip off the "test_" from the function name
name = func.__name__[5:]
def _spin():
salt.utils.appendproctitle('test_{0}'.format(name))
while True:
time.sleep(1)
setattr(self, 'spin_' + name, _spin)
return wrapper
class TestProcessManager(TestCase):
@spin
def test_basic(self):
'''
Make sure that the process is alive 2s later
'''
def spin():
salt.utils.appendproctitle('test_basic')
while True:
time.sleep(1)
process_manager = salt.utils.process.ProcessManager()
process_manager.add_process(spin)
process_manager.add_process(self.spin_basic)
initial_pid = next(six.iterkeys(process_manager._process_map))
time.sleep(2)
process_manager.check_children()
@ -48,17 +94,16 @@ class TestProcessManager(TestCase):
process_manager.stop_restarting()
process_manager.kill_children()
@spin
def test_kill(self):
def spin():
salt.utils.appendproctitle('test_kill')
while True:
time.sleep(1)
process_manager = salt.utils.process.ProcessManager()
process_manager.add_process(spin)
process_manager.add_process(self.spin_kill)
initial_pid = next(six.iterkeys(process_manager._process_map))
# kill the child
os.kill(initial_pid, signal.SIGKILL)
if salt.utils.is_windows():
os.kill(initial_pid, signal.SIGTERM)
else:
os.kill(initial_pid, signal.SIGKILL)
# give the OS time to give the signal...
time.sleep(0.1)
process_manager.check_children()
@ -74,15 +119,13 @@ class TestProcessManager(TestCase):
process_manager.stop_restarting()
process_manager.kill_children()
@die
def test_restarting(self):
'''
Make sure that the process is alive 2s later
'''
def die():
salt.utils.appendproctitle('test_restarting')
process_manager = salt.utils.process.ProcessManager()
process_manager.add_process(die)
process_manager.add_process(self.die_restarting)
initial_pid = next(six.iterkeys(process_manager._process_map))
time.sleep(2)
process_manager.check_children()
@ -99,14 +142,11 @@ class TestProcessManager(TestCase):
process_manager.kill_children()
@skipIf(sys.version_info < (2, 7), 'Needs > Py 2.7 due to bug in stdlib')
@incr
def test_counter(self):
def incr(counter, num):
salt.utils.appendproctitle('test_counter')
for _ in range(0, num):
counter.value += 1
counter = multiprocessing.Value('i', 0)
process_manager = salt.utils.process.ProcessManager()
process_manager.add_process(incr, args=(counter, 2))
process_manager.add_process(self.incr_counter, args=(counter, 2))
time.sleep(1)
process_manager.check_children()
time.sleep(1)