File: flow_test.py

package info (click to toggle)
errbot 6.1.7%2Bds-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye
  • size: 3,712 kB
  • sloc: python: 13,831; makefile: 164; sh: 97
file content (47 lines) | stat: -rw-r--r-- 1,408 bytes parent folder | download | duplicates (4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import logging

import pytest

from errbot.backends.test import TestPerson
from errbot.flow import Flow, FlowRoot, InvalidState

log = logging.getLogger(__name__)


def test_node():
    root = FlowRoot("test", "This is my flowroot")
    node = root.connect("a", lambda ctx: ctx["toto"] == "titui")

    assert root.predicate_for_node(node)({"toto": "titui"})
    assert not root.predicate_for_node(node)({"toto": "blah"})


def test_flow_predicate():
    root = FlowRoot("test", "This is my flowroot")
    node = root.connect("a", lambda ctx: "toto" in ctx and ctx["toto"] == "titui")
    somebody = TestPerson("me")

    # Non-matching predicate
    flow = Flow(root, somebody, {})
    assert node in flow.next_steps()
    assert node not in flow.next_autosteps()
    with pytest.raises(InvalidState):
        flow.advance(node)

    flow.advance(node, enforce_predicate=False)  # This will bypass the restriction
    assert flow._current_step == node

    # Matching predicate
    flow = Flow(root, somebody, {"toto": "titui"})
    assert node in flow.next_steps()
    assert node in flow.next_autosteps()
    flow.advance(node)
    assert flow._current_step == node


def test_autotrigger():
    root = FlowRoot("test", "This is my flowroot")
    node = root.connect(
        "a", lambda ctx: "toto" in ctx and ctx["toto"] == "titui", auto_trigger=True
    )
    assert node.command in root.auto_triggers