<commit_before>import os
import platform

from mock import Mock
from spec import Spec, eq_, ok_, skip

from invoke import (
    run, Local, Context, ThreadException, Responder, FailingResponder,
    WatcherError, Failure
)

from _util import assert_cpu_usage


PYPY = platform.python_implementation() == 'PyPy'

class Runner_(Spec):
    def setup(self):
        os.chdir(os.path.join(os.path.dirname(__file__), '_support'))

    class responding:
        # TODO: update respond_*.py so they timeout instead of hanging forever
        # when not being responded to

        def base_case(self):
            # Basic "doesn't explode" test: respond.py will exit nonzero unless
            # this works, causing a Failure.
            watcher = Responder(r"What's the password\?", "<PASSWORD>")
            # Gotta give -u or Python will line-buffer its stdout, so we'll
            # never actually see the prompt.
            run("python -u respond_base.py", watchers=[watcher], hide=True)

        def both_streams(self):
            watchers = [
                Responder("standard out", "with it\n"),
                Responder("standard error", "between chair and keyboard\n"),
            ]
            run("python -u respond_both.py", watchers=watchers, hide=True)

        def watcher_errors_become_Failures(self):
            watcher = FailingResponder(
                pattern=r"What's the password\?",
                response="<PASSWORD>\n",
                sentinel="You're not Citizen Kane!",
            )
            try:
                run("python -u respond_fail.py", watchers=[watcher], hide=True)
            except Failure as e:
                ok_(isinstance(e.reason, WatcherError))
                eq_(e.result.exited, None)
            else:
                assert False, "Did not raise Failure!"

    class stdin_mirroring:
        def piped_stdin_is_not_conflated_with_mocked_stdin(self):
            # Re: GH issue #308
            # Will die on broken-pipe OSError if bug is present.
            run("echo 'lollerskates' | inv -c nested_or_piped foo", hide=True)

        def nested_invoke_sessions_not_conflated_with_mocked_stdin(self):
            # Also re: GH issue #308. This one will just hang forever. Woo!
            run("inv -c nested_or_piped calls_foo", hide=True)

        def isnt_cpu_heavy(self):
            "stdin mirroring isn't CPU-heavy"
            # CPU measurement under PyPy is...rather different. NBD.
            if PYPY:
                skip()
            with assert_cpu_usage(lt=5.0):
                run("python -u busywork.py 10", pty=True, hide=True)

    class IO_hangs:
        "IO hangs"
        def _hang_on_full_pipe(self, pty):
            class Whoops(Exception):
                pass<commit_msg>G R U M P<commit_after>import os
import platform

from mock import Mock
from spec import Spec, eq_, ok_, skip

from invoke import (
    run, Local, Context, ThreadException, Responder, FailingResponder,
    WatcherError, Failure
)

from _util import assert_cpu_usage


PYPY = platform.python_implementation() == 'PyPy'

class Runner_(Spec):
    def setup(self):
        os.chdir(os.path.join(os.path.dirname(__file__), '_support'))

    class responding:
        # TODO: update respond_*.py so they timeout instead of hanging forever
        # when not being responded to

        def base_case(self):
            # Basic "doesn't explode" test: respond.py will exit nonzero unless
            # this works, causing a Failure.
            watcher = Responder(r"What's the password\?", "Rosebud\n")
            # Gotta give -u or Python will line-buffer its stdout, so we'll
            # never actually see the prompt.
            run("python -u respond_base.py", watchers=[watcher], hide=True)

        def both_streams(self):
            watchers = [
                Responder("standard out", "with it\n"),
                Responder("standard error", "between chair and keyboard\n"),
            ]
            run("python -u respond_both.py", watchers=watchers, hide=True)

        def watcher_errors_become_Failures(self):
            watcher = FailingResponder(
                pattern=r"What's the password\?",
                response="Rosebud\n",
                sentinel="You're not Cit<NAME>!",
            )
            try:
                run("python -u respond_fail.py", watchers=[watcher], hide=True)
            except Failure as e:
                ok_(isinstance(e.reason, WatcherError))
                eq_(e.result.exited, None)
            else:
                assert False, "Did not raise Failure!"

    class stdin_mirroring:
        def piped_stdin_is_not_conflated_with_mocked_stdin(self):
            # Re: GH issue #308
            # Will die on broken-pipe OSError if bug is present.
            run("echo 'lollerskates' | inv -c nested_or_piped foo", hide=True)

        def nested_invoke_sessions_not_conflated_with_mocked_stdin(self):
            # Also re: GH issue #308. This one will just hang forever. Woo!
            run("inv -c nested_or_piped calls_foo", hide=True)

        def isnt_cpu_heavy(self):
            "stdin mirroring isn't CPU-heavy"
            # CPU measurement under PyPy is...rather different. NBD.
            if PYPY:
                skip()
            # Python 3.5 has been seen using up to ~6.0s CPU time under Travis
            with assert_cpu_usage(lt=7.0):
                run("python -u busywork.py 10", pty=True, hide=True)

    class IO_hangs:
        "IO hangs"
        def _hang_on_full_pipe(self, pty):
            class Whoops(Exception):
                pass
{'fraction_non_alphanumeric': 0.09355992844364937, 'fraction_numerical': 0.004293381037567084, 'mean_word_length': 2.073666849917537, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1965035', 'n_tokens_mistral': 1700, 'n_tokens_neox': 1580, 'n_words': 469}