3 from __future__
import absolute_import
10 from gi
.repository
import GLib
11 from gi
import PyGIDeprecationWarning
14 @unittest.skipIf(os
.name
== "nt", "not on Windows")
15 class TestProcess(unittest
.TestCase
):
17 def test_deprecated_child_watch_no_data(self
):
18 cb
= lambda pid
, status
: None
20 with warnings
.catch_warnings(record
=True) as w
:
21 warnings
.simplefilter('always')
22 res
= GLib
._child
_watch
_add
_get
_args
(pid
, cb
)
23 self
.assertTrue(issubclass(w
[0].category
, PyGIDeprecationWarning
))
25 self
.assertEqual(len(res
), 4)
26 self
.assertEqual(res
[0], GLib
.PRIORITY_DEFAULT
)
27 self
.assertEqual(res
[1], pid
)
28 self
.assertTrue(callable(cb
))
29 self
.assertSequenceEqual(res
[3], [])
31 def test_deprecated_child_watch_data_priority(self
):
32 cb
= lambda pid
, status
: None
34 with warnings
.catch_warnings(record
=True) as w
:
35 warnings
.simplefilter('always')
36 res
= GLib
._child
_watch
_add
_get
_args
(pid
, cb
, 12345, GLib
.PRIORITY_HIGH
)
37 self
.assertTrue(issubclass(w
[0].category
, PyGIDeprecationWarning
))
39 self
.assertEqual(len(res
), 4)
40 self
.assertEqual(res
[0], GLib
.PRIORITY_HIGH
)
41 self
.assertEqual(res
[1], pid
)
42 self
.assertEqual(res
[2], cb
)
43 self
.assertSequenceEqual(res
[3], [12345])
45 def test_deprecated_child_watch_data_priority_kwargs(self
):
46 cb
= lambda pid
, status
: None
48 with warnings
.catch_warnings(record
=True) as w
:
49 warnings
.simplefilter('always')
50 res
= GLib
._child
_watch
_add
_get
_args
(pid
, cb
, priority
=GLib
.PRIORITY_HIGH
, data
=12345)
51 self
.assertTrue(issubclass(w
[0].category
, PyGIDeprecationWarning
))
53 self
.assertEqual(len(res
), 4)
54 self
.assertEqual(res
[0], GLib
.PRIORITY_HIGH
)
55 self
.assertEqual(res
[1], pid
)
56 self
.assertEqual(res
[2], cb
)
57 self
.assertSequenceEqual(res
[3], [12345])
59 @unittest.expectedFailure
# using keyword args is fully supported by PyGObject machinery
60 def test_child_watch_all_kwargs(self
):
61 cb
= lambda pid
, status
: None
64 res
= GLib
._child
_watch
_add
_get
_args
(priority
=GLib
.PRIORITY_HIGH
, pid
=pid
, function
=cb
, data
=12345)
65 self
.assertEqual(len(res
), 4)
66 self
.assertEqual(res
[0], GLib
.PRIORITY_HIGH
)
67 self
.assertEqual(res
[1], pid
)
68 self
.assertEqual(res
[2], cb
)
69 self
.assertSequenceEqual(res
[3], [12345])
71 def test_child_watch_no_data(self
):
77 self
.loop
= GLib
.MainLoop()
78 argv
= [sys
.executable
, '-c', 'import sys']
79 pid
, stdin
, stdout
, stderr
= GLib
.spawn_async(
80 argv
, flags
=GLib
.SpawnFlags
.DO_NOT_REAP_CHILD
)
82 id = GLib
.child_watch_add(GLib
.PRIORITY_HIGH
, pid
, cb
)
83 self
.assertEqual(self
.loop
.get_context().find_source_by_id(id).priority
,
86 self
.assertEqual(self
.status
, 0)
88 def test_child_watch_with_data(self
):
89 def cb(pid
, status
, data
):
96 self
.loop
= GLib
.MainLoop()
97 argv
= [sys
.executable
, '-c', 'import sys']
98 pid
, stdin
, stdout
, stderr
= GLib
.spawn_async(
99 argv
, flags
=GLib
.SpawnFlags
.DO_NOT_REAP_CHILD
)
100 self
.assertEqual(stdin
, None)
101 self
.assertEqual(stdout
, None)
102 self
.assertEqual(stderr
, None)
103 self
.assertNotEqual(pid
, 0)
105 id = GLib
.child_watch_add(GLib
.PRIORITY_HIGH
, pid
, cb
, 12345)
106 self
.assertEqual(self
.loop
.get_context().find_source_by_id(id).priority
,
109 self
.assertEqual(self
.data
, 12345)
110 self
.assertEqual(self
.status
, 0)
112 def test_spawn_async_fds(self
):
113 pid
, stdin
, stdout
, stderr
= GLib
.spawn_async(
114 ['cat'], flags
=GLib
.SpawnFlags
.SEARCH_PATH
, standard_input
=True,
115 standard_output
=True, standard_error
=True)
116 os
.write(stdin
, b
'hello world!\n')
118 out
= os
.read(stdout
, 50)
120 err
= os
.read(stderr
, 50)
123 self
.assertEqual(out
, b
'hello world!\n')
124 self
.assertEqual(err
, b
'')
126 def test_spawn_async_with_pipes(self
):
127 res
, pid
, stdin
, stdout
, stderr
= GLib
.spawn_async_with_pipes(
128 working_directory
=None,
131 flags
=GLib
.SpawnFlags
.SEARCH_PATH
)
133 os
.write(stdin
, b
'hello world!\n')
135 out
= os
.read(stdout
, 50)
137 err
= os
.read(stderr
, 50)
139 GLib
.spawn_close_pid(pid
)
140 self
.assertEqual(out
, b
'hello world!\n')
141 self
.assertEqual(err
, b
'')
143 def test_spawn_async_envp(self
):
144 pid
, stdin
, stdout
, stderr
= GLib
.spawn_async(
145 ['sh', '-c', 'echo $TEST_VAR'], ['TEST_VAR=moo!'],
146 flags
=GLib
.SpawnFlags
.SEARCH_PATH
, standard_output
=True)
147 self
.assertEqual(stdin
, None)
148 self
.assertEqual(stderr
, None)
149 out
= os
.read(stdout
, 50)
152 self
.assertEqual(out
, b
'moo!\n')
154 def test_backwards_compat_flags(self
):
155 with warnings
.catch_warnings():
156 warnings
.simplefilter('ignore', PyGIDeprecationWarning
)
158 self
.assertEqual(GLib
.SpawnFlags
.DO_NOT_REAP_CHILD
,
159 GLib
.SPAWN_DO_NOT_REAP_CHILD
)