openstack/oslo-incubator
Code Issues Proposed changes

Fix test cases in tests.unit.test_service

Browse Source 
Be patient in test_service while waiting service to be spawned or killed,
because:
    * Service spawning has a limit of one process a second.
    * A test server in the cloud might have limited resources, the time there
      might be several times slower.

Change-Id: Idab1d1158ecc7564d0b1f9cdd32496496ff5b3f8
This commit is contained in:
Hengqing Hu 2012-12-28 14:17:12 +08:00
parent ffeb085508
commit 29e58dabc9
1 changed files with 26 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
tests/unit/test_service.py
View File

@ -97,18 +97,24 @@ class ServiceLauncherTest(utils.BaseTestCase):
self.pid = pid
# Wait for up to a second for workers to get started
start = time.time()
while time.time() - start < 1:
workers = self._get_workers()
if len(workers) == self.workers:
break
time.sleep(.1)
# Wait at most 10 seconds to spawn workers
cond = lambda: self.workers == len(self._get_workers())
timeout = 10
self._wait(cond, timeout)
workers = self._get_workers()
self.assertEqual(len(workers), self.workers)
return workers
def _wait(self, cond, timeout):
start = time.time()
while True:
if cond():
break
if time.time() - start > timeout:
break
time.sleep(.1)
def setUp(self):
super(ServiceLauncherTest, self).setUp()
# FIXME(markmc): Ugly hack to workaround bug #1073732
@ -149,18 +155,14 @@ class ServiceLauncherTest(utils.BaseTestCase):
LOG.info('pid of first child is %s' % start_workers[0])
os.kill(start_workers[0], signal.SIGTERM)
# loop and check if new worker is spawned (for 1 second max)
start = time.time()
while time.time() - start < 1:
end_workers = self._get_workers()
LOG.info('workers: %r' % end_workers)
if start_workers != end_workers:
break
time.sleep(.1)
# Wait at most 5 seconds to respawn a worker
cond = lambda: start_workers != self._get_workers()
timeout = 5
self._wait(cond, timeout)
# Make sure worker pids don't match
end_workers = self._get_workers()
LOG.info('workers: %r' % end_workers)
self.assertNotEqual(start_workers, end_workers)
def _terminate_with_signal(self, sig):
@ -168,17 +170,13 @@ class ServiceLauncherTest(utils.BaseTestCase):
os.kill(self.pid, sig)
# loop and check if all processes are killed (for 1 second max)
start = time.time()
while time.time() - start < 1:
workers = self._get_workers()
LOG.info('workers: %r' % workers)
if not workers:
break
time.sleep(.1)
# Wait at most 5 seconds to kill all workers
cond = lambda: not self._get_workers()
timeout = 5
self._wait(cond, timeout)
workers = self._get_workers()
LOG.info('workers: %r' % workers)
self.assertFalse(workers, 'No OS processes left.')
def test_terminate_sigkill(self):