Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/context.py: 27%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import os
2import re
3from contextlib import contextmanager
4from itertools import cycle
5from os import PathLike
6from typing import (
7 Any,
8 Generator,
9 Iterator,
10 List,
11 Optional,
12 Union,
13)
14from unittest.mock import Mock
16from .config import Config, DataProxy
17from .exceptions import AuthFailure, Failure, ResponseNotAccepted
18from .runners import Result, Runner
19from .watchers import FailingResponder
22class Context(DataProxy):
23 """
24 Context-aware API wrapper & state-passing object.
26 `.Context` objects are created during command-line parsing (or, if desired,
27 by hand) and used to share parser and configuration state with executed
28 tasks (see :ref:`why-context`).
30 Specifically, the class offers wrappers for core API calls (such as `.run`)
31 which take into account CLI parser flags, configuration files, and/or
32 changes made at runtime. It also acts as a proxy for its `~.Context.config`
33 attribute - see that attribute's documentation for details.
35 Instances of `.Context` may be shared between tasks when executing
36 sub-tasks - either the same context the caller was given, or an altered
37 copy thereof (or, theoretically, a brand new one).
39 .. versionadded:: 1.0
40 """
42 # NOTE: sometime after Sphinx 1.7, autodoc stopped being able to see
43 # doc-comments inside __init__ (or something equivalent, anyway). Moving
44 # the type definitions up here seems to work better and /shouldn't/ mess up
45 # the DataProxy magic going on...
47 #: A list of commands to run (via "&&") before the main argument to any
48 #: `run` or `sudo` calls. Note that the primary API for manipulating
49 #: this list is `prefix`; see its docs for details.
50 command_prefixes: List[str]
51 #: A list of directories to 'cd' into before running commands with
52 #: `run` or `sudo`; intended for management via `cd`, please see its
53 #: docs for details.
54 command_cwds: List[str]
55 #: The CLI parser's 'remainder' value (text given after a standalone
56 #: ``--`` in the command line), or the empty string if not given.
57 remainder: str
59 def __init__(
60 self,
61 config: Optional[Config] = None,
62 remainder: str = "",
63 ) -> None:
64 """
65 :param config:
66 `.Config` object to use as the base configuration.
68 Defaults to an anonymous/default `.Config` instance.
70 :param remainder:
71 The invoking program's :ref:`parser remainder <remainder>` value,
72 if any was obtained.
73 """
74 config = config if config is not None else Config()
75 self._set(
76 _config=config,
77 command_prefixes=[],
78 command_cwds=[],
79 remainder=remainder,
80 )
82 @property
83 def config(self) -> Config:
84 """
85 The fully merged `.Config` object appropriate for this context.
87 `.Config` settings (see their documentation for details) may be
88 accessed like dictionary keys (``c.config['foo']``) or object
89 attributes (``c.config.foo``).
91 As a convenience shorthand, the `.Context` object proxies to its
92 ``config`` attribute in the same way - e.g. ``c['foo']`` or
93 ``c.foo`` returns the same value as ``c.config['foo']``.
94 """
95 # Allows Context to expose a .config attribute even though DataProxy
96 # otherwise considers it a config key.
97 return self._config
99 @config.setter
100 def config(self, value: Config) -> None:
101 # NOTE: mostly used by client libraries needing to tweak a Context's
102 # config at execution time; i.e. a Context subclass that bears its own
103 # unique data may want to be stood up when parameterizing/expanding a
104 # call list at start of a session, with the final config filled in at
105 # runtime.
106 self._set(_config=value)
108 def run(self, command: str, **kwargs: Any) -> Result:
109 """
110 Execute a local shell command, honoring config options.
112 Specifically, this method instantiates a `.Runner` subclass (according
113 to the ``runner`` config option; default is `.Local`) and calls its
114 ``.run`` method with ``command`` and ``kwargs``.
116 See `.Runner.run` for details on ``command`` and the available keyword
117 arguments.
119 .. versionadded:: 1.0
120 """
121 runner = self.config.runners.local(self)
122 return self._run(runner, command, **kwargs)
124 # NOTE: broken out of run() to allow for runner class injection in
125 # Fabric/etc, which needs to juggle multiple runner class types (local and
126 # remote).
127 def _run(self, runner: "Runner", command: str, **kwargs: Any) -> Result:
128 command = self._prefix_commands(command)
129 return runner.run(command, **kwargs)
131 def sudo(self, command: str, **kwargs: Any) -> Result:
132 """
133 Execute a shell command via ``sudo`` with password auto-response.
135 **Basics**
137 This method is identical to `run` but adds a handful of
138 convenient behaviors around invoking the ``sudo`` program. It doesn't
139 do anything users could not do themselves by wrapping `run`, but the
140 use case is too common to make users reinvent these wheels themselves.
142 .. note::
143 If you intend to respond to sudo's password prompt by hand, just
144 use ``run("sudo command")`` instead! The autoresponding features in
145 this method will just get in your way.
147 Specifically, `sudo`:
149 * Places a `.FailingResponder` into the ``watchers`` kwarg (see
150 :doc:`/concepts/watchers`) which:
152 * searches for the configured ``sudo`` password prompt;
153 * responds with the configured sudo password (``sudo.password``
154 from the :doc:`configuration </concepts/configuration>`);
155 * can tell when that response causes an authentication failure
156 (e.g. if the system requires a password and one was not
157 configured), and raises `.AuthFailure` if so.
159 * Builds a ``sudo`` command string using the supplied ``command``
160 argument, prefixed by various flags (see below);
161 * Executes that command via a call to `run`, returning the result.
163 **Flags used**
165 ``sudo`` flags used under the hood include:
167 - ``-S`` to allow auto-responding of password via stdin;
168 - ``-p <prompt>`` to explicitly state the prompt to use, so we can be
169 sure our auto-responder knows what to look for;
170 - ``-u <user>`` if ``user`` is not ``None``, to execute the command as
171 a user other than ``root``;
172 - When ``-u`` is present, ``-H`` is also added, to ensure the
173 subprocess has the requested user's ``$HOME`` set properly.
175 **Configuring behavior**
177 There are a couple of ways to change how this method behaves:
179 - Because it wraps `run`, it honors all `run` config parameters and
180 keyword arguments, in the same way that `run` does.
182 - Thus, invocations such as ``c.sudo('command', echo=True)`` are
183 possible, and if a config layer (such as a config file or env
184 var) specifies that e.g. ``run.warn = True``, that too will take
185 effect under `sudo`.
187 - `sudo` has its own set of keyword arguments (see below) and they are
188 also all controllable via the configuration system, under the
189 ``sudo.*`` tree.
191 - Thus you could, for example, pre-set a sudo user in a config
192 file; such as an ``invoke.json`` containing ``{"sudo": {"user":
193 "someuser"}}``.
195 :param str password: Runtime override for ``sudo.password``.
196 :param str user: Runtime override for ``sudo.user``.
198 .. versionadded:: 1.0
199 """
200 runner = self.config.runners.local(self)
201 return self._sudo(runner, command, **kwargs)
203 # NOTE: this is for runner injection; see NOTE above _run().
204 def _sudo(self, runner: "Runner", command: str, **kwargs: Any) -> Result:
205 prompt = self.config.sudo.prompt
206 password = kwargs.pop("password", self.config.sudo.password)
207 user = kwargs.pop("user", self.config.sudo.user)
208 env = kwargs.get("env", {})
209 # TODO: allow subclassing for 'get the password' so users who REALLY
210 # want lazy runtime prompting can have it easily implemented.
211 # TODO: want to print a "cleaner" echo with just 'sudo <command>'; but
212 # hard to do as-is, obtaining config data from outside a Runner one
213 # holds is currently messy (could fix that), if instead we manually
214 # inspect the config ourselves that duplicates logic. NOTE: once we
215 # figure that out, there is an existing, would-fail-if-not-skipped test
216 # for this behavior in test/context.py.
217 # TODO: once that is done, though: how to handle "full debug" output
218 # exactly (display of actual, real full sudo command w/ -S and -p), in
219 # terms of API/config? Impl is easy, just go back to passing echo
220 # through to 'run'...
221 user_flags = ""
222 if user is not None:
223 user_flags = "-H -u {} ".format(user)
224 env_flags = ""
225 if env:
226 env_flags = "--preserve-env='{}' ".format(",".join(env.keys()))
227 command = self._prefix_commands(command)
228 cmd_str = "sudo -S -p '{}' {}{}{}".format(
229 prompt, env_flags, user_flags, command
230 )
231 watcher = FailingResponder(
232 pattern=re.escape(prompt),
233 response="{}\n".format(password),
234 sentinel="Sorry, try again.\n",
235 )
236 # Ensure we merge any user-specified watchers with our own.
237 # NOTE: If there are config-driven watchers, we pull those up to the
238 # kwarg level; that lets us merge cleanly without needing complex
239 # config-driven "override vs merge" semantics.
240 # TODO: if/when those semantics are implemented, use them instead.
241 # NOTE: config value for watchers defaults to an empty list; and we
242 # want to clone it to avoid actually mutating the config.
243 watchers = kwargs.pop("watchers", list(self.config.run.watchers))
244 watchers.append(watcher)
245 try:
246 return runner.run(cmd_str, watchers=watchers, **kwargs)
247 except Failure as failure:
248 # Transmute failures driven by our FailingResponder, into auth
249 # failures - the command never even ran.
250 # TODO: wants to be a hook here for users that desire "override a
251 # bad config value for sudo.password" manual input
252 # NOTE: as noted in #294 comments, we MAY in future want to update
253 # this so run() is given ability to raise AuthFailure on its own.
254 # For now that has been judged unnecessary complexity.
255 if isinstance(failure.reason, ResponseNotAccepted):
256 # NOTE: not bothering with 'reason' here, it's pointless.
257 error = AuthFailure(result=failure.result, prompt=prompt)
258 raise error
259 # Reraise for any other error so it bubbles up normally.
260 else:
261 raise
263 # TODO: wonder if it makes sense to move this part of things inside Runner,
264 # which would grow a `prefixes` and `cwd` init kwargs or similar. The less
265 # that's stuffed into Context, probably the better.
266 def _prefix_commands(self, command: str) -> str:
267 """
268 Prefixes ``command`` with all prefixes found in ``command_prefixes``.
270 ``command_prefixes`` is a list of strings which is modified by the
271 `prefix` context manager.
272 """
273 prefixes = list(self.command_prefixes)
274 current_directory = self.cwd
275 if current_directory:
276 prefixes.insert(0, "cd {}".format(current_directory))
278 return " && ".join(prefixes + [command])
280 @contextmanager
281 def prefix(self, command: str) -> Generator[None, None, None]:
282 """
283 Prefix all nested `run`/`sudo` commands with given command plus ``&&``.
285 Most of the time, you'll want to be using this alongside a shell script
286 which alters shell state, such as ones which export or alter shell
287 environment variables.
289 For example, one of the most common uses of this tool is with the
290 ``workon`` command from `virtualenvwrapper
291 <https://virtualenvwrapper.readthedocs.io/en/latest/>`_::
293 with c.prefix('workon myvenv'):
294 c.run('./manage.py migrate')
296 In the above snippet, the actual shell command run would be this::
298 $ workon myvenv && ./manage.py migrate
300 This context manager is compatible with `cd`, so if your virtualenv
301 doesn't ``cd`` in its ``postactivate`` script, you could do the
302 following::
304 with c.cd('/path/to/app'):
305 with c.prefix('workon myvenv'):
306 c.run('./manage.py migrate')
307 c.run('./manage.py loaddata fixture')
309 Which would result in executions like so::
311 $ cd /path/to/app && workon myvenv && ./manage.py migrate
312 $ cd /path/to/app && workon myvenv && ./manage.py loaddata fixture
314 Finally, as alluded to above, `prefix` may be nested if desired, e.g.::
316 with c.prefix('workon myenv'):
317 c.run('ls')
318 with c.prefix('source /some/script'):
319 c.run('touch a_file')
321 The result::
323 $ workon myenv && ls
324 $ workon myenv && source /some/script && touch a_file
326 Contrived, but hopefully illustrative.
328 .. versionadded:: 1.0
329 """
330 self.command_prefixes.append(command)
331 try:
332 yield
333 finally:
334 self.command_prefixes.pop()
336 @property
337 def cwd(self) -> str:
338 """
339 Return the current working directory, accounting for uses of `cd`.
341 .. versionadded:: 1.0
342 """
343 if not self.command_cwds:
344 # TODO: should this be None? Feels cleaner, though there may be
345 # benefits to it being an empty string, such as relying on a no-arg
346 # `cd` typically being shorthand for "go to user's $HOME".
347 return ""
349 # get the index for the subset of paths starting with the last / or ~
350 for i, path in reversed(list(enumerate(self.command_cwds))):
351 if path.startswith("~") or path.startswith("/"):
352 break
354 # TODO: see if there's a stronger "escape this path" function somewhere
355 # we can reuse. e.g., escaping tildes or slashes in filenames.
356 paths = [path.replace(" ", r"\ ") for path in self.command_cwds[i:]]
357 return str(os.path.join(*paths))
359 @contextmanager
360 def cd(self, path: Union[PathLike, str]) -> Generator[None, None, None]:
361 """
362 Context manager that keeps directory state when executing commands.
364 Any calls to `run`, `sudo`, within the wrapped block will implicitly
365 have a string similar to ``"cd <path> && "`` prefixed in order to give
366 the sense that there is actually statefulness involved.
368 Because use of `cd` affects all such invocations, any code making use
369 of the `cwd` property will also be affected by use of `cd`.
371 Like the actual 'cd' shell builtin, `cd` may be called with relative
372 paths (keep in mind that your default starting directory is your user's
373 ``$HOME``) and may be nested as well.
375 Below is a "normal" attempt at using the shell 'cd', which doesn't work
376 since all commands are executed in individual subprocesses -- state is
377 **not** kept between invocations of `run` or `sudo`::
379 c.run('cd /var/www')
380 c.run('ls')
382 The above snippet will list the contents of the user's ``$HOME``
383 instead of ``/var/www``. With `cd`, however, it will work as expected::
385 with c.cd('/var/www'):
386 c.run('ls') # Turns into "cd /var/www && ls"
388 Finally, a demonstration (see inline comments) of nesting::
390 with c.cd('/var/www'):
391 c.run('ls') # cd /var/www && ls
392 with c.cd('website1'):
393 c.run('ls') # cd /var/www/website1 && ls
395 .. note::
396 Space characters will be escaped automatically to make dealing with
397 such directory names easier.
399 .. versionadded:: 1.0
400 .. versionchanged:: 1.5
401 Explicitly cast the ``path`` argument (the only argument) to a
402 string; this allows any object defining ``__str__`` to be handed in
403 (such as the various ``Path`` objects out there), and not just
404 string literals.
405 """
406 path = str(path)
407 self.command_cwds.append(path)
408 try:
409 yield
410 finally:
411 self.command_cwds.pop()
414class MockContext(Context):
415 """
416 A `.Context` whose methods' return values can be predetermined.
418 Primarily useful for testing Invoke-using codebases.
420 .. note::
421 This class wraps its ``run``, etc methods in `unittest.mock.Mock`
422 objects. This allows you to easily assert that the methods (still
423 returning the values you prepare them with) were actually called.
425 .. note::
426 Methods not given `Results <.Result>` to yield will raise
427 ``NotImplementedError`` if called (since the alternative is to call the
428 real underlying method - typically undesirable when mocking.)
430 .. versionadded:: 1.0
431 .. versionchanged:: 1.5
432 Added ``Mock`` wrapping of ``run`` and ``sudo``.
433 """
435 def __init__(self, config: Optional[Config] = None, **kwargs: Any) -> None:
436 """
437 Create a ``Context``-like object whose methods yield `.Result` objects.
439 :param config:
440 A Configuration object to use. Identical in behavior to `.Context`.
442 :param run:
443 A data structure indicating what `.Result` objects to return from
444 calls to the instantiated object's `~.Context.run` method (instead
445 of actually executing the requested shell command).
447 Specifically, this kwarg accepts:
449 - A single `.Result` object.
450 - A boolean; if True, yields a `.Result` whose ``exited`` is ``0``,
451 and if False, ``1``.
452 - An iterable of the above values, which will be returned on each
453 subsequent call to ``.run`` (the first item on the first call,
454 the second on the second call, etc).
455 - A dict mapping command strings or compiled regexen to the above
456 values (including an iterable), allowing specific
457 call-and-response semantics instead of assuming a call order.
459 :param sudo:
460 Identical to ``run``, but whose values are yielded from calls to
461 `~.Context.sudo`.
463 :param bool repeat:
464 A flag determining whether results yielded by this class' methods
465 repeat or are consumed.
467 For example, when a single result is indicated, it will normally
468 only be returned once, causing ``NotImplementedError`` afterwards.
469 But when ``repeat=True`` is given, that result is returned on
470 every call, forever.
472 Similarly, iterable results are normally exhausted once, but when
473 this setting is enabled, they are wrapped in `itertools.cycle`.
475 Default: ``True``.
477 :raises:
478 ``TypeError``, if the values given to ``run`` or other kwargs
479 aren't of the expected types.
481 .. versionchanged:: 1.5
482 Added support for boolean and string result values.
483 .. versionchanged:: 1.5
484 Added support for regex dict keys.
485 .. versionchanged:: 1.5
486 Added the ``repeat`` keyword argument.
487 .. versionchanged:: 2.0
488 Changed ``repeat`` default value from ``False`` to ``True``.
489 """
490 # Set up like any other Context would, with the config
491 super().__init__(config)
492 # Pull out behavioral kwargs
493 self._set("__repeat", kwargs.pop("repeat", True))
494 # The rest must be things like run/sudo - mock Context method info
495 for method, results in kwargs.items():
496 # For each possible value type, normalize to iterable of Result
497 # objects (possibly repeating).
498 singletons = (Result, bool, str)
499 if isinstance(results, dict):
500 for key, value in results.items():
501 results[key] = self._normalize(value)
502 elif isinstance(results, singletons) or hasattr(
503 results, "__iter__"
504 ):
505 results = self._normalize(results)
506 # Unknown input value: cry
507 else:
508 err = "Not sure how to yield results from a {!r}"
509 raise TypeError(err.format(type(results)))
510 # Save results for use by the method
511 self._set("__{}".format(method), results)
512 # Wrap the method in a Mock
513 self._set(method, Mock(wraps=getattr(self, method)))
515 def _normalize(self, value: Any) -> Iterator[Any]:
516 # First turn everything into an iterable
517 if not hasattr(value, "__iter__") or isinstance(value, str):
518 value = [value]
519 # Then turn everything within into a Result
520 results = []
521 for obj in value:
522 if isinstance(obj, bool):
523 obj = Result(exited=0 if obj else 1)
524 elif isinstance(obj, str):
525 obj = Result(obj)
526 results.append(obj)
527 # Finally, turn that iterable into an iteratOR, depending on repeat
528 return cycle(results) if getattr(self, "__repeat") else iter(results)
530 # TODO: _maybe_ make this more metaprogrammy/flexible (using __call__ etc)?
531 # Pretty worried it'd cause more hard-to-debug issues than it's presently
532 # worth. Maybe in situations where Context grows a _lot_ of methods (e.g.
533 # in Fabric 2; though Fabric could do its own sub-subclass in that case...)
535 def _yield_result(self, attname: str, command: str) -> Result:
536 try:
537 obj = getattr(self, attname)
538 # Dicts need to try direct lookup or regex matching
539 if isinstance(obj, dict):
540 try:
541 obj = obj[command]
542 except KeyError:
543 # TODO: could optimize by skipping this if not any regex
544 # objects in keys()?
545 for key, value in obj.items():
546 if hasattr(key, "match") and key.match(command):
547 obj = value
548 break
549 else:
550 # Nope, nothing did match.
551 raise KeyError
552 # Here, the value was either never a dict or has been extracted
553 # from one, so we can assume it's an iterable of Result objects due
554 # to work done by __init__.
555 result: Result = next(obj)
556 # Populate Result's command string with what matched unless
557 # explicitly given
558 if not result.command:
559 result.command = command
560 return result
561 except (AttributeError, IndexError, KeyError, StopIteration):
562 # raise_from(NotImplementedError(command), None)
563 raise NotImplementedError(command)
565 def run(self, command: str, *args: Any, **kwargs: Any) -> Result:
566 # TODO: perform more convenience stuff associating args/kwargs with the
567 # result? E.g. filling in .command, etc? Possibly useful for debugging
568 # if one hits unexpected-order problems with what they passed in to
569 # __init__.
570 return self._yield_result("__run", command)
572 def sudo(self, command: str, *args: Any, **kwargs: Any) -> Result:
573 # TODO: this completely nukes the top-level behavior of sudo(), which
574 # could be good or bad, depending. Most of the time I think it's good.
575 # No need to supply dummy password config, etc.
576 # TODO: see the TODO from run() re: injecting arg/kwarg values
577 return self._yield_result("__sudo", command)
579 def set_result_for(
580 self, attname: str, command: str, result: Result
581 ) -> None:
582 """
583 Modify the stored mock results for given ``attname`` (e.g. ``run``).
585 This is similar to how one instantiates `MockContext` with a ``run`` or
586 ``sudo`` dict kwarg. For example, this::
588 mc = MockContext(run={'mycommand': Result("mystdout")})
589 assert mc.run('mycommand').stdout == "mystdout"
591 is functionally equivalent to this::
593 mc = MockContext()
594 mc.set_result_for('run', 'mycommand', Result("mystdout"))
595 assert mc.run('mycommand').stdout == "mystdout"
597 `set_result_for` is mostly useful for modifying an already-instantiated
598 `MockContext`, such as one created by test setup or helper methods.
600 .. versionadded:: 1.0
601 """
602 attname = "__{}".format(attname)
603 heck = TypeError(
604 "Can't update results for non-dict or nonexistent mock results!"
605 )
606 # Get value & complain if it's not a dict.
607 # TODO: should we allow this to set non-dict values too? Seems vaguely
608 # pointless, at that point, just make a new MockContext eh?
609 try:
610 value = getattr(self, attname)
611 except AttributeError:
612 raise heck
613 if not isinstance(value, dict):
614 raise heck
615 # OK, we're good to modify, so do so.
616 value[command] = self._normalize(result)