Operator testing

Kopf provides some tools to test the Kopf-based operators via kopf.testing module (requires explicit importing).

Background runner

kopf.testing.KopfRunner runs an arbitrary operator in the background, while the original testing thread does the object manipulation and assertions:

When the with block exits, the operator stops, and its exceptions, exit code and output are available to the test (for additional assertions).

test_example_operator.py

import time
import subprocess
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.

        subprocess.run("kubectl apply -f examples/obj.yaml", shell=True, check=True)
        time.sleep(1)  # give it some time to react and to sleep and to retry

        subprocess.run("kubectl delete -f examples/obj.yaml", shell=True, check=True)
        time.sleep(1)  # give it some time to react

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

Note

The operator runs against the cluster which is currently authenticated — same as if would be executed with kopf run.