Merge "Add a test for worker termination"

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())
 
 

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)