From f7a2b658c97e900a1b287da090a6786cbb07f3f4 Mon Sep 17 00:00:00 2001 From: "James E. Blair" Date: Wed, 2 Dec 2015 11:03:25 -0800 Subject: Add a test for worker termination This was an attempt to find a code path that could cause stopWaitingForJobs to raise an exception. So far, that has failed, but add this test anyway to exercise some of that code. Change-Id: I39955fc7250af9b93f336c7c3e0315528bd8d4a9 --- gear/tests/__init__.py | 6 +++++- gear/tests/test_functional.py | 12 ++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/gear/tests/__init__.py b/gear/tests/__init__.py index 834b0f0..6d5edb4 100644 --- a/gear/tests/__init__.py +++ b/gear/tests/__init__.py @@ -17,6 +17,7 @@ """Common utilities used in testing""" import errno +import logging import os import socket @@ -49,7 +50,10 @@ class BaseTestCase(testtools.TestCase, testresources.ResourcedTestCase): stderr = self.useFixture(fixtures.StringStream('stderr')).stream self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr)) - self.useFixture(fixtures.FakeLogger()) + self.useFixture(fixtures.FakeLogger( + level=logging.DEBUG, + format='%(asctime)s %(name)-32s ' + '%(levelname)-8s %(message)s')) self.useFixture(fixtures.NestedTempfile()) diff --git a/gear/tests/test_functional.py b/gear/tests/test_functional.py index 247649a..26e72e3 100644 --- a/gear/tests/test_functional.py +++ b/gear/tests/test_functional.py @@ -14,11 +14,13 @@ # limitations under the License. import os +import threading import time from OpenSSL import crypto import fixtures import testscenarios +import testtools import gear from gear import tests @@ -125,6 +127,16 @@ class TestFunctional(tests.BaseTestCase): self.assertTrue(job.complete) self.assertEqual(job.data, [b'workdata']) + def test_worker_termination(self): + def getJob(): + with testtools.ExpectedException(gear.InterruptedError): + self.worker.getJob() + self.worker.registerFunction('test') + jobthread = threading.Thread(target=getJob) + jobthread.daemon = True + jobthread.start() + self.worker.stopWaitingForJobs() + def load_tests(loader, in_tests, pattern): return testscenarios.load_tests_apply_scenarios(loader, in_tests, pattern) -- cgit v1.2.1