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

267 lines, 169 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 self.ext_prog = process.ExternalProgram('', self.fd_state, errfmt,
76 util.NullDebugFile())
77
78 def _ExtProc(self, argv):
79 arg_vec = cmd_value.Argv(argv, [loc.Missing] * len(argv), None, None,
80 None, None)
81 argv0_path = None
82 for path_entry in ['/bin', '/usr/bin']:
83 full_path = os.path.join(path_entry, argv[0])
84 if os.path.exists(full_path):
85 argv0_path = full_path
86 break
87 if not argv0_path:
88 argv0_path = argv[0] # fallback that tests failure case
89 thunk = ExternalThunk(self.ext_prog, argv0_path, arg_vec, {})
90 return Process(thunk, self.job_control, self.job_list, self.tracer)
91
92 def testStdinRedirect(self):
93 PATH = '_tmp/one-two.txt'
94 # Write two lines
95 with open(PATH, 'w') as f:
96 f.write('one\ntwo\n')
97
98 # Should get the first line twice, because Pop() closes it!
99
100 r = RedirValue(Id.Redir_Less, runtime.NO_SPID, redir_loc.Fd(0),
101 redirect_arg.Path(PATH))
102
103 class CommandEvaluator(object):
104
105 def RunPendingTraps(self):
106 pass
107
108 cmd_ev = CommandEvaluator()
109
110 err_out = []
111 self.fd_state.Push([r], err_out)
112 line1, _ = read_osh._ReadPortion(pyos.NEWLINE_CH, -1, cmd_ev)
113 self.fd_state.Pop(err_out)
114
115 self.fd_state.Push([r], err_out)
116 line2, _ = read_osh._ReadPortion(pyos.NEWLINE_CH, -1, cmd_ev)
117 self.fd_state.Pop(err_out)
118
119 # sys.stdin.readline() would erroneously return 'two' because of buffering.
120 self.assertEqual('one', line1)
121 self.assertEqual('one', line2)
122
123 def testProcess(self):
124 # 3 fds. Does Python open it? Shell seems to have it too. Maybe it
125 # inherits from the shell.
126 print('FDS BEFORE', os.listdir('/dev/fd'))
127
128 Banner('date')
129 argv = ['date']
130 p = self._ExtProc(argv)
131 why = trace.External(argv)
132 status = p.RunProcess(self.waiter, why)
133 log('date returned %d', status)
134 self.assertEqual(0, status)
135
136 Banner('does-not-exist')
137 p = self._ExtProc(['does-not-exist'])
138 print(p.RunProcess(self.waiter, why))
139
140 # 12 file descriptors open!
141 print('FDS AFTER', os.listdir('/dev/fd'))
142
143 def testPipeline(self):
144 node = _CommandNode('uniq -c', self.arena)
145 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
146 ext_prog=self.ext_prog)
147 print('BEFORE', os.listdir('/dev/fd'))
148
149 p = process.Pipeline(False, self.job_control, self.job_list,
150 self.tracer)
151 p.Add(self._ExtProc(['ls']))
152 p.Add(self._ExtProc(['cut', '-d', '.', '-f', '2']))
153 p.Add(self._ExtProc(['sort']))
154
155 p.AddLast((cmd_ev, node))
156
157 p.StartPipeline(self.waiter)
158 pipe_status = p.RunLastPart(self.waiter, self.fd_state)
159 log('pipe_status: %s', pipe_status)
160
161 print('AFTER', os.listdir('/dev/fd'))
162
163 def testPipeline2(self):
164 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
165 ext_prog=self.ext_prog)
166
167 Banner('ls | cut -d . -f 1 | head')
168 p = process.Pipeline(False, self.job_control, self.job_list,
169 self.tracer)
170 p.Add(self._ExtProc(['ls']))
171 p.Add(self._ExtProc(['cut', '-d', '.', '-f', '1']))
172
173 node = _CommandNode('head', self.arena)
174 p.AddLast((cmd_ev, node))
175
176 p.StartPipeline(self.waiter)
177 print(p.RunLastPart(self.waiter, self.fd_state))
178
179 # Simulating subshell for each command
180 node1 = _CommandNode('ls', self.arena)
181 node2 = _CommandNode('head', self.arena)
182 node3 = _CommandNode('sort --reverse', self.arena)
183
184 thunk1 = process.SubProgramThunk(cmd_ev, node1, self.trap_state, None)
185 thunk2 = process.SubProgramThunk(cmd_ev, node2, self.trap_state, None)
186 thunk3 = process.SubProgramThunk(cmd_ev, node3, self.trap_state, None)
187
188 p = process.Pipeline(False, self.job_control, self.job_list,
189 self.tracer)
190 p.Add(Process(thunk1, self.job_control, self.job_list, self.tracer))
191 p.Add(Process(thunk2, self.job_control, self.job_list, self.tracer))
192 p.Add(Process(thunk3, self.job_control, self.job_list, self.tracer))
193
194 last_thunk = (cmd_ev, _CommandNode('cat', self.arena))
195 p.AddLast(last_thunk)
196
197 p.StartPipeline(self.waiter)
198 print(p.RunLastPart(self.waiter, self.fd_state))
199
200 # TODO: Combine pipelines for other things:
201
202 # echo foo 1>&2 | tee stdout.txt
203 #
204 # foo=$(ls | head)
205 #
206 # foo=$(<<EOF ls | head)
207 # stdin
208 # EOF
209 #
210 # ls | head &
211
212 # Or technically we could fork the whole interpreter for foo|bar|baz and
213 # capture stdout of that interpreter.
214
215 def makeTestPipeline(self, jc):
216 cmd_ev = test_lib.InitCommandEvaluator(arena=self.arena,
217 ext_prog=self.ext_prog)
218
219 pi = process.Pipeline(False, jc, self.job_list, self.tracer)
220
221 node1 = _CommandNode('/bin/echo testpipeline', self.arena)
222 node2 = _CommandNode('cat', self.arena)
223
224 thunk1 = process.SubProgramThunk(cmd_ev, node1, self.trap_state, None)
225 thunk2 = process.SubProgramThunk(cmd_ev, node2, self.trap_state, None)
226
227 pi.Add(Process(thunk1, jc, self.job_list, self.tracer))
228 pi.Add(Process(thunk2, jc, self.job_list, self.tracer))
229
230 return pi
231
232 def testPipelinePgidField(self):
233 jc = FakeJobControl(False)
234
235 pi = self.makeTestPipeline(jc)
236 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
237
238 pi.StartPipeline(self.waiter)
239 # No pgid
240 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
241
242 jc = FakeJobControl(True)
243
244 pi = self.makeTestPipeline(jc)
245 self.assertEqual(process.INVALID_PGID, pi.ProcessGroupId())
246
247 pi.StartPipeline(self.waiter)
248 # first process is the process group leader
249 self.assertEqual(pi.pids[0], pi.ProcessGroupId())
250
251 def testOpen(self):
252 # Disabled because mycpp translation can't handle it. We do this at a
253 # higher layer.
254 return
255
256 # This function used to raise BOTH OSError and IOError because Python 2 is
257 # inconsistent.
258 # We follow Python 3 in preferring OSError.
259 # https://stackoverflow.com/questions/29347790/difference-between-ioerror-and-oserror
260 self.assertRaises(OSError, self.fd_state.Open, '_nonexistent_')
261 self.assertRaises(OSError, self.fd_state.Open, 'metrics/')
262
263
264if __name__ == '__main__':
265 unittest.main()
266
267# vim: sw=4