Invoke: pytest does not like invoke

[anarcat]

with a simple task/test like this:

@task
def ipv6_slaac(con, ipv6_subnet, mac, hide=True):
    '''compute IPv6 SLAAC address from subnet and MAC address

    This uses the ipv6calc command.

    .. TODO:: rewrite in python-only?
    '''
    command = ['ipv6calc', '--action', 'prefixmac2ipv6',
               '--in', 'prefix+mac', '--out', 'ipv6',
               ipv6_subnet, mac]
    logging.debug('manual SLAAC allocation with: %s', ' '.join(command))
    try:
        return con.run(' '.join(command), hide=hide).stdout.strip()
    except invoke.exceptions.UnexpectedExit as e:
        logging.error('cannot find IPv6 address, install ipv6calc: %s', e)


def test_ipv6_slaac():
    con = invoke.Context()
    mac = '00:66:37:f1:bb:6b'
    network = '2a01:4f8:fff0:4f::'
    expected = '2a01:4f8:fff0:4f:266:37ff:fef1:bb6b'
    assert expected == ipv6_slaac(con, network, mac)

i'm getting the following exception:

anarcat@angela:tsa-misc(master)$ pytest-3 -W ignore fabric_tpa/host.py
============================= test session starts ==============================
platform linux -- Python 3.7.3, pytest-3.10.1, py-1.7.0, pluggy-0.8.0
rootdir: /home/anarcat/src/tor/tsa-misc, inifile:
plugins: remotedata-0.3.1, openfiles-0.3.2, doctestplus-0.2.0, cov-2.6.0, arraydiff-0.3, betamax-0.8.1
collected 1 item                                                               

fabric_tpa/host.py F                                                     [100%]

=================================== FAILURES ===================================
_______________________________ test_ipv6_slaac ________________________________

    def test_ipv6_slaac():
        con = invoke.Context()
        mac = '00:66:37:f1:bb:6b'
        network = '2a01:4f8:fff0:4f::'
>       assert ipv6_slaac(con, network, mac) == '2a01:4f8:fff0:4f:266:37ff:fef1:bb6b'

fabric_tpa/host.py:153: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/lib/python3/dist-packages/invoke/tasks.py:127: in __call__
    result = self.body(*args, **kwargs)
fabric_tpa/host.py:144: in ipv6_slaac
    return con.run(' '.join(command), hide=hide).stdout.strip()
/usr/lib/python3/dist-packages/invoke/context.py:94: in run
    return self._run(runner, command, **kwargs)
/usr/lib/python3/dist-packages/invoke/context.py:101: in _run
    return runner.run(command, **kwargs)
