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

def test_operator():
    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.output

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 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, args) are lost.

Parameters:
property future: Future[Result][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]