Passed
Push — main ( 38bdeb...9355b0 )
by Eran
01:21
created

test_builders.test_networkx_builder__defaults()   A

Complexity

Conditions 2

Size

Total Lines 17
Code Lines 11

Duplication

Lines 17
Ratio 100 %

Importance

Changes 0
Metric Value
cc 2
eloc 11
nop 0
dl 17
loc 17
rs 9.85
c 0
b 0
f 0
1
import graphinate.builders
2
import networkx as nx
3
import pytest
4
5
6
@pytest.mark.parametrize('case', [0, None, "", False])
7
def test_label_converter__value__falsy(case):
8
    actual = graphinate.builders.label_converter(case, delimiter=graphinate.builders.DEFAULT_NODE_DELIMITER)
9
    assert actual == case
10
11
12
def test_encoding():
13
    expected_edge = (("parent_a", "child_a"), ("parent_b", "child_b"))
14
15
    edge_id = graphinate.builders.encode_edge_id(expected_edge)
16
    actual_edge = graphinate.builders.decode_edge_id(edge_id)
17
18
    assert actual_edge == expected_edge
19
20
21
def test_networkx_builder__empty_model():
22
    # arrange
23
    name = ""
24
    graph_model = graphinate.model(name=name)
25
26
    # act
27
    builder = graphinate.builders.NetworkxBuilder(graph_model)
28
    graph = builder.build()
29
30
    # assert
31
    assert isinstance(graph, nx.Graph)
32
    assert graph.graph['name'] == name
33
34
35
@pytest.mark.parametrize('graph_type', list(graphinate.GraphType))
36
def test_networkx_builder__graph_type(graph_type):
37
    # arrange
38
    name = str(graph_type)
39
    graph_model = graphinate.model(name=name)
40
41
    @graph_model.edge()
42
    def edge():
43
        for i in range(5):
44
            yield {'source': i, 'target': i + 1}
45
            if graph_type in (graphinate.GraphType.DiGraph, graphinate.GraphType.MultiDiGraph):
46
                yield {'source': i + 1, 'target': i}
47
            if graph_type in (graphinate.GraphType.MultiGraph, graphinate.GraphType.MultiDiGraph):
48
                yield {'source': i, 'target': i + 1}
49
50
    # act
51
    builder = graphinate.builders.NetworkxBuilder(graph_model, graph_type=graph_type)
52
    graph = builder.build()
53
54
    # assert
55
    assert isinstance(graph, nx.Graph)
56
    assert graph.graph['name'] == name
57
58
59
def test_networkx_builder_repeating_nodes():
60
    # arrange
61
    name = 'Repeating Nodes'
62
    graph_model = graphinate.GraphModel(name=name)
63
64
    @graph_model.node()
65
    def node():
66
        for i in range(5):
67
            yield i
68
            yield i
69
70
    # act
71
    builder = graphinate.builders.NetworkxBuilder(graph_model)
72
    graph: nx.Graph = builder.build()
73
74
    # assert
75
    assert isinstance(graph, nx.Graph)
76
    assert graph.graph['name'] == name
77
    assert all(graph.nodes[n]['magnitude'] == 2 for n in graph)
78
79
80
@pytest.mark.parametrize('weight', [1.0, 1.5])
81
def test_networkx_builder_repeating_edges(weight):
82
    # arrange
83
    name = 'Repeating Edges'
84
    graph_model = graphinate.GraphModel(name=name)
85
86
    @graph_model.edge(weight=weight)
87
    def edge():
88
        for i in range(5):
89
            e = {'source': i, 'target': i + 1}
90
            yield e
91
            yield e
92
93
    # act
94
    builder = graphinate.builders.NetworkxBuilder(graph_model)
95
    graph = builder.build()
96
97
    # assert
98
    assert isinstance(graph, nx.Graph)
99
    assert graph.graph['name'] == name
100
    assert all(m == weight * 2 for *_, m in graph.edges.data('weight'))
