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