| 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 error
|
| 10 | from core import state
|
| 11 | from core import vm
|
| 12 | from frontend import flag_util
|
| 13 | from frontend import typed_args
|
| 14 | from mycpp import mylib
|
| 15 | from mycpp.mylib import tagswitch, NewDict
|
| 16 |
|
| 17 | from typing import TYPE_CHECKING, cast, Any, Dict, List
|
| 18 |
|
| 19 | if TYPE_CHECKING:
|
| 20 | from display import ui
|
| 21 | from osh.cmd_eval import CommandEvaluator
|
| 22 |
|
| 23 |
|
| 24 | class Shvar(vm._Builtin):
|
| 25 |
|
| 26 | def __init__(self, mem, search_path, cmd_ev):
|
| 27 | # type: (state.Mem, state.SearchPath, CommandEvaluator) -> None
|
| 28 | self.mem = mem
|
| 29 | self.search_path = search_path # to clear PATH
|
| 30 | self.cmd_ev = cmd_ev # To run blocks
|
| 31 |
|
| 32 | def Run(self, cmd_val):
|
| 33 | # type: (cmd_value.Argv) -> int
|
| 34 | _, arg_r = flag_util.ParseCmdVal('shvar',
|
| 35 | cmd_val,
|
| 36 | accept_typed_args=True)
|
| 37 |
|
| 38 | cmd = typed_args.OptionalBlock(cmd_val)
|
| 39 | if not cmd:
|
| 40 | # TODO: I think shvar LANG=C should just mutate
|
| 41 | # But should there be a whitelist?
|
| 42 | raise error.Usage('expected a block', loc.Missing)
|
| 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.EvalCommand(cmd)
|
| 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.EvalCommand(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.RequiredBlock()
|
| 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 = typed_args.OptionalBlock(cmd_val)
|
| 180 | if not cmd:
|
| 181 | raise error.Usage('expected a block', loc.Missing)
|
| 182 |
|
| 183 | with state.ctx_Registers(self.mem):
|
| 184 | unused = self.cmd_ev.EvalCommand(cmd)
|
| 185 |
|
| 186 | # make it "SILENT" in terms of not mutating $?
|
| 187 | # TODO: Revisit this. It might be better to provide the headless shell
|
| 188 | # with a way to SET $? instead. Needs to be tested/prototyped.
|
| 189 | return self.mem.last_status[-1]
|
| 190 |
|
| 191 |
|
| 192 | class Append(vm._Builtin):
|
| 193 | """Push word args onto an List.
|
| 194 |
|
| 195 | Not doing typed args since you can do
|
| 196 |
|
| 197 | :: mylist->append(42)
|
| 198 | """
|
| 199 |
|
| 200 | def __init__(self, mem, errfmt):
|
| 201 | # type: (state.Mem, ui.ErrorFormatter) -> None
|
| 202 | self.mem = mem
|
| 203 | self.errfmt = errfmt
|
| 204 |
|
| 205 | def Run(self, cmd_val):
|
| 206 | # type: (cmd_value.Argv) -> int
|
| 207 |
|
| 208 | # This means we ignore -- , which is consistent
|
| 209 | arg, arg_r = flag_util.ParseCmdVal('append',
|
| 210 | cmd_val,
|
| 211 | accept_typed_args=True)
|
| 212 |
|
| 213 | rd = typed_args.ReaderForProc(cmd_val)
|
| 214 | val = rd.PosValue()
|
| 215 | rd.Done()
|
| 216 |
|
| 217 | UP_val = val
|
| 218 | with tagswitch(val) as case:
|
| 219 | if case(value_e.BashArray):
|
| 220 | val = cast(value.BashArray, UP_val)
|
| 221 | val.strs.extend(arg_r.Rest())
|
| 222 | elif case(value_e.List):
|
| 223 | val = cast(value.List, UP_val)
|
| 224 | typed = [value.Str(s)
|
| 225 | for s in arg_r.Rest()] # type: List[value_t]
|
| 226 | val.items.extend(typed)
|
| 227 | else:
|
| 228 | raise error.TypeErr(val, 'expected List or BashArray',
|
| 229 | loc.Missing)
|
| 230 |
|
| 231 | return 0
|