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

zincware / ZnFlow / 4449865450

pending completion
4449865450

Pull #53

github

GitHub
Merge 7cdc1e0a2 into 3cf29bd73
Pull Request #53: add tests for result attribute access

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

1138 of 1169 relevant lines covered (97.35%)

0.97 hits per line

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

97.83
/tests/test_node.py
1
import dataclasses
1✔
2

3
import attrs
1✔
4
import pytest
1✔
5
import zninit
1✔
6

7
import znflow
1✔
8
from znflow.node import _mark_init_in_construction
1✔
9

10

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

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

18
    @property
1✔
19
    def output(self):
1✔
20
        return znflow.get_attribute(self, "value")
1✔
21

22
    def do_stuff(self):
1✔
23
        raise NotImplementedError
24

25

26
@dataclasses.dataclass
1✔
27
class DataclassNode(znflow.Node):
1✔
28
    value: int
1✔
29

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

33
    @property
1✔
34
    def output(self):
1✔
35
        return znflow.get_attribute(self, "value")
1✔
36

37

38
class ZnInitNode(zninit.ZnInit, znflow.Node):
1✔
39
    value: int = zninit.Descriptor()
1✔
40

41
    def run(self):
1✔
42
        self.value += 1
×
43

44
    @property
1✔
45
    def output(self):
1✔
46
        return znflow.get_attribute(self, "value")
1✔
47

48

49
@attrs.define
1✔
50
class AttrsNode(znflow.Node):
1✔
51
    value: int
1✔
52

53
    def run(self):
1✔
54
        self.value += 1
×
55

56
    @property
1✔
57
    def output(self):
1✔
58
        return znflow.get_attribute(self, "value")
1✔
59

60

61
@znflow.nodify
1✔
62
def add(value):
1✔
63
    return value
×
64

65

66
@znflow.nodify
1✔
67
def compute_sum(*args):
1✔
68
    return sum(args)
×
69

70

71
@pytest.mark.parametrize("cls", [PlainNode, DataclassNode, ZnInitNode, add, AttrsNode])
1✔
72
def test_Node_init(cls):
1✔
73
    with pytest.raises((TypeError, AttributeError)):
1✔
74
        # TODO only raise TypeError and not AttributeError when TypeError is expected.
75
        with znflow.DiGraph():
1✔
76
            cls()
1✔
77

78

79
@pytest.mark.parametrize("cls", [PlainNode, DataclassNode, ZnInitNode, add, AttrsNode])
1✔
80
def test_Node(cls):
1✔
81
    with znflow.DiGraph() as graph:
1✔
82
        node = cls(value=42)
1✔
83

84
    if isinstance(node, (PlainNode, DataclassNode, ZnInitNode)):
1✔
85
        assert node.value == 42
1✔
86
    elif isinstance(node, znflow.FunctionFuture):
1✔
87
        assert node.kwargs["value"] == 42
1✔
88

89
    assert node.uuid in graph
1✔
90
    assert graph.nodes[node.uuid]["value"] is node
1✔
91

92