101
102
103
def test_networkx_builder_simple_tuple():
104
    # arrange
105
    name = 'Simple Tuple'
106
    graph_model = graphinate.GraphModel(name=name)
107
108
    @graph_model.edge()
109
    def edge():
110
        for i in range(5):
111
            yield {'source': (i,), 'target': (i + 1,)}
112
113
    # act
114
    builder = graphinate.builders.NetworkxBuilder(graph_model)
115
    graph = builder.build()
116
117
    # assert
118
    assert isinstance(graph, nx.Graph)
119
    assert graph.graph['name'] == name
120
121
122
@pytest.mark.parametrize('execution_number', range(5))
123
def test_networkx_builder__map_graph_model(execution_number, map_graph_model):
124
    # arrange
125
    country_count, city_count, graph_model = map_graph_model
126
    person_count = city_count
127
128
    # act
129
    builder = graphinate.builders.NetworkxBuilder(graph_model)
130
    graph = builder.build()
131
132
    # assert
133
    assert graph.order() == country_count + city_count + person_count
134
    assert graph.graph['node_types']['country'] == country_count
135
    assert graph.graph['node_types']['city'] == city_count
136
137
138
@pytest.mark.parametrize('execution_number', range(5))
139
def test_d3_builder__map_graph_model(execution_number, map_graph_model):
140
    # arrange
141
    country_count, city_count, graph_model = map_graph_model
142
    person_count = city_count
143
144
    # act
145
    builder = graphinate.builders.D3Builder(graph_model)
146
    actual_graph = builder.build()
147
148
    # assert
149
    assert actual_graph['directed'] is False
150
    assert actual_graph['multigraph'] is False
151
    assert actual_graph['graph']['name'] == 'Map'
152
    assert actual_graph['graph']['node_types']['country'] == country_count
153
    assert actual_graph['graph']['node_types']['city'] == city_count
154
    assert len(actual_graph['nodes']) == country_count + city_count + person_count
155
156
157
def test_d3_builder__map_graph_model__both_specific_ids(map_graph_model):
158
    # arrange
159
    country_count, city_count, graph_model = map_graph_model
160
161
    # act
162
    builder = graphinate.builders.D3Builder(graph_model)
163
    actual_graph = builder.build(country_id="1", city_id="1")
164
165
    # assert
166
    assert actual_graph['directed'] is False
167
    assert actual_graph['multigraph'] is False
168
    assert actual_graph['graph']['name'] == 'Map'
169
    assert actual_graph['graph']['node_types'].get('city', 0) in (0, 1)
170
    assert actual_graph['graph']['node_types']['country'] == 1
171
    assert len(actual_graph['nodes']) in (1, 3)
172
173
174
@pytest.mark.parametrize('execution_number', range(5))
175
def test_graphql_builder__map_graph_model(execution_number, map_graph_model, graphql_query):
176
    # arrange
177
    country_count, city_count, graph_model = map_graph_model
178
    person_count = city_count
179
180
    # act
181
    builder = graphinate.builders.GraphQLBuilder(graph_model)
182
183
    import strawberry
184
    schema: strawberry.Schema = builder.build()
185
    execution_result = schema.execute_sync(graphql_query)
186
    actual_graph = execution_result.data
187
188
    # assert
189
    assert actual_graph
190
    assert actual_graph['graph']
191
    assert actual_graph['nodes']
192
    assert actual_graph['edges']
193
    assert actual_graph['graph']['name'] == 'Map'
194
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
195
    assert node_types_counts['country'] == country_count
196
    assert node_types_counts['city'] == city_count
197
    assert len(actual_graph['nodes']) == country_count + city_count + person_count
