8 from gi.repository import GLib
9 from gi import PyGIDeprecationWarning
12 class TestProcess(unittest.TestCase):
14 def test_deprecated_child_watch_no_data(self):
20 self.loop = GLib.MainLoop()
21 argv = [sys.executable, '-c', 'import sys']
22 pid, stdin, stdout, stderr = GLib.spawn_async(
23 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
25 with warnings.catch_warnings(record=True) as w:
26 warnings.simplefilter('always')
27 GLib.child_watch_add(pid, cb)
28 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
30 self.assertEqual(self.status, 0)
32 def test_deprecated_child_watch_data_priority(self):
33 def cb(pid, status, data):
40 self.loop = GLib.MainLoop()
41 argv = [sys.executable, '-c', 'import sys']
42 pid, stdin, stdout, stderr = GLib.spawn_async(
43 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
45 with warnings.catch_warnings(record=True) as w:
46 warnings.simplefilter('always')
47 id = GLib.child_watch_add(pid, cb, 12345, GLib.PRIORITY_HIGH)
48 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
49 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
52 self.assertEqual(self.data, 12345)
53 self.assertEqual(self.status, 0)
55 def test_deprecated_child_watch_data_priority_kwargs(self):
56 def cb(pid, status, data):
63 self.loop = GLib.MainLoop()
64 argv = [sys.executable, '-c', 'import sys']
65 pid, stdin, stdout, stderr = GLib.spawn_async(
66 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
68 with warnings.catch_warnings(record=True) as w:
69 warnings.simplefilter('always')
70 id = GLib.child_watch_add(pid, cb, priority=GLib.PRIORITY_HIGH, data=12345)
71 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
72 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
75 self.assertEqual(self.data, 12345)
76 self.assertEqual(self.status, 0)
78 def test_child_watch_no_data(self):
84 self.loop = GLib.MainLoop()
85 argv = [sys.executable, '-c', 'import sys']
86 pid, stdin, stdout, stderr = GLib.spawn_async(
87 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
89 id = GLib.child_watch_add(GLib.PRIORITY_HIGH, pid, cb)
90 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
93 self.assertEqual(self.status, 0)
95 def test_child_watch_with_data(self):
96 def cb(pid, status, data):
103 self.loop = GLib.MainLoop()
104 argv = [sys.executable, '-c', 'import sys']
105 pid, stdin, stdout, stderr = GLib.spawn_async(
106 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD)
107 self.assertEqual(stdin, None)
108 self.assertEqual(stdout, None)
109 self.assertEqual(stderr, None)
111 id = GLib.child_watch_add(GLib.PRIORITY_HIGH, pid, cb, 12345)
112 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority,
115 self.assertEqual(self.data, 12345)
116 self.assertEqual(self.status, 0)
118 def test_spawn_async_fds(self):
119 pid, stdin, stdout, stderr = GLib.spawn_async(
120 ['cat'], flags=GLib.SpawnFlags.SEARCH_PATH, standard_input=True,
121 standard_output=True, standard_error=True)
122 os.write(stdin, b'hello world!\n')
124 out = os.read(stdout, 50)
126 err = os.read(stderr, 50)
129 self.assertEqual(out, b'hello world!\n')
130 self.assertEqual(err, b'')
132 def test_spawn_async_envp(self):
133 pid, stdin, stdout, stderr = GLib.spawn_async(
134 ['sh', '-c', 'echo $TEST_VAR'], ['TEST_VAR=moo!'],
135 flags=GLib.SpawnFlags.SEARCH_PATH, standard_output=True)
136 self.assertEqual(stdin, None)
137 self.assertEqual(stderr, None)
138 out = os.read(stdout, 50)
141 self.assertEqual(out, b'moo!\n')
143 def test_backwards_compat_flags(self):
144 self.assertEqual(GLib.SpawnFlags.DO_NOT_REAP_CHILD,
145 GLib.SPAWN_DO_NOT_REAP_CHILD)