Extended testing capabilities

[kopf-archiver]

An issue by nolar at 2019-11-21 19:30:50+00:00
Original URL: zalando-incubator/kopf#254
 

Currently, koof.testing.KopfRunner provides a way to run an operator in the background while running a test or a piece of code. The only way to assert anything is to check the resources themselves.

That would be much more helpful, if some internals are exposed on what is happening inside.

A/C:

• Handler outcomes exposed for tests (as introduces in [PR] Refactoring 4/3: In-memory handler outcomes & persistent state merges #221). Including all the outcomes if there were few retries.

• Easier assertions on the outcomes of the operator actions, in a form of a little DSL for the Outcome & Outcomes & State classes.

• A runner for predefined registry rather than only a CLI runner (as it is now).

• Pytest fixtures to simplify a background operator running, while populating it inside of the tests/fixtures (not pre-populated).

Maybe something more. The issue is to be finished.

A general goal: Testing the operator should be as easy as developing it, i.e. expressing the ideas in few lines of code, and preferably so that a test fits on one screen/slide (as a measure of simplicity).

For example (but not for definition):

# ./myapp/handlers.py
import kopf

@kopf.on.create('zalando.org', 'v1', 'kopfexamples')
def create_fn(**kwargs):
    return 'result'

import kopf
import pykube

class KopfExample(pykube.objects.NamespacedAPIObject):
    version = "zalando.org/v1"
    endpoint = "kopfexamples"
    kind = "KopfExample"

def test_me(kopf_operator):

    # Add handlers to a LIVE operator on the go.
    kopf_runner.load('myapp/handlers.py')

    # Create an resource.
    api = pykube.HTTPClient(pykube.KubeConfig.from_env())
    kex1 = KopfExample(api, {
        'metadata': {'name': 'kex1'},
        'spec': {'duration': '3m'},
    })
    kex1.create()

    # Block the test until some internal condition is reached.
    kopf_runner.wait_until_handled(kex1.obj, timeout=100)

    # Assert what happened internally.
    assert len(kopf_runner.reactions) == 1  # not overreacting :-) 
    assert kopf_runner.reactions[-1].cause.reason == kopf.Reason.CREATE
    assert kopf_runner.reactions[-1].cause.body['metadata']['name'] == 'kex1'
    assert kopf_runner.reactions[-1].handlers['create_fn'].duration <= 30  # seconds
    assert kopf_runner.reactions[-1].handlers['create_fn'].retries == 1  # no retries
    assert kopf_runner.reactions[-1].handlers['create_fn'].outcomes[-1].result == 'result'
    assert kopf_runner.reactions[-1].handlers['create_fn'].outcomes[-1].exception is None

    # Assert what happened externally.
    kex1.reload()
    assert kex1.status['create_fn'] == 'result'