How to use the invoke.Context function in invoke

def _runner(out='', err='', **kwargs):
    klass = kwargs.pop('klass', Dummy)
    runner = klass(Context(config=Config(overrides=kwargs)))
    if 'exits' in kwargs:
        runner.returncode = Mock(return_value=kwargs.pop('exits'))
    out_file = BytesIO(b(out))
    err_file = BytesIO(b(err))
    runner.read_proc_stdout = out_file.read
    runner.read_proc_stderr = err_file.read
    return runner

def _build():
    """
    Build local support docs tree and return the build target dir for cleanup.
    """
    c = Context()
    support = join(dirname(__file__), "_support")
    docs = join(support, "docs")
    build = join(support, "_build")
    command = "sphinx-build -c {} -W {} {}".format(support, docs, build)
    with c.cd(support):
        # Turn off stdin mirroring to avoid irritating pytest.
        c.run(command, in_stream=False)
    return build

def user_kwarg_wins_over_config(self, Local):
            runner = Local.return_value
            config = Config(overrides={"sudo": {"user": "rando"}})
            Context(config=config).sudo("whoami", user="calrissian")
            cmd = "sudo -S -p '[sudo] password: ' -H -u calrissian whoami"
            assert runner.run.call_args[0][0] == cmd

def creates_a_new_Context_from_given_config(self):
            conf = Config(defaults={"foo": "bar"})
            c = Call(_).make_context(conf)
            assert isinstance(c, Context)
            assert c.foo == "bar"

def honors_config(self):
            c = Context(Config(overrides={"run": {"encoding": "UTF-7"}}))
            runner = _Dummy(c)
            runner.default_encoding = Mock(return_value="UTF-not-7")
            runner.run(_)
            assert runner.encoding == "UTF-7"

def triggers_exception_when_command_slow(self):
            before = time.time()
            with raises(CommandTimedOut) as info:
                Local(Context()).run("sleep 5", timeout=0.5)
            after = time.time()
            # Fudge real time check a bit, <=0.5 typically fails due to
            # overhead etc. May need raising further to avoid races? Meh.
            assert (after - before) <= 0.75
            # Sanity checks of the exception obj
            assert info.value.timeout == 0.5
            assert info.value.result.command == "sleep 5"

Watch both doc trees & rebuild them if files change.

    This includes e.g. rebuilding the API docs if the source code changes;
    rebuilding the WWW docs if the README changes; etc.

    Reuses the configuration values ``packaging.package`` or ``tests.package``
    (the former winning over the latter if both defined) when determining which
    source directory to scan for API doc updates.
    """
    # TODO: break back down into generic single-site version, then create split
    # tasks as with docs/www above. Probably wants invoke#63.

    # NOTE: 'www'/'docs' refer to the module level sub-collections. meh.

    # Readme & WWW triggers WWW
    www_c = Context(config=c.config.clone())
    www_c.update(**www.configuration())
    www_handler = make_handler(
        ctx=www_c,
        task_=www["build"],
        regexes=[r"\./README.rst", r"\./sites/www"],
        ignore_regexes=[r".*/\..*\.swp", r"\./sites/www/_build"],
    )

    # Code and docs trigger API
    docs_c = Context(config=c.config.clone())
    docs_c.update(**docs.configuration())
    regexes = [r"\./sites/docs"]
    package = c.get("packaging", {}).get("package", None)
    if package is None:
        package = c.get("tests", {}).get("package", None)
    if package:

def _hang_on_full_pipe(self, pty):
            class Whoops(Exception):
                pass

            runner = Local(Context())
            # Force runner IO thread-body method to raise an exception to mimic
            # real world encoding explosions/etc. When bug is present, this
            # will make the test hang until forcibly terminated.
            runner.handle_stdout = Mock(side_effect=Whoops, __name__="sigh")
            # NOTE: both Darwin (10.10) and Linux (Travis' docker image) have
            # this file. It's plenty large enough to fill most pipe buffers,
            # which is the triggering behavior.
            try:
                runner.run("cat /usr/share/dict/words", pty=pty)
            except ThreadException as e:
                assert len(e.exceptions) == 1
                assert e.exceptions[0].type is Whoops
            else:
                assert False, "Did not receive expected ThreadException!"

def npm_install(repopath, path=".", **kwargs):
    import invoke
    c = invoke.Context()

    with c.cd(repopath):
        with c.cd(path):
            res = c.run("npm install", warn=True)
            # Dump the output to a file to avoid 
            # https://github.com/npm/npm/issues/17331
            deps = c.run("npm ls --json > /tmp/npmdeps.json", warn=True)

    yield AppendStdout(res.stdout)
    yield AppendStderr(res.stderr)
    with open("/tmp/npmdeps.json", "r", encoding='utf-8') as f:
        yield SetProperty("dependencies",
                          list(npmjson2deps(f.read())))

    return True