Skip to content
This repository has been archived by the owner on Sep 14, 2020. It is now read-only.

Commit

Permalink
Provide negation helpers for callbacks in when=, labels/annotations
Browse files Browse the repository at this point in the history
  • Loading branch information
nolar committed Apr 7, 2020
1 parent b1a33af commit c33c349
Show file tree
Hide file tree
Showing 4 changed files with 66 additions and 43 deletions.
7 changes: 7 additions & 0 deletions docs/filters.rst
Original file line number Diff line number Diff line change
Expand Up @@ -105,3 +105,10 @@ Kopf provides few helpers to combine multiple callbacks into one
labels={'somelabel': kopf.any_([value_fn1, value_fn2])})
def create_fn2(**_):
pass

The following wrappers are available:

* `kopf.not_(fn)` -- the function must return ``False`` to pass the filters.
* `kopf.any_([...])` -- at least one of the functions must return ``True``.
* `kopf.all_([...])` -- all of the functions must return ``True``.
* `kopf.none_([...])` -- all of the functions must return ``False``.
4 changes: 4 additions & 0 deletions kopf/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,10 @@
build_owner_reference,
)
from kopf.structs.callbacks import (
not_,
all_,
any_,
none_,
)
from kopf.structs.configuration import (
OperatorSettings,
Expand Down Expand Up @@ -166,8 +168,10 @@
'event', 'info', 'warn', 'exception',
'spawn_tasks', 'run_tasks', 'operator', 'run', 'create_tasks',
'adopt', 'label',
'not_',
'all_',
'any_',
'none_',
'get_default_lifecycle', 'set_default_lifecycle',
'build_object_reference', 'build_owner_reference',
'append_owner_reference', 'remove_owner_reference',
Expand Down
46 changes: 11 additions & 35 deletions kopf/structs/callbacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,49 +176,25 @@ def __call__( # lgtm[py/similar-function]
_FnT = TypeVar('_FnT', WhenFilterFn, MetaFilterFn)


def all_(fns: Collection[_FnT]) -> _FnT:
"""
Combine few callbacks into one::
import kopf
def whole_fn1(name, **_): return name.startswith('kopf-')
def whole_fn2(spec, **_): return spec.get('field') == 'value'
def value_fn1(value, **_): return value.startswith('some')
def value_fn2(value, **_): return value.endswith('label')
def not_(fn: _FnT) -> _FnT:
def not_fn(*args: Any, **kwargs: Any) -> bool:
return not fn(*args, **kwargs)
return not_fn

@kopf.on.create('zalando.org', 'v1', 'kopfexamples',
when=kopf.all_([whole_fn1, whole_fn2]),
labels={'somelabel': kopf.all_([value_fn1, value_fn2])})
def create_fn(**_):
pass

The semantics is the same as for Python's built-in :func:`all`.
"""
def all_(fns: Collection[_FnT]) -> _FnT:
def all_fn(*args: Any, **kwargs: Any) -> bool:
return all(fn(*args, **kwargs) for fn in fns)
return all_fn


def any_(fns: Collection[_FnT]) -> _FnT:
"""
Combine few callbacks into one::
import kopf
def whole_fn1(name, **_): return name.startswith('kopf-')
def whole_fn2(spec, **_): return spec.get('field') == 'value'
def value_fn1(value, **_): return value.startswith('some')
def value_fn2(value, **_): return value.endswith('label')
@kopf.on.create('zalando.org', 'v1', 'kopfexamples',
when=kopf.any_([whole_fn1, whole_fn2]),
labels={'somelabel': kopf.any_([value_fn1, value_fn2])})
def create_fn(**_):
pass
The semantics is the same as for Python's built-in :func:`any`.
"""
def any_fn(*args: Any, **kwargs: Any) -> bool:
return any(fn(*args, **kwargs) for fn in fns)
return any_fn


def none_(fns: Collection[_FnT]) -> _FnT:
def none_fn(*args: Any, **kwargs: Any) -> bool:
return not any(fn(*args, **kwargs) for fn in fns)
return none_fn
52 changes: 44 additions & 8 deletions tests/test_filtering_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,49 +17,85 @@ def _always2(*_, **__):
return True


def test_all_when_all_are_true():
def test_notfn_when_true():
combined = kopf.not_(_always1)
result = combined()
assert result is False


def test_notfn_when_false():
combined = kopf.not_(_never1)
result = combined()
assert result is True


def test_allfn_when_all_are_true():
combined = kopf.all_([_always1, _always2])
result = combined()
assert result is True


def test_all_when_one_is_false():
def test_allfn_when_one_is_false():
combined = kopf.all_([_always1, _never1])
result = combined()
assert result is False


def test_all_when_all_are_false():
def test_allfn_when_all_are_false():
combined = kopf.all_([_never1, _never2])
result = combined()
assert result is False


def test_all_when_no_functions():
def test_allfn_when_no_functions():
combined = kopf.all_([])
result = combined()
assert result is True


def test_any_when_all_are_true():
def test_anyfn_when_all_are_true():
combined = kopf.any_([_always1, _always2])
result = combined()
assert result is True


def test_any_when_one_is_false():
def test_anyfn_when_one_is_false():
combined = kopf.any_([_always1, _never1])
result = combined()
assert result is True


def test_any_when_all_are_false():
def test_anyfn_when_all_are_false():
combined = kopf.any_([_never1, _never2])
result = combined()
assert result is False


def test_any_when_no_functions():
def test_anyfn_when_no_functions():
combined = kopf.any_([])
result = combined()
assert result is False


def test_nonefn_when_all_are_true():
combined = kopf.none_([_always1, _always2])
result = combined()
assert result is False


def test_nonefn_when_one_is_false():
combined = kopf.none_([_always1, _never1])
result = combined()
assert result is False


def test_nonefn_when_all_are_false():
combined = kopf.none_([_never1, _never2])
result = combined()
assert result is True


def test_nonefn_when_no_functions():
combined = kopf.none_([])
result = combined()
assert result is True

0 comments on commit c33c349

Please sign in to comment.