test_control

Complexity

Total Complexity

12

Size/Duplication

Total Lines	89
Duplicated Lines	0 %

Importance

Changes

0

10 Functions

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

from contextvars import ContextVar

from curio import run

from libellule.control_flow.common import *
from libellule.control_flow.control import *


async def a_func():
    return "a"


async def b_func():
    return "b"


async def failure_func():
    return FAILURE

The variable FAILURE does not seem to be defined.

async def success_func():
    return SUCCESS

The variable SUCCESS does not seem to be defined.

async def exception_func():
    raise RuntimeError()


def test_sequence():
    assert not run(
        sequence(children=[a_func, failure_func, success_func])

The variable sequence does not seem to be defined.

    ), "default behaviour fail of one failed"

    assert run(
        sequence(children=[a_func, failure_func, success_func], succes_threshold=2)
    )
    assert run(
        sequence(children=[a_func, success_func, failure_func], succes_threshold=2)
    )
    assert run(
        sequence(children=[failure_func, a_func, success_func], succes_threshold=2)
    )

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

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


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

The variable fallback does not seem to be defined.

    assert not run(fallback(children=[exception_func, failure_func]))
    assert not run(fallback(children=[]))


def test_selector():
    assert run(selector(children=[exception_func, failure_func, a_func]))

The variable selector does not seem to be defined.

    assert not run(selector(children=[exception_func, failure_func]))
    assert selector(children=[]).__node_metadata.name == "selector"


def test_decision():
    assert run(decision(condition=success_func, success_tree=a_func)) == "a"

The variable decision does not seem to be defined.

    assert not run(decision(condition=failure_func, success_tree=a_func))
    assert (
        run(decision(condition=failure_func, success_tree=a_func, failure_tree=b_func))
        == "b"
    )


def test_repeat_until():

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

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

The variable FAILURE does not seem to be defined.

        if value == 3:
            raise RuntimeError("3")
        return SUCCESS

The variable SUCCESS does not seem to be defined.

    assert run(repeat_until(condition=tick, child=a_func)) == "a"

The variable repeat_until does not seem to be defined.