198
199
200
graphql_operations_cases = [
201
    ("""{
202
      empty: measure(measure: is_empty) {
203
        name
204
        value
205
      }
206
      directed: measure(measure: is_directed) {
207
        name
208
        value
209
      }
210
      planar: measure(measure: is_planar) {
211
        name
212
        value
213
      }
214
      connectivity: measure(measure: is_connected) {
215
        name
216
        value
217
      }
218
      node_connectivity: measure(measure: node_connectivity) {
219
        name
220
        value
221
      }
222
      threshold_graph: measure(measure: is_threshold_graph) {
223
        name
224
        value
225
      }
226
    }""", {
227
        "empty": {
228
            "name": "is_empty",
229
            "value": 0
230
        },
231
        "directed": {
232
            "name": "is_directed",
233
            "value": 0
234
        },
235
        "planar": {
236
            "name": "is_planar",
237
            "value": 1
238
        },
239
        "connectivity": {
240
            "name": "is_connected",
241
            "value": 1
242
        },
243
        "node_connectivity": {
244
            "name": "node_connectivity",
245
            "value": 2
246
        },
247
        "threshold_graph": {
248
            "name": "is_threshold_graph",
249
            "value": 0
250
        }
251
252
    }),
253
    ("""
254
    query Graph {
255
      nodes(nodeId: "H4sIAFs7E2UC/2tgmcrKAAHeDK1T9ADQP1FHEAAAAA==") {type label}
256
      edges(edgeId: "H4sIAFs7E2UC/2tgmZrIAAE9Oh4mxZ6ObsXmrkahzvpGJem5yUXejo4eqS7ehiGWji6BAYZuHq6OIGBrOwW38iywcmfDcH9jfbjytil6AHhudC5sAAAA") {type label}
257
    }
258
    """, {
259
        "nodes": [
260
            {
261
                "type": "node",
262
                "label": "0"
263
            }
264
        ],
265
        "edges": [
266
            {
267
                "type": "edge",
268
                "label": "{'source': 0, 'target': 1}"
269
            }
270
        ]
271
    }),
272
    ("""
273
    query Graph {
274
      nodes(nodeId: "H4sIAFs7E2UC/2tgmcrKAAHeDK1T9ADQP1FHEAAAAA==") {type label}
275
      edges(edgeId: "H4sIAFs7E2UC/2tgmZrIAAE9Oh4mxZ6ObsXmrkahzvpGJem5yUXejo4eqS7ehiGWji6BAYZuHq6OIGBrOwW38iywcmfDcH9jfbjytil6AHhudC5sAAAA") {type label}
276
    }
277
    """, {
278
        "nodes": [
279
            {
280
                "type": "node",
281
                "label": "0"
282
            }
283
        ],
284
        "edges": [
285
            {
286
                "type": "edge",
287
                "label": "{'source': 0, 'target': 1}"
288
            }
289
        ]
290
    }),
291
    ("mutation {refresh}", {'refresh': True})
292
]
293
294
295
@pytest.mark.parametrize(('graphql_query', 'expected_response'), graphql_operations_cases)
296
def test_graphql_builder_query(octagonal_graph_model, graphql_query, expected_response):
297
    # act
298
    builder = graphinate.builders.GraphQLBuilder(octagonal_graph_model)
299
300
    import strawberry
301
    schema: strawberry.Schema = builder.build(
302
        default_node_attributes=graphinate.builders.Builder.default_node_attributes
303
    )
304
    execution_result = schema.execute_sync(graphql_query)
305
    actual_response = execution_result.data
306
307
    # assert
308
    assert actual_response == expected_response
309
310
311
def test_graphql_builder__ast_model__graph_query(ast_graph_model, graphql_query):
312
    # act
313
    builder = graphinate.builders.GraphQLBuilder(ast_graph_model)
314
    import strawberry
315
    schema: strawberry.Schema = builder.build()
316
    execution_result = schema.execute_sync(graphql_query)
317
    actual_graph = execution_result.data
318
319
    # assert
320
    assert actual_graph
321
    assert actual_graph['graph']
322
    assert actual_graph['nodes']
323
    assert actual_graph['edges']
324
    assert actual_graph['graph']['name'] == 'AST Graph'
325
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
326
    assert node_types_counts
327