test_curiosity.test_run_curio() - Code Metrics - Inspection of "Merge pull request #3 from geronimo-iia/develop" - geronimo-iia/async-btree - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d5e257...2cfe0f )

by Guibert

created 2019-07-03 02:52 UTC

test_curiosity.test_run_curio() A

↳ Parent: test_curiosity

Complexity

Conditions

Size

Total Lines	17
Code Lines	14

Duplication

Lines	17
Ratio	100 %

Importance

Changes

Metric	Value
eloc	14
dl	17
loc	17
rs	9.7
c	0
b	0
f	0
cc	3
nop	0

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():

    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


1		from contextvars import ContextVar
2
3		from curio import Kernel, run, sleep
4
5		from libellule.control_flow.common import FAILURE
6		from libellule.control_flow.curiosity import parallele, run as run_curio
7
8
9		async def a_func():
10		await sleep(1)
11		return "a"
12
13
14		async def b_func():
15		await sleep(3)
16		return "b"
17
18
19		async def failure_func():
20		await sleep(2)
21		return FAILURE
22
23
24		def test_parallele():
25		assert run(parallele(children=[a_func]))
26		assert run(parallele(children=[a_func, b_func]))
27		assert not run(parallele(children=[a_func, b_func, failure_func]))
28		assert run(parallele(children=[a_func, b_func, failure_func], succes_threshold=2))
29
30
31	View Code Duplication	def test_run_curio():
		0 ignored issues – show Duplication introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
32		counter = ContextVar("counter", default=5)
33
34		async def reset_counter():
35		if counter.get() == 5:
36		counter.set(0)
37		return 0
38		return -1
39
40		with Kernel() as kernel:
41		assert run_curio(kernel, reset_counter) == 0
42		assert run_curio(kernel, reset_counter) == 0
43		assert counter.get() == 5
44
45		assert run(reset_counter) == 0
46		assert run(reset_counter) == -1
47		assert counter.get() == 0
48

geronimo-iia / async-btree

Push — master ( d5e257...2cfe0f )

test_curiosity.test_run_curio() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like