93
@pytest.mark.parametrize("cls2", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
94
@pytest.mark.parametrize("cls1", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
95
def test_ConnectionNodeNode(cls1, cls2):
1✔
96
    with znflow.DiGraph() as graph:
1✔
97
        node1 = cls1(value=42)
1✔
98
        node2 = cls2(value=node1.value)
1✔
99

100
    assert isinstance(node2.value, znflow.Connection)
1✔
101
    assert node2.value.uuid == node1.uuid
1✔
102

103
    assert node1.uuid in graph
1✔
104
    assert node2.uuid in graph
1✔
105

106
    edge: dict = graph.get_edge_data(node1.uuid, node2.uuid)
1✔
107
    assert edge is not None
1✔
108
    # we have one connection, so we use 0
109
    assert edge[0]["u_attr"] == "value"
1✔
110
    assert edge[0]["v_attr"] == "value"
1✔
111

112

113
@pytest.mark.parametrize("cls2", [add])
1✔
114
@pytest.mark.parametrize("cls1", [add])
1✔
115
def test_ConnectionNodifyNodify(cls1, cls2):
1✔
116
    with znflow.DiGraph() as graph:
1✔
117
        node1 = cls1(value=42)
1✔
118
        node2 = cls2(value=node1)
1✔
119

120
    assert node1.uuid in graph
1✔
121
    assert node2.uuid in graph
1✔
122

123
    edge: dict = graph.get_edge_data(node1.uuid, node2.uuid)
1✔
124

125
    assert edge is not None
1✔
126
    # # we have one connection, so we use 0
127
    assert edge[0]["u_attr"] is None
1✔
128

129

130
@pytest.mark.parametrize("cls1", [add])
1✔
131
@pytest.mark.parametrize("cls2", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
132
def test_ConnectionNodeNodify(cls1, cls2):
1✔
133
    with znflow.DiGraph() as graph:
1✔
134
        node1 = cls1(value=42)
1✔
135
        node2 = cls2(value=node1)
1✔
136

137
    assert node1.uuid in graph
1✔
138
    assert node2.uuid in graph
1✔
139

140
    assert isinstance(node2.value, znflow.Connection)
1✔
141

142
    edge: dict = graph.get_edge_data(node1.uuid, node2.uuid)
1✔
143
    assert edge is not None
1✔
144
    # we have one connection, so we use 0
145
    assert edge[0]["u_attr"] is None
1✔
146
    assert edge[0]["v_attr"] == "value"
1✔
147

148

149
@pytest.mark.parametrize("cls2", [add])
1✔
150
@pytest.mark.parametrize("cls1", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
151
def test_ConnectionNodifyNode(cls1, cls2):
1✔
152
    with znflow.DiGraph() as graph:
1✔
153
        node1 = cls1(value=42)
1✔
154
        node2 = cls2(value=node1)
1✔
155

156
    assert node1.uuid in graph
1✔
157
    assert node2.uuid in graph
1✔
158

159
    edge: dict = graph.get_edge_data(node1.uuid, node2.uuid)
1✔
160
    assert edge is not None
1✔
161
    # we have one connection, so we use 0
162
    assert edge[0]["u_attr"] is None
1✔
163

164

165
@pytest.mark.parametrize("cls2", [compute_sum])
1✔
166
@pytest.mark.parametrize("cls1", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
167
def test_ConnectionNodifyMultiNode(cls1, cls2):
1✔
168
    with znflow.DiGraph() as graph:
1✔
169
        node1 = cls1(value=42)
1✔
170
        node2 = cls1(value=42)
1✔
171
        node3 = cls2(node1.value, node2.value)
1✔
172

173
    assert node1.uuid in graph
1✔
174
    assert node2.uuid in graph
1✔
175
    assert node3.uuid in graph
1✔
176
    #
177
    edge1: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
178
    edge2: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
179
    assert edge1 is not None
1✔
180
    assert edge2 is not None
1✔
181
    # # we have one connection, so we use 0
182
    assert edge1[0]["u_attr"] == "value"
1✔
183
    assert "v_attr" not in edge1[0]
1✔
184
    assert edge2[0]["u_attr"] == "value"
1✔
185
    assert "v_attr" not in edge2[0]
1✔
186

187

188
@pytest.mark.parametrize("cls1", [compute_sum])
1✔
189
@pytest.mark.parametrize("cls2", [PlainNode, DataclassNode, ZnInitNode, AttrsNode])
1✔
190
def test_ConnectionNodeMultiNodify(cls1, cls2):
1✔
191
    with znflow.DiGraph() as graph:
1✔
192
        node1 = cls1(42)
1✔
193
        node2 = cls1(42)
1✔
194
        node3 = cls2(value=[node1, node2])
1✔
195

196
    assert node1.uuid in graph
1✔
197
    assert node2.uuid in graph
1✔
198
    assert node3.uuid in graph
1✔
199

200
    edge1: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
201
    edge2: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
202
    assert edge1 is not None
1✔
203
    assert edge2 is not None
1✔
204

205
    assert edge1[0]["u_attr"] is None
1✔
206
    assert edge1[0]["v_attr"] == "value"
1✔
207

208
    assert edge2[0]["u_attr"] is None
1✔
209
    assert edge2[0]["v_attr"] == "value"
1✔
210

211

212
def test_Connection():
1✔
213
    with znflow.DiGraph() as graph:
1✔
214
        node1 = PlainNode(value=42)
1✔
215
        node2 = PlainNode(value=node1)
1✔
216

217
        # node1.value = node2.value
218
    assert isinstance(node2.value, znflow.Connection)
1✔
219
    assert node2.value.uuid == node1.uuid
1✔
220
    assert node2.value.attribute is None
1✔
221
    #
222
    assert node1.uuid in graph
1✔
223
    assert node2.uuid in graph
1✔
224
    #
225
    edge: dict = graph.get_edge_data(node1.uuid, node2.uuid)
1✔
226
    assert edge is not None
1✔
227
    assert edge[0]["u_attr"] is None
1✔
228
    assert edge[0]["v_attr"] == "value"
1✔
229

230

231
def test_CheckWrapInit():
1✔
232
    @_mark_init_in_construction
1✔
233
    class CheckWrapInit:
1✔
234
        _in_construction: bool = False
1✔
235

236
        def __init__(self):
1✔
237
            assert self._in_construction
1✔
238

239
            return 42
1✔
240

241
    with pytest.raises(TypeError):
1✔
242
        CheckWrapInit()
1✔
243

244

245
@dataclasses.dataclass
1✔
246
class DictionaryConnection(znflow.Node):
1✔
247
    nodes: dict
1✔
248
    results: float = None
1✔
249

250
    def run(self):
1✔
251
        return sum(self.nodes.values())
1✔
252

253

254
@dataclasses.dataclass
1✔
255
class ListConnection(znflow.Node):
1✔
256
    nodes: list
1✔
257
    results: float = None
1✔
258

259
    def run(self):
1✔
260
        return sum(self.nodes)
1✔
261

262

263
def test_DictionaryConnection():
1✔
264
    with znflow.DiGraph() as graph:
1✔
265
        node1 = PlainNode(value=42)
1✔
266
        node2 = PlainNode(value=42)
1✔
267
        node3 = DictionaryConnection(nodes={"node1": node1.value, "node2": node2.value})
1✔
268

269
    assert node1.uuid in graph
1✔
270
    assert node2.uuid in graph
1✔
271
    assert node3.uuid in graph
1✔
272

273
    assert "node1" in node3.nodes
1✔
274
    assert "node2" in node3.nodes
1✔
275

276
    assert isinstance(node3.nodes["node1"], znflow.Connection)
1✔
277
    assert node3.nodes["node1"].uuid == node1.uuid
1✔
278
    assert node3.nodes["node1"].attribute == "value"
1✔
279

280
    assert isinstance(node3.nodes["node2"], znflow.Connection)
1✔
281
    assert node3.nodes["node2"].uuid == node2.uuid
1✔
282
    assert node3.nodes["node2"].attribute == "value"
1✔
283

284
    graph.run()
1✔
285

286
    edge1: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
287
    edge2: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
288
    assert edge1 is not None
1✔
289
    assert edge2 is not None
1✔
290

291
    assert isinstance(edge1, dict)
1✔
292
    assert edge1[0]["u_attr"] == "value"
1✔
293
    assert edge1[0]["v_attr"] == "nodes"
1✔
294

295
    assert isinstance(edge2, dict)
1✔
296
    assert edge2[0]["u_attr"] == "value"
1✔
297
    assert edge2[0]["v_attr"] == "nodes"
1✔
298

299

300
def test_ListConnection():
1✔
301
    with znflow.DiGraph() as graph:
1✔
302
        node1 = PlainNode(value=42)
1✔
303
        node2 = PlainNode(value=42)
1✔
304
        node3 = ListConnection(nodes=[node1.value, node2.value])
1✔
305

306
    assert node1.uuid in graph
1✔
307
    assert node2.uuid in graph
1✔
308
    assert node3.uuid in graph
1✔
309

310
    assert isinstance(node3.nodes[0], znflow.Connection)
1✔
311
    assert node3.nodes[0].uuid == node1.uuid
1✔
312
    assert node3.nodes[0].attribute == "value"
1✔
313

314
    assert isinstance(node3.nodes[1], znflow.Connection)
1✔
315
    assert node3.nodes[1].uuid == node2.uuid
1✔
316
    assert node3.nodes[1].attribute == "value"
1✔
317

318
    graph.run()
1✔
319

320
    edge1: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
321
    edge2: dict = graph.get_edge_data(node1.uuid, node3.uuid)
1✔
322
    assert edge1 is not None
1✔
323
    assert edge2 is not None
1✔
324

325
    assert isinstance(edge1, dict)
1✔
326
    assert edge1[0]["u_attr"] == "value"
1✔
327
    assert edge1[0]["v_attr"] == "nodes"
1✔
328

329
    assert isinstance(edge2, dict)
1✔
330
    assert edge2[0]["u_attr"] == "value"
1✔
331
    assert edge2[0]["v_attr"] == "nodes"
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