OILS / core / process_test.py View on Github | oilshell.org

268 lines, 170 significant
1#!/usr/bin/env python2
2"""process_test.py: Tests for process.py."""
3
4import os
5import unittest
6
7from _devbuild.gen.id_kind_asdl import Id
8from _devbuild.gen.runtime_asdl import (RedirValue, redirect_arg, cmd_value,
9 trace)
10from _devbuild.gen.syntax_asdl import loc, redir_loc
11from asdl import runtime
12from builtin import read_osh
13from builtin import trap_osh
14from core import dev
15from core import process # module under test
16from core import pyos
17from core import test_lib
18from core import ui
19from core import util
20from mycpp.mylib import log
21from core import state
22from mycpp import mylib
23
24import posix_ as posix
25
26Process = process.Process
27ExternalThunk = process.ExternalThunk
28
29
30def Banner(msg):
31 print('-' * 60)
32 print(msg)
33
34
35def _CommandNode(code_str, arena):
36 c_parser = test_lib.InitCommandParser(code_str, arena=arena)
37 return c_parser.ParseLogicalLine()
38
39
40class FakeJobControl(object):
41
42 def __init__(self, enabled):
43 self.enabled = enabled
44
45 def Enabled(self):
46 return self.enabled
47
48
49class ProcessTest(unittest.TestCase):
50
51 def setUp(self):
52 self.arena = test_lib.MakeArena('process_test.py')
53
54 mem = state.Mem('', [], self.arena, [])
55 parse_opts, exec_opts, mutable_opts = state.MakeOpts(mem, None)
56 mem.exec_opts = exec_opts
57
58 state.InitMem(mem, {}, '0.1')
59
60 self.job_control = process.JobControl()
61 self.job_list = process.JobList()
62
63 signal_safe = pyos.InitSignalSafe()
64 self.trap_state = trap_osh.TrapState(signal_safe)
65
66 fd_state = None
67 multi_trace = dev.MultiTracer(posix.getpid(), '', '', '', fd_state)
68 self.tracer = dev.Tracer(None, exec_opts, mutable_opts, mem,
69 mylib.Stderr(), multi_trace)
70 self.waiter = process.Waiter(self.job_list, exec_opts, self.trap_state,
71 self.tracer)
72 errfmt = ui.ErrorFormatter()
73 self.fd_state = process.FdState(errfmt, self.job_control,
74 self.job_list, None, self.tracer, None,
75 exec_opts)
76 self.ext_prog = process.ExternalProgram('', self.fd_state, errfmt,
77 util.NullDebugFile())
78
79 def _ExtProc(self, argv):
80 arg_vec = cmd_value.Argv(argv, [loc.Missing] * len(argv), None, None,
81 None, None)
82 argv0_path = None
83 for path_entry in ['/bin', '/usr/bin']:
84 full_path = os.path.join(path_entry, argv[0])
85 if os.path.exists(full_path):
86 argv0_path = full_path
87 break
88 if not argv0_path:
89 argv0_path = argv[0] # fallback that tests failure case
90 thunk = ExternalThunk(self.ext_prog, argv0_path, arg_vec, {})
91 return Process(thunk, self.job_control, self.job_list, self.tracer)
92
93 def testStdinRedirect(self):
94 PATH = '_tmp/one-two.txt'
95 # Write two lines
96 with open(PATH, 'w') as f:
97 f.write('one\ntwo\n')
98
99 # Should get the first line twice, because Pop() closes it!
100
101 r = RedirValue(Id.Redir_Less, runtime.NO_SPID, redir_loc.Fd(0),
102 redirect_arg.Path(PATH))
103
104 class CommandEvaluator(object):
105
106 def RunPendingTraps(self):
107 pass
108
109 cmd_ev = CommandEvaluator()
110
111 err_out = []
112 self.fd_state.Push([r], err_out)
113 line1, _ = read_osh._ReadPortion(pyos.NEWLINE_CH, -1, cmd_ev)
114 self.fd_state.Pop(err_out)
115
116 self.fd_state.Push([r], err_out)
117 line2, _ = read_osh._ReadPortion(pyos.NEWLINE_CH, -1, cmd_ev)
118 self.fd_state.Pop(err_out)
119
120 # sys.stdin.readline() would erroneously return 'two' because of buffering.
121 self.assertEqual('one', line1)
122 self.assertEqual('one', line2)
123
124 def testProcess(self):
125 # 3 fds. Does Python open it? Shell seems to have it too. Maybe it
126 # inherits from the shell.
127 print('FDS BEFORE', os.listdir('/dev/fd'))
128
129 Banner('date')
130 argv = ['date']
131 p = self._ExtProc(argv)
132 why = trace.External(argv)
133 status = p.RunProcess(self.waiter, why)
134 log('date returned %d', status)
135 self.assertEqual(0, status)
136
137 Banner('does-not-exist')
138 p = self._ExtProc(['does-not-exist'])
139 print(p.RunProcess(self.waiter, why))
140
141 # 12 file descriptors open!
142 print('FDS AFTER', os.listdir('/dev/fd'))
143
144 def testPipeline(self):
145 node = _CommandNode('uniq -c', self.arena)
146 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
147 ext_prog=self.ext_prog)
148 print('BEFORE', os.listdir('/dev/fd'))
149
150 p = process.Pipeline(False, self.job_control, self.job_list,
151 self.tracer)
152 p.Add(self._ExtProc(['ls']))
153 p.Add(self._ExtProc(['cut', '-d', '.', '-f', '2']))
154 p.Add(self._ExtProc(['sort']))
155
156 p.AddLast((cmd_ev, node))
157
158 p.StartPipeline(self.waiter)
159 pipe_status = p.RunLastPart(self.waiter, self.fd_state)
160 log('pipe_status: %s', pipe_status)
161
162 print('AFTER', os.listdir('/dev/fd'))
163
164 def testPipeline2(self):
165 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
166 ext_prog=self.ext_prog)
167
168 Banner('ls | cut -d . -f 1 | head')
169 p = process.Pipeline(False, self.job_control, self.job_list,
170 self.tracer)
171 p.Add(self._ExtProc(['ls']))
172 p.Add(self._ExtProc(['cut', '-d', '.', '-f', '1']))
173
174 node = _CommandNode('head', self.arena)
175 p.AddLast((cmd_ev, node))
176
177 p.StartPipeline(self.waiter)
178 print(p.RunLastPart(self.waiter, self.fd_state))
179
180 # Simulating subshell for each command
181 node1 = _CommandNode('ls', self.arena)
182 node2 = _CommandNode('head', self.arena)
183 node3 = _CommandNode('sort --reverse', self.arena)
184
185 thunk1 = process.SubProgramThunk(cmd_ev, node1, self.trap_state, None)
186 thunk2 = process.SubProgramThunk(cmd_ev, node2, self.trap_state, None)
187 thunk3 = process.SubProgramThunk(cmd_ev, node3, self.trap_state, None)
188
189 p = process.Pipeline(False, self.job_control, self.job_list,
190 self.tracer)
191 p.Add(Process(thunk1, self.job_control, self.job_list, self.tracer))
192 p.Add(Process(thunk2, self.job_control, self.job_list, self.tracer))
193 p.Add(Process(thunk3, self.job_control, self.job_list, self.tracer))
194
195 last_thunk = (cmd_ev, _CommandNode('cat', self.arena))
196 p.AddLast(last_thunk)
197
198 p.StartPipeline(self.waiter)
199 print(p.RunLastPart(self.waiter, self.fd_state))
200
201 # TODO: Combine pipelines for other things:
202
203 # echo foo 1>&2 | tee stdout.txt
204 #
205 # foo=$(ls | head)
206 #
207 # foo=$(<<EOF ls | head)
208 # stdin
209 # EOF
210 #
211 # ls | head &
212
213 # Or technically we could fork the whole interpreter for foo|bar|baz and
214 # capture stdout of that interpreter.
215
216 def makeTestPipeline(self, jc):
217 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
218 ext_prog=self.ext_prog)
219
220 pi = process.Pipeline(False, jc, self.job_list, self.tracer)
221
222 node1 = _CommandNode('/bin/echo testpipeline', self.arena)
223 node2 = _CommandNode('cat', self.arena)
224
225 thunk1 = process.SubProgramThunk(cmd_ev, node1, self.trap_state, None)
226 thunk2 = process.SubProgramThunk(cmd_ev, node2, self.trap_state, None)
227
228 pi.Add(Process(thunk1, jc, self.job_list, self.tracer))
229 pi.Add(Process(thunk2, jc, self.job_list, self.tracer))
230
231 return pi
232
233 def testPipelinePgidField(self):
234 jc = FakeJobControl(False)
235
236 pi = self.makeTestPipeline(jc)
237 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
238
239 pi.StartPipeline(self.waiter)
240 # No pgid
241 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
242
243 jc = FakeJobControl(True)
244
245 pi = self.makeTestPipeline(jc)
246 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
247
248 pi.StartPipeline(self.waiter)
249 # first process is the process group leader
250 self.assertEqual(pi.pids[0], pi.ProcessGroupId())
251
252 def testOpen(self):
253 # Disabled because mycpp translation can't handle it. We do this at a
254 # higher layer.
255 return
256
257 # This function used to raise BOTH OSError and IOError because Python 2 is
258 # inconsistent.
259 # We follow Python 3 in preferring OSError.
260 # https://stackoverflow.com/questions/29347790/difference-between-ioerror-and-oserror
261 self.assertRaises(OSError, self.fd_state.Open, '_nonexistent_')
262 self.assertRaises(OSError, self.fd_state.Open, 'metrics/')
263
264
265if __name__ == '__main__':
266 unittest.main()
267
268# vim: sw=4