test_curiosity

Complexity

Total Complexity

7

Size/Duplication

Total Lines	48
Duplicated Lines	35.42 %

Importance

Changes

0

5 Functions

Rating	Name	Duplication	Size	Complexity
A	test_run_curio()	17	17	3
A	failure_func()	0	3	1
A	b_func()	0	3	1
A	test_parallele()	0	5	1
A	a_func()	0	3	1

from contextvars import ContextVar

from curio import Kernel, run, sleep

from libellule.control_flow.common import FAILURE
from libellule.control_flow.curiosity import parallele, run as run_curio


async def a_func():
    await sleep(1)
    return "a"


async def b_func():
    await sleep(3)
    return "b"


async def failure_func():
    await sleep(2)
    return FAILURE


def test_parallele():
    assert run(parallele(children=[a_func]))
    assert run(parallele(children=[a_func, b_func]))
    assert not run(parallele(children=[a_func, b_func, failure_func]))
    assert run(parallele(children=[a_func, b_func, failure_func], succes_threshold=2))


def test_run_curio():

This code seems to be duplicated in your project.

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

    async def reset_counter():
        if counter.get() == 5:
            counter.set(0)
            return 0
        return -1

    with Kernel() as kernel:
        assert run_curio(kernel, reset_counter) == 0
        assert run_curio(kernel, reset_counter) == 0
        assert counter.get() == 5

    assert run(reset_counter) == 0
    assert run(reset_counter) == -1
    assert counter.get() == 0