test_control.test_repeat_until() - Code Metrics - Inspection of "prepare 0.1.0" - geronimo-iia/async-btree - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3)

by Guibert

created 2019-07-03 02:35 UTC

test_control.test_repeat_until() A

↳ Parent: test_control

Complexity

Conditions

Size

Total Lines	14
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	14
rs	9.85
c	0
b	0
f	0
cc	3
nop	0

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



async def success_func():
    return SUCCESS



async def exception_func():
    raise RuntimeError()


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

    ), "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]))

    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]))

    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"

    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

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


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



1			from contextvars import ContextVar
2
3			from curio import run
4
5			from libellule.control_flow.common import *
6			from libellule.control_flow.control import *
7
8
9			async def a_func():
10			return "a"
11
12
13			async def b_func():
14			return "b"
15
16
17			async def failure_func():
18			return FAILURE
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `FAILURE` does not seem to be defined. Loading history...
19
20
21			async def success_func():
22			return SUCCESS
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `SUCCESS` does not seem to be defined. Loading history...
23
24
25			async def exception_func():
26			raise RuntimeError()
27
28
29			def test_sequence():
30			assert not run(
31			sequence(children=[a_func, failure_func, success_func])
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `sequence` does not seem to be defined. Loading history...
32			), "default behaviour fail of one failed"
33
34			assert run(
35			sequence(children=[a_func, failure_func, success_func], succes_threshold=2)
36			)
37			assert run(
38			sequence(children=[a_func, success_func, failure_func], succes_threshold=2)
39			)
40			assert run(
41			sequence(children=[failure_func, a_func, success_func], succes_threshold=2)
42			)
43
44			assert run(
45			sequence(children=[exception_func, failure_func, a_func], succes_threshold=1)
46			)
47			assert not run(
48			sequence(children=[exception_func, failure_func], succes_threshold=1)
49			)
50
51			assert not run(sequence(children=[]))
52
53
54			def test_fallback():
55			assert run(fallback(children=[exception_func, failure_func, a_func]))
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `fallback` does not seem to be defined. Loading history...
56			assert not run(fallback(children=[exception_func, failure_func]))
57			assert not run(fallback(children=[]))
58
59
60			def test_selector():
61			assert run(selector(children=[exception_func, failure_func, a_func]))
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `selector` does not seem to be defined. Loading history...
62			assert not run(selector(children=[exception_func, failure_func]))
63			assert selector(children=[]).__node_metadata.name == "selector"
64
65
66			def test_decision():
67			assert run(decision(condition=success_func, success_tree=a_func)) == "a"
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `decision` does not seem to be defined. Loading history...
68			assert not run(decision(condition=failure_func, success_tree=a_func))
69			assert (
70			run(decision(condition=failure_func, success_tree=a_func, failure_tree=b_func))
71			== "b"
72			)
73
74
75			def test_repeat_until():
76
77			counter = ContextVar("counter", default=5)
78
79			async def tick():
80			value = counter.get()
81			counter.set(value - 1)
82			if value <= 0:
83			return FAILURE
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `FAILURE` does not seem to be defined. Loading history...
84			if value == 3:
85			raise RuntimeError("3")
86			return SUCCESS
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `SUCCESS` does not seem to be defined. Loading history...
87
88			assert run(repeat_until(condition=tick, child=a_func)) == "a"
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-03 02:29 UTC by Report Bug Copy Issue Report The variable `repeat_until` does not seem to be defined. Loading history...
89

geronimo-iia / async-btree

Pull Request — master (#3)

test_control.test_repeat_until() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like