Imported Upstream version 3.25.1

Change-Id: I31412b37aa390505e71295f3a2c2e3cf2ba88b41
Signed-off-by: DongHun Kwak <dh0128.kwak@samsung.com>

1 files changed, 15 insertions, 31 deletions

diff --git a/tests/test_mainloop.py b/tests/test_mainloop.py
index 44197b3..fda6787 100644
--- a/tests/test_mainloop.py
+++ b/tests/test_mainloop.py
@@ -4,21 +4,14 @@ import os
 import sys
 import select
 import signal
-import time
 import unittest
 
-try:
-    from _thread import start_new_thread
-    start_new_thread  # pyflakes
-except ImportError:
-    # Python 2
-    from thread import start_new_thread
 from gi.repository import GLib
 
-from compathelper import _bytes
-
 
 class TestMainLoop(unittest.TestCase):
+
+    @unittest.skipUnless(hasattr(os, "fork"), "no os.fork available")
     def test_exception_handling(self):
         pipe_r, pipe_w = os.pipe()
 
@@ -37,7 +30,7 @@ class TestMainLoop(unittest.TestCase):
         GLib.child_watch_add(GLib.PRIORITY_DEFAULT, pid, child_died, loop)
 
         os.close(pipe_r)
-        os.write(pipe_w, _bytes("Y"))
+        os.write(pipe_w, b"Y")
         os.close(pipe_w)
 
         def excepthook(type, value, traceback):
@@ -60,33 +53,24 @@ class TestMainLoop(unittest.TestCase):
         #
         self.assertFalse(got_exception)
 
-    def test_concurrency(self):
-        def on_usr1(signum, frame):
-            pass
-
-        try:
-            # create a thread which will terminate upon SIGUSR1 by way of
-            # interrupting sleep()
-            orig_handler = signal.signal(signal.SIGUSR1, on_usr1)
-            start_new_thread(time.sleep, (10,))
-
-            # now create two main loops
-            loop1 = GLib.MainLoop()
-            loop2 = GLib.MainLoop()
-            GLib.timeout_add(100, lambda: os.kill(os.getpid(), signal.SIGUSR1))
-            GLib.timeout_add(500, loop1.quit)
-            loop1.run()
-            loop2.quit()
-        finally:
-            signal.signal(signal.SIGUSR1, orig_handler)
-
+    @unittest.skipUnless(hasattr(os, "fork"), "no os.fork available")
     def test_sigint(self):
+        r, w = os.pipe()
         pid = os.fork()
         if pid == 0:
-            time.sleep(0.5)
+            # wait for the parent process loop to start
+            os.read(r, 1)
+            os.close(r)
+
             os.kill(os.getppid(), signal.SIGINT)
             os._exit(0)
 
+        def notify_child():
+            # tell the child that it can kill the parent
+            os.write(w, b"X")
+            os.close(w)
+
+        GLib.idle_add(notify_child)
         loop = GLib.MainLoop()
         try:
             loop.run()