tests.test_control

Complexity

Total Complexity

14

Size/Duplication

Total Lines	115
Duplicated Lines	0 %

Importance

Changes

0

11 Functions

Rating	Name	Duplication	Size	Complexity
A	test_fallback()	0	4	1
A	test_sequence()	0	23	1
A	test_repeat_until_falsy_condition()	0	17	3
A	test_selector()	0	4	1
A	test_decision()	0	8	1
A	success_func()	0	2	1
A	exception_func()	0	2	1
A	failure_func()	0	2	1
A	a_func()	0	2	1
A	b_func()	0	2	1
A	test_repeat_until_return_last_result()	0	14	2

from contextvars import ContextVar

from async_btree import (
    FAILURE,
    SUCCESS,
    ExceptionDecorator,
    decision,
    fallback,
    repeat_until,
    selector,
    sequence,
)


async def a_func():
    return 'a'


async def b_func():
    return 'b'


async def failure_func():
    return FAILURE


async def success_func():
    return SUCCESS


async def exception_func():
    raise RuntimeError("ops")


def test_sequence(kernel):
    assert not kernel.run(
        sequence(children=[a_func, failure_func, success_func])
    ), "default behaviour fail of one failed"

    assert kernel.run(
        sequence(children=[a_func, failure_func, success_func], succes_threshold=2)
    )
    assert kernel.run(
        sequence(children=[a_func, success_func, failure_func], succes_threshold=2)
    )
    assert kernel.run(
        sequence(children=[failure_func, a_func, success_func], succes_threshold=2)
    ), 'must continue after first failure'

    assert kernel.run(
        sequence(children=[exception_func, failure_func, a_func], succes_threshold=1)
    )
    assert not kernel.run(
        sequence(children=[exception_func, failure_func], succes_threshold=1)
    )

    assert not kernel.run(sequence(children=[]))


def test_fallback(kernel):
    assert kernel.run(fallback(children=[exception_func, failure_func, a_func]))
    assert not kernel.run(fallback(children=[exception_func, failure_func]))
    assert not kernel.run(fallback(children=[]))


def test_selector(kernel):
    assert kernel.run(selector(children=[exception_func, failure_func, a_func]))
    assert not kernel.run(selector(children=[exception_func, failure_func]))
    assert selector(children=[]).__node_metadata.name == 'selector'


def test_decision(kernel):
    assert kernel.run(decision(condition=success_func, success_tree=a_func)) == 'a'
    assert not kernel.run(decision(condition=failure_func, success_tree=a_func))

    result = kernel.run(
        decision(condition=failure_func, success_tree=a_func, failure_tree=b_func)
    )
    assert result == 'b', 'failure tree must be called'


def test_repeat_until_falsy_condition(kernel):

    counter = ContextVar('counter', default=5)

    async def tick():
        value = counter.get()
        counter.set(value - 1)
        if value <= 0:
            return FAILURE
        if value == 3:
            raise RuntimeError('3')
        return SUCCESS

    assert (
        kernel.run(repeat_until(condition=tick, child=a_func)) == 'a'
    ), 'return last sucess result'
    assert counter.get() == 2


def test_repeat_until_return_last_result(kernel):

    counter = ContextVar('tick_test_repeat_until_return_last_result', default=5)

    async def tick():
        value = counter.get()
        counter.set(value - 1)
        if value <= 0:
            return FAILURE
        return SUCCESS

    result = kernel.run(repeat_until(condition=tick, child=exception_func))
    assert counter.get() == -1
    assert isinstance(result, ExceptionDecorator)