1# -*- Mode: Python -*- 2 3import sys 4import os 5import unittest 6import warnings 7 8import pytest 9 10from gi.repository import GLib 11from gi import PyGIDeprecationWarning 12 13from .helper import capture_gi_deprecation_warnings 14 15 16def test_child_watch_add_get_args_various(): 17 cb = lambda pid, status: None 18 get_args = GLib._child_watch_add_get_args 19 pid = 42 20 with capture_gi_deprecation_warnings(): 21 assert get_args(pid, cb, 2) == (0, pid, cb, (2,)) 22 23 with pytest.raises(TypeError): 24 get_args(pid, cb, 2, 3, 4) 25 26 assert get_args(0, pid, 2, 3, function=cb) == (0, pid, cb, (2, 3)) 27 assert get_args(0, pid, cb, 2, 3) == (0, pid, cb, (2, 3)) 28 assert get_args(0, pid, cb, data=99) == (0, pid, cb, (99,)) 29 30 with pytest.raises(TypeError): 31 get_args(0, pid, 24) 32 33 with pytest.raises(TypeError): 34 get_args(0, pid, cb, 2, 3, data=99) 35 36 37@unittest.skipIf(os.name == "nt", "not on Windows") 38class TestProcess(unittest.TestCase): 39 40 def test_deprecated_child_watch_no_data(self): 41 cb = lambda pid, status: None 42 pid = object() 43 with warnings.catch_warnings(record=True) as w: 44 warnings.simplefilter('always') 45 res = GLib._child_watch_add_get_args(pid, cb) 46 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning)) 47 48 self.assertEqual(len(res), 4) 49 self.assertEqual(res[0], GLib.PRIORITY_DEFAULT) 50 self.assertEqual(res[1], pid) 51 self.assertTrue(callable(cb)) 52 self.assertSequenceEqual(res[3], []) 53 54 def test_deprecated_child_watch_data_priority(self): 55 cb = lambda pid, status: None 56 pid = object() 57 with warnings.catch_warnings(record=True) as w: 58 warnings.simplefilter('always') 59 res = GLib._child_watch_add_get_args(pid, cb, 12345, GLib.PRIORITY_HIGH) 60 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning)) 61 62 self.assertEqual(len(res), 4) 63 self.assertEqual(res[0], GLib.PRIORITY_HIGH) 64 self.assertEqual(res[1], pid) 65 self.assertEqual(res[2], cb) 66 self.assertSequenceEqual(res[3], [12345]) 67 68 def test_deprecated_child_watch_data_priority_kwargs(self): 69 cb = lambda pid, status: None 70 pid = object() 71 with warnings.catch_warnings(record=True) as w: 72 warnings.simplefilter('always') 73 res = GLib._child_watch_add_get_args(pid, cb, priority=GLib.PRIORITY_HIGH, data=12345) 74 self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning)) 75 76 self.assertEqual(len(res), 4) 77 self.assertEqual(res[0], GLib.PRIORITY_HIGH) 78 self.assertEqual(res[1], pid) 79 self.assertEqual(res[2], cb) 80 self.assertSequenceEqual(res[3], [12345]) 81 82 @unittest.expectedFailure # using keyword args is fully supported by PyGObject machinery 83 def test_child_watch_all_kwargs(self): 84 cb = lambda pid, status: None 85 pid = object() 86 87 res = GLib._child_watch_add_get_args(priority=GLib.PRIORITY_HIGH, pid=pid, function=cb, data=12345) 88 self.assertEqual(len(res), 4) 89 self.assertEqual(res[0], GLib.PRIORITY_HIGH) 90 self.assertEqual(res[1], pid) 91 self.assertEqual(res[2], cb) 92 self.assertSequenceEqual(res[3], [12345]) 93 94 def test_child_watch_no_data(self): 95 def cb(pid, status): 96 self.status = status 97 self.loop.quit() 98 99 self.status = None 100 self.loop = GLib.MainLoop() 101 argv = [sys.executable, '-c', 'import sys'] 102 pid, stdin, stdout, stderr = GLib.spawn_async( 103 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD) 104 pid.close() 105 id = GLib.child_watch_add(GLib.PRIORITY_HIGH, pid, cb) 106 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority, 107 GLib.PRIORITY_HIGH) 108 self.loop.run() 109 self.assertEqual(self.status, 0) 110 111 def test_child_watch_with_data(self): 112 def cb(pid, status, data): 113 self.status = status 114 self.data = data 115 self.loop.quit() 116 117 self.data = None 118 self.status = None 119 self.loop = GLib.MainLoop() 120 argv = [sys.executable, '-c', 'import sys'] 121 pid, stdin, stdout, stderr = GLib.spawn_async( 122 argv, flags=GLib.SpawnFlags.DO_NOT_REAP_CHILD) 123 self.assertEqual(stdin, None) 124 self.assertEqual(stdout, None) 125 self.assertEqual(stderr, None) 126 self.assertNotEqual(pid, 0) 127 pid.close() 128 id = GLib.child_watch_add(GLib.PRIORITY_HIGH, pid, cb, 12345) 129 self.assertEqual(self.loop.get_context().find_source_by_id(id).priority, 130 GLib.PRIORITY_HIGH) 131 self.loop.run() 132 self.assertEqual(self.data, 12345) 133 self.assertEqual(self.status, 0) 134 135 def test_spawn_async_fds(self): 136 pid, stdin, stdout, stderr = GLib.spawn_async( 137 ['cat'], flags=GLib.SpawnFlags.SEARCH_PATH, standard_input=True, 138 standard_output=True, standard_error=True) 139 os.write(stdin, b'hello world!\n') 140 os.close(stdin) 141 out = os.read(stdout, 50) 142 os.close(stdout) 143 err = os.read(stderr, 50) 144 os.close(stderr) 145 pid.close() 146 self.assertEqual(out, b'hello world!\n') 147 self.assertEqual(err, b'') 148 149 def test_spawn_async_with_pipes(self): 150 res, pid, stdin, stdout, stderr = GLib.spawn_async_with_pipes( 151 working_directory=None, 152 argv=['cat'], 153 envp=None, 154 flags=GLib.SpawnFlags.SEARCH_PATH) 155 156 os.write(stdin, b'hello world!\n') 157 os.close(stdin) 158 out = os.read(stdout, 50) 159 os.close(stdout) 160 err = os.read(stderr, 50) 161 os.close(stderr) 162 GLib.spawn_close_pid(pid) 163 self.assertEqual(out, b'hello world!\n') 164 self.assertEqual(err, b'') 165 166 def test_spawn_async_envp(self): 167 pid, stdin, stdout, stderr = GLib.spawn_async( 168 ['sh', '-c', 'echo $TEST_VAR'], ['TEST_VAR=moo!'], 169 flags=GLib.SpawnFlags.SEARCH_PATH, standard_output=True) 170 self.assertEqual(stdin, None) 171 self.assertEqual(stderr, None) 172 out = os.read(stdout, 50) 173 os.close(stdout) 174 pid.close() 175 self.assertEqual(out, b'moo!\n') 176 177 def test_backwards_compat_flags(self): 178 with warnings.catch_warnings(): 179 warnings.simplefilter('ignore', PyGIDeprecationWarning) 180 181 self.assertEqual(GLib.SpawnFlags.DO_NOT_REAP_CHILD, 182 GLib.SPAWN_DO_NOT_REAP_CHILD) 183