test_decorator

Complexity

Total Complexity

19

Size/Duplication

Total Lines	120
Duplicated Lines	0 %

Importance

Changes

0

16 Functions

Rating	Name	Duplication	Size	Complexity
A	test_inverter()	0	7	2
A	a_func()	0	2	1
A	test_decorate()	0	7	1
A	test_retry_until_success()	0	3	1
A	success_func()	0	2	1
A	exception_func()	0	2	1
A	test_retry_until_failed()	0	3	1
A	empty_func()	0	2	1
A	failure_func()	0	2	1
A	test_always_failure()	0	6	1
A	tick()	0	8	3
A	test_retry()	0	12	1
A	test_root_name()	0	4	1
A	test_is_success()	0	6	1
A	test_always_success()	0	5	1
A	test_is_failure()	0	6	1

from contextvars import ContextVar

import pytest
from curio import run

from libellule.control_flow.common import FAILURE, SUCCESS, ControlFlowException
from libellule.control_flow.decorator import *


async def a_func():
    return "a"


async def failure_func():
    return FAILURE


async def success_func():
    return SUCCESS


async def exception_func():
    raise RuntimeError()


async def empty_func():
    return []


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


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


def test_root_name():
    rooted = alias(a_func, name="a_func")

The variable alias does not seem to be defined.

    assert rooted.__node_metadata.name == "a_func"
    assert run(rooted) == "a"


def test_decorate():
    async def b_decorator(child_value, other=""):
        return f"b{child_value}{other}"

    assert run(decorate(a_func, b_decorator)) == "ba"

The variable decorate does not seem to be defined.

    assert run(decorate(a_func, b_decorator, other="c")) == "bac"


def test_always_success():
    assert run(always_success(success_func)) == SUCCESS

The variable always_success does not seem to be defined.

    assert run(always_success(failure_func)) == SUCCESS
    assert run(always_success(exception_func)) == SUCCESS
    assert run(always_success(a_func)) == "a"


def test_always_failure():
    assert run(always_failure(success_func)) == FAILURE

The variable always_failure does not seem to be defined.

    assert run(always_failure(failure_func)) == FAILURE
    assert not run(always_failure(exception_func))
    assert isinstance(run(always_failure(exception_func)), ControlFlowException)
    assert run(always_failure(empty_func)) == []


def test_is_success():
    assert run(is_success(success_func))

The variable is_success does not seem to be defined.

    assert not run(is_success(failure_func))
    assert not run(is_success(exception_func))
    assert run(is_success(a_func))
    assert not run(is_success(empty_func))


def test_is_failure():
    assert not run(is_failure(success_func))

The variable is_failure does not seem to be defined.

    assert run(is_failure(failure_func))
    assert run(is_failure(exception_func))
    assert not run(is_failure(a_func))
    assert run(is_failure(empty_func))


def test_inverter():
    assert not run(inverter(success_func))

The variable inverter does not seem to be defined.

    assert run(inverter(failure_func))
    with pytest.raises(RuntimeError):
        run(inverter(exception_func))
    assert not run(inverter(a_func))
    assert run(inverter(empty_func))


def test_retry():

    result = run(retry(tick))

The variable retry does not seem to be defined.

    assert not result
    assert isinstance(result, ControlFlowException)
    assert run(retry(tick))

    counter.set(10)
    assert run(retry(tick, max_retry=11))

    counter.set(100)
    assert run(retry(tick, max_retry=-1))


def test_retry_until_success():
    counter.set(100)
    assert run(retry_until_success(tick))

The variable retry_until_success does not seem to be defined.

def test_retry_until_failed():
    counter.set(100)
    assert run(retry_until_failed(tick))

The variable retry_until_failed does not seem to be defined.