/usr/lib/python3/dist-packages/invoke/runners.py:291: in run
    return self._run_body(command, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <invoke.runners.Local object at 0x7f5e2152d2b0>
command = 'ipv6calc --action prefixmac2ipv6 --in prefix+mac --out ipv6 2a01:4f8:fff0:4f:: 00:66:37:f1:bb:6b'
kwargs = {'buffer_': [], 'hide': True, 'output': <_pytest.capture.EncodedFile object at 0x7f5e21ac9518>}
opts = {'dry': False, 'echo': False, 'echo_stdin': None, 'encoding': None, ...}
out_stream = <_pytest.capture.EncodedFile object at 0x7f5e21ac9470>
err_stream = <_pytest.capture.EncodedFile object at 0x7f5e21ac9518>
in_stream = <_pytest.capture.DontReadFromInput object at 0x7f5e21ac9400>
shell = '/bin/bash'
env = {'BLOCKSIZE': 'K', 'CDPATH': '.:~:~/src:~/dist:~/wikis:~/go/src:~/src/tor', 'COLORFGBG': '12;default;8', 'COLORTERM': 'rxvt-xpm', ...}
stdout = ['2a01:4f8:fff0:4f:266:37ff:fef1:bb6b\n']

    def _run_body(self, command, **kwargs):
        # Normalize kwargs w/ config
        opts, out_stream, err_stream, in_stream = self._run_opts(kwargs)
        shell = opts["shell"]
        # Environment setup
        env = self.generate_env(opts["env"], opts["replace_env"])
        # Echo running command
        if opts["echo"]:
            print("\033[1;37m{}\033[0m".format(command))
        # If dry-run, stop here.
        if opts["dry"]:
            return self.generate_result(
                command=command,
                stdout="",
                stderr="",
                exited=0,
                pty=self.using_pty,
            )
        # Start executing the actual command (runs in background)
        self.start(command, shell, env)
        self.start_timer(opts["timeout"])
        # Arrive at final encoding if neither config nor kwargs had one
        self.encoding = opts["encoding"] or self.default_encoding()
        # Set up IO thread parameters (format - body_func: {kwargs})
        stdout, stderr = [], []
        thread_args = {
            self.handle_stdout: {
                "buffer_": stdout,
                "hide": "stdout" in opts["hide"],
                "output": out_stream,
            }
        }
        # After opt processing above, in_stream will be a real stream obj or
        # False, so we can truth-test it. We don't even create a stdin-handling
        # thread if it's False, meaning user indicated stdin is nonexistent or
        # problematic.
        if in_stream:
            thread_args[self.handle_stdin] = {
                "input_": in_stream,
                "output": out_stream,
                "echo": opts["echo_stdin"],
            }
        if not self.using_pty:
            thread_args[self.handle_stderr] = {
                "buffer_": stderr,
                "hide": "stderr" in opts["hide"],
                "output": err_stream,
            }
        # Kick off IO threads
        self.threads = {}
        exceptions = []
        for target, kwargs in six.iteritems(thread_args):
            t = ExceptionHandlingThread(target=target, kwargs=kwargs)
            self.threads[target] = t
            t.start()
        # Wait for completion, then tie things off & obtain result
        # And make sure we perform that tying off even if things asplode.
        exception = None
        while True:
            try:
                self.wait()
                break  # done waiting!
            # NOTE: we handle all this now instead of at
            # actual-exception-handling time because otherwise the stdout/err
            # reader threads may block until the subprocess exits.
            # TODO: honor other signals sent to our own process and transmit
            # them to the subprocess before handling 'normally'.
            except KeyboardInterrupt as e:
                self.send_interrupt(e)
                # NOTE: no break; we want to return to self.wait() since we
                # can't know if subprocess is actually terminating due to this
                # or not (think REPLs-within-shells, editors, other interactive
                # use cases)
            except BaseException as e:  # Want to handle SystemExit etc still
                # Store exception for post-shutdown reraise
                exception = e
                # Break out of return-to-wait() loop - we want to shut down
                break
        # Inform stdin-mirroring worker to stop its eternal looping
        self.program_finished.set()
        # Join threads, setting a timeout if necessary
        for target, thread in six.iteritems(self.threads):
            thread.join(self._thread_join_timeout(target))
            e = thread.exception()
            if e is not None:
                exceptions.append(e)
        # If we got a main-thread exception while wait()ing, raise it now that
        # we've closed our worker threads.
        if exception is not None:
            raise exception
        # Strip out WatcherError from any thread exceptions; they are bundled
        # into Failure handling at the end.
        watcher_errors = []
        thread_exceptions = []
        for exception in exceptions:
            real = exception.value
            if isinstance(real, WatcherError):
                watcher_errors.append(real)
            else:
                thread_exceptions.append(exception)
        # If any exceptions appeared inside the threads, raise them now as an
        # aggregate exception object.
        if thread_exceptions:
>           raise ThreadException(thread_exceptions)
E           invoke.exceptions.ThreadException: 
E           Saw 1 exceptions within threads (OSError):
E           
E           
E           Thread args: {'kwargs': {'echo': None,
E                       'input_': <_pytest.capture.DontReadFromInput object at 0x7f5e21ac9400>,
E                       'output': <_pytest.capture.EncodedFile object at 0x7f5e21ac9470>},
E            'target': <bound method Runner.handle_stdin of <invoke.runners.Local object at 0x7f5e2152d2b0>>}
E           
E           Traceback (most recent call last):
E           
E             File "/usr/lib/python3/dist-packages/invoke/util.py", line 229, in run
E               super(ExceptionHandlingThread, self).run()
E           
E             File "/usr/lib/python3.7/threading.py", line 865, in run
E               self._target(*self._args, **self._kwargs)
E           
E             File "/usr/lib/python3/dist-packages/invoke/runners.py", line 687, in handle_stdin
E               data = self.read_our_stdin(input_)
E           
E             File "/usr/lib/python3/dist-packages/invoke/runners.py", line 646, in read_our_stdin
E               bytes_ = input_.read(bytes_to_read(input_))
E           
E             File "/usr/lib/python3/dist-packages/_pytest/capture.py", line 653, in read
E               raise IOError("reading from stdin while output is captured")
E           
E           OSError: reading from stdin while output is captured

/usr/lib/python3/dist-packages/invoke/runners.py:399: ThreadException
=========================== 1 failed in 0.13 seconds ===========================

the only workaround i found is to pass -s (--capture=no) to the command, but that's really too bad because i would definitely like to capture and test invoke output...

now i understand if my function would prompt the user for some stuff - then i could mock the input and go from there, but this doesn't require anything from stdin so it's not clear to me why invoke is reading from stdin here.

(invoke 1.3 on Debian buster)

[thedrow]

I think you need to disable stdout capturing by invoke.

[shlomiLan]

I managed to solve it with:

context.run(command, pty=True, in_stream=False)

[thedrow]

Should we close the issue?

[anarcat]

I think you need to disable stdout capturing by invoke.

Yes, that is what I suggested in my original report:

the only workaround i found is to pass -s (--capture=no) to the command, but that's really too bad because i would definitely like to capture and test invoke output...

That defeats the point of running tests: I want to capture that output!

I managed to solve it with:

context.run(command, pty=True, in_stream=False)

Should we close the issue?

No. We shouldn't have to modify invoke code to run tests on it. pty=True, in_stream=False have side-effects that are in some cases undesirable, otherwise why would they be options in the first place?

invoke should be somehow patched to allow unit testing of tasks. At the very least, it should be documented, if impossible, or at least capture=no should be documented, and there should be documentation on how to test tasks with output.