1 | """
|
2 | builtin/pure_ysh.py - YSH builtins that don't do I/O.
|
3 | """
|
4 | from __future__ import print_function
|
5 |
|
6 | from _devbuild.gen.runtime_asdl import cmd_value
|
7 | from _devbuild.gen.syntax_asdl import command_t, loc, loc_t
|
8 | from _devbuild.gen.value_asdl import value, value_e, value_t
|
9 | from core import bash_impl
|
10 | from core import error
|
11 | from core import state
|
12 | from core import vm
|
13 | from frontend import flag_util
|
14 | from frontend import typed_args
|
15 | from mycpp import mylib
|
16 | from mycpp.mylib import tagswitch, NewDict
|
17 |
|
18 | from typing import TYPE_CHECKING, cast, Any, Dict, List
|
19 |
|
20 | if TYPE_CHECKING:
|
21 | from core import executor
|
22 | from display import ui
|
23 | from osh.cmd_eval import CommandEvaluator
|
24 |
|
25 |
|
26 | class Shvar(vm._Builtin):
|
27 |
|
28 | def __init__(self, mem, search_path, cmd_ev):
|
29 | # type: (state.Mem, executor.SearchPath, CommandEvaluator) -> None
|
30 | self.mem = mem
|
31 | self.search_path = search_path # to clear PATH
|
32 | self.cmd_ev = cmd_ev # To run blocks
|
33 |
|
34 | def Run(self, cmd_val):
|
35 | # type: (cmd_value.Argv) -> int
|
36 | _, arg_r = flag_util.ParseCmdVal('shvar',
|
37 | cmd_val,
|
38 | accept_typed_args=True)
|
39 |
|
40 | # TODO: I think shvar LANG=C should just mutate
|
41 | # But should there be a whitelist?
|
42 | cmd_frag = typed_args.RequiredBlockAsFrag(cmd_val)
|
43 |
|
44 | vars = NewDict() # type: Dict[str, value_t]
|
45 | args, arg_locs = arg_r.Rest2()
|
46 | if len(args) == 0:
|
47 | raise error.Usage('Expected name=value', loc.Missing)
|
48 |
|
49 | for i, arg in enumerate(args):
|
50 | name, s = mylib.split_once(arg, '=')
|
51 | if s is None:
|
52 | raise error.Usage('Expected name=value', arg_locs[i])
|
53 | v = value.Str(s) # type: value_t
|
54 | vars[name] = v
|
55 |
|
56 | # Important fix: shvar PATH='' { } must make all binaries invisible
|
57 | if name == 'PATH':
|
58 | self.search_path.ClearCache()
|
59 |
|
60 | with state.ctx_Eval(self.mem, None, None, vars):
|
61 | unused = self.cmd_ev.EvalCommandFrag(cmd_frag)
|
62 |
|
63 | return 0
|
64 |
|
65 |
|
66 | class ctx_Context(object):
|
67 | """For ctx push (context) { ... }"""
|
68 |
|
69 | def __init__(self, mem, context):
|
70 | # type: (state.Mem, Dict[str, value_t]) -> None
|
71 | self.mem = mem
|
72 | self.mem.PushContextStack(context)
|
73 |
|
74 | def __enter__(self):
|
75 | # type: () -> None
|
76 | pass
|
77 |
|
78 | def __exit__(self, type, value, traceback):
|
79 | # type: (Any, Any, Any) -> None
|
80 | self.mem.PopContextStack()
|
81 |
|
82 |
|
83 | class Ctx(vm._Builtin):
|
84 |
|
85 | def __init__(self, mem, cmd_ev):
|
86 | # type: (state.Mem, CommandEvaluator) -> None
|
87 | self.mem = mem
|
88 | self.cmd_ev = cmd_ev # To run blocks
|
89 |
|
90 | def _GetContext(self):
|
91 | # type: () -> Dict[str, value_t]
|
92 | ctx = self.mem.GetContext()
|
93 | if ctx is None:
|
94 | raise error.Expr(
|
95 | "Could not find a context. Did you forget to 'ctx push'?",
|
96 | loc.Missing)
|
97 | return ctx
|
98 |
|
99 | def _Push(self, context, block):
|
100 | # type: (Dict[str, value_t], command_t) -> int
|
101 | with ctx_Context(self.mem, context):
|
102 | return self.cmd_ev.EvalCommandFrag(block)
|
103 |
|
104 | def _Set(self, updates):
|
105 | # type: (Dict[str, value_t]) -> int
|
106 | ctx = self._GetContext()
|
107 | ctx.update(updates)
|
108 | return 0
|
109 |
|
110 | def _Emit(self, field, item, blame):
|
111 | # type: (str, value_t, loc_t) -> int
|
112 | ctx = self._GetContext()
|
113 |
|
114 | if field not in ctx:
|
115 | ctx[field] = value.List([])
|
116 |
|
117 | UP_arr = ctx[field]
|
118 | if UP_arr.tag() != value_e.List:
|
119 | raise error.TypeErr(
|
120 | UP_arr,
|
121 | "Expected the context item '%s' to be a List" % (field), blame)
|
122 |
|
123 | arr = cast(value.List, UP_arr)
|
124 | arr.items.append(item)
|
125 |
|
126 | return 0
|
127 |
|
128 | def Run(self, cmd_val):
|
129 | # type: (cmd_value.Argv) -> int
|
130 | rd = typed_args.ReaderForProc(cmd_val)
|
131 | _, arg_r = flag_util.ParseCmdVal('ctx',
|
132 | cmd_val,
|
133 | accept_typed_args=True)
|
134 |
|
135 | verb, verb_loc = arg_r.ReadRequired2(
|
136 | 'Expected a verb (push, set, emit)')
|
137 |
|
138 | if verb == "push":
|
139 | context = rd.PosDict()
|
140 | block = rd.RequiredBlockAsFrag()
|
141 | rd.Done()
|
142 | arg_r.AtEnd()
|
143 |
|
144 | return self._Push(context, block)
|
145 |
|
146 | elif verb == "set":
|
147 | updates = rd.RestNamed()
|
148 | rd.Done()
|
149 | arg_r.AtEnd()
|
150 |
|
151 | return self._Set(updates)
|
152 |
|
153 | elif verb == "emit":
|
154 | field, field_loc = arg_r.ReadRequired2(
|
155 | "A target field is required")
|
156 | item = rd.PosValue()
|
157 | rd.Done()
|
158 | arg_r.AtEnd()
|
159 |
|
160 | return self._Emit(field, item, field_loc)
|
161 |
|
162 | else:
|
163 | raise error.Usage("Unknown verb '%s'" % verb, verb_loc)
|
164 |
|
165 |
|
166 | class PushRegisters(vm._Builtin):
|
167 |
|
168 | def __init__(self, mem, cmd_ev):
|
169 | # type: (state.Mem, CommandEvaluator) -> None
|
170 | self.mem = mem
|
171 | self.cmd_ev = cmd_ev # To run blocks
|
172 |
|
173 | def Run(self, cmd_val):
|
174 | # type: (cmd_value.Argv) -> int
|
175 | _, arg_r = flag_util.ParseCmdVal('push-registers',
|
176 | cmd_val,
|
177 | accept_typed_args=True)
|
178 |
|
179 | cmd_frag = typed_args.RequiredBlockAsFrag(cmd_val)
|
180 |
|
181 | with state.ctx_Registers(self.mem):
|
182 | unused = self.cmd_ev.EvalCommandFrag(cmd_frag)
|
183 |
|
184 | # make it "SILENT" in terms of not mutating $?
|
185 | # TODO: Revisit this. It might be better to provide the headless shell
|
186 | # with a way to SET $? instead. Needs to be tested/prototyped.
|
187 | return self.mem.last_status[-1]
|
188 |
|
189 |
|
190 | class Append(vm._Builtin):
|
191 | """Push word args onto an List.
|
192 |
|
193 | Not doing typed args since you can do
|
194 |
|
195 | :: mylist->append(42)
|
196 | """
|
197 |
|
198 | def __init__(self, mem, errfmt):
|
199 | # type: (state.Mem, ui.ErrorFormatter) -> None
|
200 | self.mem = mem
|
201 | self.errfmt = errfmt
|
202 |
|
203 | def Run(self, cmd_val):
|
204 | # type: (cmd_value.Argv) -> int
|
205 |
|
206 | # This means we ignore -- , which is consistent
|
207 | _, arg_r = flag_util.ParseCmdVal('append',
|
208 | cmd_val,
|
209 | accept_typed_args=True)
|
210 |
|
211 | rd = typed_args.ReaderForProc(cmd_val)
|
212 | val = rd.PosValue()
|
213 | rd.Done()
|
214 |
|
215 | UP_val = val
|
216 | with tagswitch(val) as case:
|
217 | if case(value_e.BashArray):
|
218 | val = cast(value.BashArray, UP_val)
|
219 | bash_impl.BashArray_AppendValues(val, arg_r.Rest())
|
220 | elif case(value_e.List):
|
221 | val = cast(value.List, UP_val)
|
222 | typed = [value.Str(s)
|
223 | for s in arg_r.Rest()] # type: List[value_t]
|
224 | val.items.extend(typed)
|
225 | else:
|
226 | raise error.TypeErr(val, 'expected List or BashArray',
|
227 | loc.Missing)
|
228 |
|
229 | return 0
|