kopf.testing module

Helper tools to test the Kopf-based operators.

This module is a part of the framework’s public interface.

class kopf.testing.KopfRunner(*args, reraise=True, timeout=None, registry=None, settings=None, **kwargs)[source]

Bases: _AbstractKopfRunner

A context manager to run a Kopf-based operator in parallel with the tests.

Usage:

from kopf.testing import KopfRunner

with KopfRunner(['run', '-A', '--verbose', 'examples/01-minimal/example.py']) as runner:
    # do something while the operator is running.
    time.sleep(3)

assert runner.exit_code == 0
assert runner.exception is None
assert 'And here we are!' in runner.stdout

All the args & kwargs are passed directly to Click’s invocation method. See: click.testing.CliRunner. All properties proxy directly to Click’s click.testing.Result object when it is available (i.e. after the context manager exits).

CLI commands have to be invoked in parallel threads, never in processes:

First, with multiprocessing, they are unable to pickle and pass exceptions (specifically, their traceback objects) from a child thread (Kopf’s CLI) to the parent thread (pytest).

Second, mocking works within one process (all threads), but not across processes — the mock’s calls (counts, arrgs) are lost.

Parameters:
property future: Future[source]
property output: str[source]
property stdout: str[source]
property stdout_bytes: bytes[source]
property stderr: str[source]
property stderr_bytes: bytes[source]
property exit_code: int[source]
property exception: BaseException[source]
property exc_info: Tuple[Type[BaseException], BaseException, TracebackType][source]