Add test case

test/jpypetest/test_signals.py

@@ -0,0 +1,67 @@
+# *****************************************************************************
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#
+#   See NOTICE file for details.
+#
+# *****************************************************************************
+import os
+import signal
+import threading
+import unittest
+import jpype
+import subrun
+
+
+@subrun.TestCase
+class SignalsTest(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        # set up signal handling before starting jpype
+        cls.sigint_event = threading.Event()
+        cls.sigterm_event = threading.Event()
+
+        def sigint_handler(sig, frame):
+            cls.sigint_event.set()
+
+        def sigterm_handler(sig, frame):
+            cls.sigterm_event.set()
+
+        signal.signal(signal.SIGINT, sigint_handler)
+        signal.signal(signal.SIGTERM, sigterm_handler)
+
+        # start jpype with interrupt=False to pass back control to python
+        jpype.startJVM(interrupt=False)
+
+    def setUp(self):
+        self.sigint_event.clear()
+        self.sigterm_event.clear()
+
+    def testSigInt(self):
+        os.kill(os.getpid(), signal.SIGINT)
+
+        # the test is executed in the main thread. The signal cannot interrupt the threading.Event.wait() call
+        # so asserting the return value of `wait` does not work.
+        # However, after returning from the wait, the control should go to the signal handler, and the next `is_set`
+        # call will reflect the actual value of the flag.
+        self.sigint_event.wait(0.1)
+        self.assertTrue(self.sigint_event.is_set())
+        self.assertFalse(self.sigterm_event.is_set())
+
+    def testSigTerm(self):
+        os.kill(os.getpid(), signal.SIGTERM)
+
+        self.sigterm_event.wait(0.1)
+        self.assertTrue(self.sigterm_event.is_set())
+        self.assertFalse(self.sigint_event.is_set())