Passed
Push — main ( be7884...25608f )
by Eran
01:51
created

test_graphql_builder_query_specific_elements()   A

Complexity

Conditions 1

Size

Total Lines 35
Code Lines 16

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 16
nop 1
dl 0
loc 35
rs 9.6
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 View Code Duplication
def test_networkx_builder_simple_tuple():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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 View Code Duplication
def test_networkx_builder__defaults():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
123
    # arrange
124
    name = 'Simple Tuple'
125
    graph_model = graphinate.GraphModel(name=name)
126
127
    @graph_model.edge()
128
    def edge():
129
        for i in range(5):
130
            yield {'source': (i,), 'target': (i + 1,)}
131
132
    # act
133
    builder = graphinate.builders.NetworkxBuilder(graph_model)
134
    graph = builder.build()
135
136
    # assert
137
    assert isinstance(graph, nx.Graph)
138
    assert graph.graph['name'] == name
139
140
141
@pytest.mark.parametrize('execution_number', range(5))
142
def test_networkx_builder__map_graph_model(execution_number, map_graph_model):
143
    # arrange
144
    country_count, city_count, graph_model = map_graph_model
145
    person_count = city_count
146
147
    # act
148
    builder = graphinate.builders.NetworkxBuilder(graph_model)
149
    graph = builder.build()
150
151
    # assert
152
    assert graph.order() == country_count + city_count + person_count
153
    assert graph.graph['node_types']['country'] == country_count
154
    assert graph.graph['node_types']['city'] == city_count
155
156
157
@pytest.mark.parametrize('execution_number', range(5))
158
def test_d3_builder__map_graph_model(execution_number, map_graph_model):
159
    # arrange
160
    country_count, city_count, graph_model = map_graph_model
161
    person_count = city_count
162
    # act
163
    builder = graphinate.builders.D3Builder(graph_model)
164
    actual_graph = builder.build()
165
166
    # assert
167
    assert actual_graph['directed'] is False
168
    assert actual_graph['multigraph'] is False
169
    assert actual_graph['graph']['name'] == 'Map'
170
    assert actual_graph['graph']['node_types']['city'] == city_count
171
    assert actual_graph['graph']['node_types']['country'] == country_count
172
    assert len(actual_graph['nodes']) == country_count + city_count + person_count
173
174
175
@pytest.mark.parametrize('execution_number', range(5))
176
def test_graphql_builder__map_graph_model(execution_number, map_graph_model, graphql_query):
177
    # arrange
178
    country_count, city_count, graph_model = map_graph_model
179
    person_count = city_count
180
181
    # act
182
    builder = graphinate.builders.GraphQLBuilder(graph_model)
183
184
    import strawberry
185
    schema: strawberry.Schema = builder.build()
186
    execution_result = schema.execute_sync(graphql_query)
187
    actual_graph = execution_result.data
188
189
    # assert
190
    assert actual_graph
191
    assert actual_graph['graph']
192
    assert actual_graph['nodes']
193
    assert actual_graph['edges']
194
    assert actual_graph['graph']['name'] == 'Map'
195
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
196
    assert node_types_counts['country'] == country_count
197
    assert node_types_counts['city'] == city_count
198
    assert len(actual_graph['nodes']) == country_count + city_count + person_count
199
200
201
def test_graphql_builder_measures(octagonal_graph_model):
202
    # arrange
203
    measures_graphql_query = """{
204
      empty: measure(measure: is_empty) {
205
        name
206
        value
207
      }
208
      directed: measure(measure: is_directed) {
209
        name
210
        value
211
      }
212
      planar: measure(measure: is_planar) {
213
        name
214
        value
215
      }
216
      connectivity: measure(measure: is_connected) {
217
        name
218
        value
219
      }
220
      node_connectivity: measure(measure: node_connectivity) {
221
        name
222
        value
223
      }
224
      threshold_graph: measure(measure: is_threshold_graph) {
225
        name
226
        value
227
      }
228
    }"""
229
    expected_response = {
230
        "empty": {
231
            "name": "is_empty",
232
            "value": 0
233
        },
234
        "directed": {
235
            "name": "is_directed",
236
            "value": 0
237
        },
238
        "planar": {
239
            "name": "is_planar",
240
            "value": 1
241
        },
242
        "connectivity": {
243
            "name": "is_connected",
244
            "value": 1
245
        },
246
        "node_connectivity": {
247
            "name": "node_connectivity",
248
            "value": 2
249
        },
250
        "threshold_graph": {
251
            "name": "is_threshold_graph",
252
            "value": 0
253
        }
254
255
    }
256
257
    # act
258
    builder = graphinate.builders.GraphQLBuilder(octagonal_graph_model)
259
260
    import strawberry
261
    schema: strawberry.Schema = builder.build(
262
        default_node_attributes=graphinate.builders.Builder.default_node_attributes
263
    )
264
    execution_result = schema.execute_sync(measures_graphql_query)
265
    actual_response = execution_result.data
266
267
    # assert
268
    assert actual_response == expected_response
269
270
271
def test_graphql_builder_query_specific_elements(octagonal_graph_model):
272
    # arrange
273
    graphql_query = """
274
    query Graph {
275
      nodes(nodeId: "H4sIAFs7E2UC/2tgmcrKAAHeDK1T9ADQP1FHEAAAAA==") {type label}
276
      edges(edgeId: "H4sIAFs7E2UC/2tgmZrIAAE9Oh4mxZ6ObsXmrkahzvpGJem5yUXejo4eqS7ehiGWji6BAYZuHq6OIGBrOwW38iywcmfDcH9jfbjytil6AHhudC5sAAAA") {type label}
277
    }
278
    """
279
    expected_response = {
280
        "nodes": [
281
            {
282
                "type": "node",
283
                "label": "0"
284
            }
285
        ],
286
        "edges": [
287
            {
288
                "type": "edge",
289
                "label": "{'source': 0, 'target': 1}"
290
            }
291
        ]
292
    }
293
294
    # act
295
    builder = graphinate.builders.GraphQLBuilder(octagonal_graph_model)
296
297
    import strawberry
298
    schema: strawberry.Schema = builder.build(
299
        default_node_attributes=graphinate.builders.Builder.default_node_attributes
300
    )
301
    execution_result = schema.execute_sync(graphql_query)
302
    actual_response = execution_result.data
303
304
    # assert
305
    assert actual_response == expected_response
306
307
308
def test_graphql_builder__ast_model__graph_query(ast_graph_model, graphql_query):
309
    # act
310
    builder = graphinate.builders.GraphQLBuilder(ast_graph_model)
311
    import strawberry
312
    schema: strawberry.Schema = builder.build()
313
    execution_result = schema.execute_sync(graphql_query)
314
    actual_graph = execution_result.data
315
316
    # assert
317
    assert actual_graph
318
    assert actual_graph['graph']
319
    assert actual_graph['nodes']
320
    assert actual_graph['edges']
321
    assert actual_graph['graph']['name'] == 'AST Graph'
322
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
323
    assert node_types_counts
324
325
326
def test_graphql_builder__ast_model__refresh_mutation(ast_graph_model):
327
    # act
328
    builder = graphinate.builders.GraphQLBuilder(ast_graph_model)
329
    import strawberry
330
    schema: strawberry.Schema = builder.build()
331
    execution_result = schema.execute_sync("mutation {refresh}")
332
    actual = execution_result.data
333
334
    # assert
335
    assert actual
336