• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

zincware / ZnFlow / 4464408602

pending completion
4464408602

push

github

PythonFZ
bump version

2 of 2 new or added lines in 2 files covered. (100.0%)

1134 of 1164 relevant lines covered (97.42%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.61
/tests/test_graph.py
1
import dataclasses
1✔
2

3
import pytest
1✔
4
import zninit
1✔
5

6
import znflow
1✔
7
from znflow.base import empty
1✔
8

9

10
class PlainNode(znflow.Node):
1✔
11
    def __init__(self, value):
1✔
12
        self.value = value
×
13

14
    def run(self):
1✔
15
        self.value += 1
×
16

17

18
@dataclasses.dataclass
1✔
19
class DataclassNode(znflow.Node):
1✔
20
    value: int
1✔
21

22
    def run(self):
1✔
23
        self.value += 1
1✔
24

25

26
class ZnInitNode(zninit.ZnInit, znflow.Node):
1✔
27
    value: int = zninit.Descriptor()
1✔
28

29
    def run(self):
1✔
30
        self.value += 1
×
31

32

33
class ComputeSum(znflow.Node):
1✔
34
    def __init__(self, *args):
1✔
35
        self.args = args
1✔
36

37
    def run(self):
1✔
38
        self.result = sum(x.value for x in self.args)
1✔
39

40

41
@znflow.nodify
1✔
42
def compute_sum(*args):
1✔
43
    return sum(args)
1✔
44

45

46
def test_run_graph():
1✔
47
    with znflow.DiGraph() as graph:
1✔
48
        node1 = DataclassNode(value=42)
1✔
49
        node2 = DataclassNode(value=18)
1✔
50
        node3 = compute_sum(node1.value, node2.value)
1✔
51

52
    assert node1.uuid in graph
1✔
53
    assert node2.uuid in graph
1✔
54
    assert node3.uuid in graph
1✔
55

56
    assert node1.value == 42
1✔
57
    assert node2.value == 18
1✔
58

59
    with pytest.raises(TypeError):
1✔
60
        node3.run()
1✔
61

62
    graph.run()
1✔
63

64
    assert node1.value == 43
1✔
65
    assert node2.value == 19
1✔
66
    assert node3.result == 62
1✔
67

68

69
def test_nested_graph():
1✔
70
    """Test nested DiGraph."""
71
    with znflow.DiGraph():
1✔
72
        with pytest.raises(ValueError):
1✔
73
            with znflow.DiGraph():
1✔
74
                pass
×
75

76

77
def test_changed_graph():
1✔
78
    """Test changed DiGraph."""
79
    with pytest.raises(ValueError):
1✔
80
        with znflow.DiGraph():
1✔
81
            znflow.base.NodeBaseMixin._graph_ = znflow.DiGraph()
1✔
82
    znflow.base.NodeBaseMixin._graph_ = empty  # reset after test
1✔
83

84

85
def test_add_others():
1✔
86
    graph = znflow.DiGraph()
1✔
87
    with pytest.raises(ValueError):
1✔
88
        # it is only possible to add classes inheriting from NodeBaseMixin
89
        graph.add_node(42)
1✔
90

91

92
def test_add_connections():
1✔
93
    graph = znflow.DiGraph()
1✔
94
    with pytest.raises(ValueError):
1✔
95
        # it is only to connect Connection and NodeBaseMixin
96
        graph.add_connections(42, 42)
1✔
97

98

99
@dataclasses.dataclass
1✔
100
class ParameterContainer:
1✔
101
    value: int
1✔
102

103

104
def test_add_not_Node():
1✔
105
    parameter = ParameterContainer(value=42)
1✔
106
    with znflow.DiGraph() as graph:
1✔
107
        # ParameterContainer is not of type NodeBaseMixin,
108
        #  therefore, it won't be added to the graph
109
        node1 = DataclassNode(value=parameter.value)
1✔
110
        assert parameter.value == 42
1✔
111

112
    assert node1.value == 42
1✔
113

114
    graph.run()
1✔
115
    assert len(graph) == 1
1✔
116
    assert node1.value == 43
1✔
117

118

119
def test_not_added_to_graph():
1✔
120
    """Test, if a Node has a deliberately disabled _graph_ attribute
121

122
    You can set instance._graph_ = None to disable adding the node to the graph.
123
    """
124
    node1 = DataclassNode(value=42)
1✔
125
    node1._graph_ = None
1✔
126
    assert node1.value == 42
1✔
127

128
    with znflow.DiGraph() as graph:
1✔
129
        assert node1._graph_ is None
1✔
130
        node2 = DataclassNode(value=18)
1✔
131
        assert node2._graph_ is graph
1✔
132

133
        assert node1.value == 42
1✔
134

135
        node3 = compute_sum(node1.value, node2.value)  # test getattr
1✔
136
        node4 = ComputeSum(node1, node2)  # test AttributeToConnection
1✔
137
        assert node3.args[0] == 42  # not a connection
1✔
138
        assert isinstance(node3.args[1], znflow.Connection)
1✔
139

140
    assert node1.uuid not in graph
1✔
141
    assert node2.uuid in graph
1✔
142
    assert node3.uuid in graph
1✔
143
    assert node4.uuid in graph
1✔
144

145
    assert node1.value == 42
1✔
146
    assert node2.value == 18
1✔
147

148
    assert len(graph) == 3
1✔
149

150
    graph.run()
1✔
151

152
    assert node1.value == 42  # the value is not +1 because
1✔
153
    #  it was not added to the graph
154
    assert node2.value == 19
1✔
155
    assert node3.result == 61
1✔
156
    assert node4.result == 61
1✔
157

158

159
def test_disable_graph():
1✔
160
    graph = znflow.DiGraph()
1✔
161
    with graph:
1✔
162
        node1 = DataclassNode(value=42)
1✔
163
        assert node1._graph_ is graph
1✔
164
        with znflow.base.disable_graph():
1✔
165
            assert node1._graph_ is empty
1✔
166
        assert node1._graph_ is graph
1✔
167
    assert node1._graph_ is empty
1✔
168

169

170
def test_get_attribute():
1✔
171
    graph = znflow.DiGraph()
1✔
172
    with graph:
1✔
173
        node1 = DataclassNode(value=42)
1✔
174
        assert znflow.base.get_attribute(node1, "value") == 42
1✔
175
        assert isinstance(node1.value, znflow.Connection)
1✔
176

177
    assert znflow.base.get_attribute(node1, "value") == 42
1✔
178
    with pytest.raises(AttributeError):
1✔
179
        znflow.base.get_attribute(node1, "not_existing")
1✔
180

181
    assert znflow.base.get_attribute(node1, "not_existing", None) is None
1✔
182
    assert znflow.base.get_attribute(node1, "not_existing", 13) == 13
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc