Provide a way to write unit tests

johnlinp commented 2 years ago

Problem

I would like to write unit tests for my Kopf operator. However, I don't see Kopf providing any way to do it. I do see that there's a way to write integration tests: https://kopf.readthedocs.io/en/stable/testing/, which requires a running Kubernetes instance. I would like to write tests that are more light-weight, so that I can know which test cases are failing more quickly.

Proposal

I would like to see an official way to write unit tests, without the requirement of a running Kubernetes instance.

Code

No response

Additional information

No response

nolar commented 2 years ago

Hello. Can you propose the desired syntax for tests, please? I.e., how you would like to express the test scenarios you currently have at hand?

johnlinp commented 2 years ago

Hi @nolar,

I have 2 handlers, one for creation of the CRD, and another for deletion.

The handler for deletion is rather simple:

# controller.py
@kopf.on.delete('johnlinp.com', 'v1', 'FooResource')
def on_delete(name, namespace, **_):
    foo.delete_pod(namespace, name)

So I can simply write a unit test like this:

# controller_test.py
def test_on_delete(mocker):
    mocker.patch('foo.delete_pod')

    controller.on_delete(name='test_name', namespace='test_namespace')

    foo.delete_pod.assert_called_once_with('test_namespace', 'test_name')

However, the handler for creation is more complex:

# controller.py
@kopf.on.create('johnlinp.com', 'v1', 'FooResource')
def on_create(spec, name, namespace, patch, **_):
    @kopf.subhandler(id='init_crd_status')
    def init_crd_status(**_):
        update_status(patch, 'PENDING')

    @kopf.subhandler(id='create_underlying_pod')
    def create_underlying_pod(**_):
        try:
            foo.create_pod(namespace, name, spec)
        except Exception as e:
            update_status(patch, 'ERROR')
            raise kopf.PermanentError('error occurred when create underlying pod') from e

        update_status(patch, 'READY')

def update_status(patch, phase):
    patch.status['phase'] = phase

The unit test will look like this:

# controller_test.py
@pytest.fixture
def on_create_setup(mocker):
    mocker.patch('foo.create_pod')
    mocker.patch('controller.update_status')

    def mock_kopf_subhandler(id):
        return lambda func: func()
    mocker.patch('kopf.subhandler', wraps=mock_kopf_subhandler)

@pytest.fixture
def mock_patch(mocker):
    return mocker.Mock()

def test_on_create_success(mocker, on_create_setup, mock_patch):
    controller.on_create(spec={'ram': '2Gi'}, name='test_name', namespace='test_namespace', patch=mock_patch)

    foo.create_pod.assert_called_once_with('test_namespace', 'test_name', {'ram': '2Gi'})
    controller.update_status.assert_has_calls([
        mocker.call(mock_patch, 'PENDING'),
        mocker.call(mock_patch, 'READY')
    ])

def test_on_create_fail(mocker, on_create_setup, mock_patch):
    mocker.patch('foo.create_pod', side_effect=Exception)

    with pytest.raises(kopf.PermanentError):
        controller.on_create(spec={'ram': '2Gi'}, name='test_name', namespace='test_namespace', patch=mock_patch)

    controller.update_status.assert_has_calls([
        mocker.call(mock_patch, 'PENDING'),
        mocker.call(mock_patch, 'ERROR')
    ])

There are 2 things that I think can be improved:

I have to patch the decorator kopf.subhandler by myself in the unit test. I'm not sure if my patch covers all use cases.
I have to wrap the patch object with a function (i.e. controller.update_status()), so that I can check the calls against it, as in controller.update_status.assert_has_calls([...]).

It will be better if Kopf can provide some test fixtures that can handle these cases. That way, I can simplify my unit tests into something like this:

# controller_test.py
@pytest.fixture
def on_create_setup(mocker):
    mocker.patch('foo.create_pod')

def test_on_create_success(mocker, on_create_setup, kopf_setup, kopf_mock_patch):
    controller.on_create(spec={'ram': '2Gi'}, name='test_name', namespace='test_namespace', patch=kopf_mock_patch)

    foo.create_pod.assert_called_once_with('test_namespace', 'test_name', {'ram': '2Gi'})
    kopf_mock_patch.status.__setitem__.assert_has_calls([
        mocker.call('phase', 'PENDING'),
        mocker.call('phase', 'READY')
    ])

def test_on_create_fail(mocker, on_create_setup, kopf_setup, kopf_mock_patch):
    mocker.patch('foo.create_pod', side_effect=Exception)

    with pytest.raises(kopf.PermanentError):
        controller.on_create(spec={'ram': '2Gi'}, name='test_name', namespace='test_namespace', patch=kopf_mock_patch)

    kopf_mock_patch.status.__setitem__.assert_has_calls([
        mocker.call('phase', 'PENDING'),
        mocker.call('phase', 'ERROR')
    ])

where the fixtures kopf_setup and kopf_mock_patch are provided by Kopf.

Please let me know if it's unclear. Thank you.

james-mchugh commented 5 months ago

I'd love to see some improvements here as well. I'm trying to write unit tests for my operators, and I'd like to be able to exercise the filtering logic for the handlers to ensure it is operating as expected. I've been reading through the kopf code a bit to see if there was an easily-available function I could use to run a handler end-to-end (including running filters), but I haven't had much luck yet.

nolar / kopf