test_builders.test_networkx_builder_repeating_nodes() - Code Metrics - erivlis/graphinate - Measure and Improve Code Quality continuously with Scrutinizer

test_networkx_builder_repeating_nodes() A
last analyzed 2025-10-26 22:50 UTC

↳ Parent: test_builders

Complexity

Conditions

Size

Total Lines	19
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	13
nop	0
dl	0
loc	19
rs	9.75
c	0
b	0
f	0

from collections import Counter

import networkx as nx
import pytest

import graphinate
import graphinate.builders
from graphinate import GraphType
from graphinate.modeling import GraphModel, Multiplicity
from graphinate.typing import UniverseNode

graph_types = [
    (nx.Graph(), GraphType.Graph),
    (nx.DiGraph(), GraphType.DiGraph),
    (nx.MultiGraph(), GraphType.MultiGraph),
    (nx.MultiDiGraph(), GraphType.MultiDiGraph)
]


@pytest.mark.parametrize(('graph', 'expected_graph_type'), graph_types)
def test_returns_graph_type_for_graph(graph, expected_graph_type):
    # Act
    actual_graph_type = GraphType.of(graph)

    # Assert
    assert actual_graph_type == expected_graph_type


def test_networkx_builder__empty_model():
    # arrange
    name = ""
    graph_model = graphinate.model(name=name)

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model)
    graph = builder.build()

    # assert
    assert isinstance(graph, nx.Graph)
    assert graph.graph['name'] == name


@pytest.mark.parametrize('graph_type', list(graphinate.GraphType))
def test_networkx_builder__graph_type(graph_type):
    # arrange
    name = str(graph_type)
    graph_model = graphinate.model(name=name)

    @graph_model.edge()
    def edge():
        for i in range(5):
            yield {'source': i, 'target': i + 1}
            if graph_type in (graphinate.GraphType.DiGraph, graphinate.GraphType.MultiDiGraph):
                yield {'source': i + 1, 'target': i}
            if graph_type in (graphinate.GraphType.MultiGraph, graphinate.GraphType.MultiDiGraph):
                yield {'source': i, 'target': i + 1}

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model, graph_type=graph_type)
    graph = builder.build()

    # assert
    assert isinstance(graph, nx.Graph)
    assert graph.graph['name'] == name


def test_networkx_builder_repeating_nodes():
    # arrange
    name = 'Repeating Nodes'
    graph_model = graphinate.GraphModel(name=name)

    @graph_model.node()
    def node():
        for i in range(5):
            yield i
            yield i

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model)
    graph: nx.Graph = builder.build()

    # assert
    assert isinstance(graph, nx.Graph)
    assert graph.graph['name'] == name
    assert all(graph.nodes[n]['magnitude'] == 2 for n in graph)


@pytest.mark.parametrize('weight', [1.0, 1.5])
def test_networkx_builder_repeating_edges(weight):
    # arrange
    name = 'Repeating Edges'
    graph_model = graphinate.GraphModel(name=name)

    @graph_model.edge(weight=weight)
    def edge():
        for i in range(5):
            e = {'source': i, 'target': i + 1}
            yield e
            yield e

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model)
    graph = builder.build()

    # assert
    assert isinstance(graph, nx.Graph)
    assert graph.graph['name'] == name
    assert all(m == weight * 2 for *_, m in graph.edges.data('weight'))


def test_networkx_builder_simple_tuple():
    # arrange
    name = 'Simple Tuple'
    graph_model = graphinate.GraphModel(name=name)

    @graph_model.edge()
    def edge():
        for i in range(5):
            yield {'source': (i,), 'target': (i + 1,)}

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model)
    graph = builder.build()

    # assert
    assert isinstance(graph, nx.Graph)
    assert graph.graph['name'] == name


@pytest.mark.parametrize('execution_number', range(5))
def test_networkx_builder__map_graph_model(execution_number, map_graph_model):
    # arrange
    country_count, city_count, graph_model = map_graph_model
    person_count = city_count

    # act
    builder = graphinate.builders.NetworkxBuilder(graph_model)
    graph = builder.build()

    # assert
    assert graph.order() == country_count + city_count + person_count
    assert graph.graph['node_types']['country'] == country_count
    assert graph.graph['node_types']['city'] == city_count


@pytest.mark.parametrize('execution_number', range(5))
def test_d3_builder__map_graph_model(execution_number, map_graph_model):
    # arrange
    country_count, city_count, graph_model = map_graph_model
    person_count = city_count

    # act
    builder = graphinate.builders.D3Builder(graph_model)
    actual_graph = builder.build()

    # assert
    assert actual_graph['directed'] is False
    assert actual_graph['multigraph'] is False
    assert actual_graph['graph']['name'] == 'Map'
    assert actual_graph['graph']['node_types']['country'] == country_count
    assert actual_graph['graph']['node_types']['city'] == city_count
    assert len(actual_graph['nodes']) == country_count + city_count + person_count


def test_d3_builder__map_graph_model__both_specific_ids(map_graph_model):
    # arrange
    _, _, graph_model = map_graph_model

    # act
    builder = graphinate.builders.D3Builder(graph_model)
    actual_graph = builder.build(country_id="1", city_id="1")

    # assert
    assert actual_graph['directed'] is False
    assert actual_graph['multigraph'] is False
    assert actual_graph['graph']['name'] == 'Map'
    assert actual_graph['graph']['node_types'].get('city', 0) in (0, 1)
    assert actual_graph['graph']['node_types']['country'] == 1
    assert len(actual_graph['nodes']) in (1, 3)


@pytest.mark.parametrize('execution_number', range(5))
def test_graphql_builder__map_graph_model(execution_number, map_graph_model, graphql_query):
    # arrange
    expected_country_count, expected_city_count, graph_model = map_graph_model
    expected_person_count = expected_city_count

    # act
    builder = graphinate.builders.GraphQLBuilder(graph_model)

    import strawberry
    schema: strawberry.Schema = builder.build()
    execution_result = schema.execute_sync(graphql_query)
    actual_graph = execution_result.data

    node_ids: set = {v['id'] for v in actual_graph['nodes']}
    edges = actual_graph['edges']
    edges_source_ids: set = {v['source']['id'] for v in edges}
    edges_targets_ids: set = {v['target']['id'] for v in edges}

    # assert
    assert actual_graph
    assert actual_graph['graph']
    assert actual_graph['nodes']
    assert actual_graph['edges']
    assert actual_graph['graph']['name'] == 'Map'
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
    assert node_types_counts['country'] == expected_country_count
    assert node_types_counts['city'] == expected_city_count
    assert len(actual_graph['nodes']) == expected_country_count + expected_city_count + expected_person_count
    assert edges_source_ids.issubset(node_ids)
    assert edges_targets_ids.issubset(node_ids)
    assert node_ids.issuperset(edges_source_ids)
    assert node_ids.issuperset(edges_targets_ids)


graphql_operations_cases = [
    ("""{
      empty: measure(measure: is_empty){...Details}
      directed: measure(measure: is_directed){...Details}
      planar: measure(measure: is_planar){...Details}
      connectivity: measure(measure: is_connected){...Details}
      node_connectivity: measure(measure: node_connectivity){...Details}
      threshold_graph: measure(measure: is_threshold_graph){...Details}
    }
    fragment Details on Measure {name value}
    """, {
        "empty": {
            "name": "is_empty",
            "value": 0
        },
        "directed": {
            "name": "is_directed",
            "value": 0
        },
        "planar": {
            "name": "is_planar",
            "value": 1
        },
        "connectivity": {
            "name": "is_connected",
            "value": 1
        },
        "node_connectivity": {
            "name": "node_connectivity",
            "value": 2
        },
        "threshold_graph": {
            "name": "is_threshold_graph",
            "value": 0
        }
    }),
    ((
         'query Graph {\n'
         'nodes(nodeId: "KDAsKQ==") {type label}\n'
         'edges(edgeId: "KCdLREFzS1E9PScsICdLREVzS1E9PScp") {type label}\n'
         '}'
     ),
     # noqa: E501
     {
         "nodes": [
             {
                 "type": "node",
                 "label": "0"
             }
         ],
         "edges": [
             {
                 "type": "edge",
                 "label": "0 ⟹ 1"
             }
         ]
     }),
    ("mutation {refresh}", {'refresh': True})
]


@pytest.mark.parametrize(('graphql_query', 'expected_response'), graphql_operations_cases)
def test_graphql_builder_query(octagonal_graph_model, graphql_query, expected_response):
    # act
    builder = graphinate.builders.GraphQLBuilder(octagonal_graph_model)

    import strawberry
    schema: strawberry.Schema = builder.build(
        default_node_attributes=graphinate.builders.Builder.default_node_attributes
    )
    execution_result = schema.execute_sync(graphql_query)
    actual_response = execution_result.data

    # assert
    assert actual_response == expected_response


def test_graphql_builder__ast_model__graph_query(ast_graph_model, graphql_query):
    # act
    builder = graphinate.builders.GraphQLBuilder(ast_graph_model)
    import strawberry
    schema: strawberry.Schema = builder.build()
    execution_result = schema.execute_sync(graphql_query)
    actual_graph = execution_result.data

    # assert
    assert actual_graph
    assert actual_graph['graph']
    assert actual_graph['nodes']
    assert actual_graph['edges']
    assert actual_graph['graph']['name'] == 'AST Graph'
    node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
    assert node_types_counts


class DummyNode:
    def __init__(self, key, value):
        self.key = key
        self.value = value


class DummyNodeModel:
    def __init__(self, uniqueness=True, label=None, type_='dummy', multiplicity=Multiplicity.ALL,
                 parent_type=UniverseNode, generator=None):
        self.uniqueness = uniqueness
        self.label = label
        self.type = type_
        self.multiplicity = multiplicity
        self.parent_type = parent_type
        self.generator = generator or (lambda **kwargs: [])


@pytest.fixture
def empty_graph_model():
    model = GraphModel(name="test")
    model._node_models = {}
    return model


@pytest.fixture
def builder_with_graph(empty_graph_model):
    builder = graphinate.builders.NetworkxBuilder(empty_graph_model)
    builder._graph = nx.Graph(name="test", node_types=Counter(), edge_types=Counter())
    return builder


def test_populate_nodes_adds_new_nodes_with_attributes(builder_with_graph):
    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n1', value=42)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    assert node_id in builder_with_graph._graph
    attrs = builder_with_graph._graph.nodes[node_id]
    assert attrs['label'] == node.key
    assert attrs['type'] == 'dummynode'
    assert attrs['value'] == [node.value]
    assert attrs['magnitude'] == 1
    assert attrs['lineage'] == [node.key]
    assert 'created' in attrs


def test_populate_nodes_adds_edges_for_non_universe_parent(builder_with_graph):

    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n1', value=1)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator, parent_type='not_universe')
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    parent_node_id = builder_with_graph._parent_node_id(node_type_absolute_id)
    node_id = (node.key,)
    assert (parent_node_id, node_id) in builder_with_graph._graph.edges


def test_populate_nodes_handles_empty_generator(builder_with_graph):
    node_type_absolute_id = ('parent', 'child')

    def generator(**kwargs):
        if False:
            yield

    node_model = DummyNodeModel(generator=generator)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    assert len(builder_with_graph._graph.nodes) == 0
    assert len(builder_with_graph._graph.edges) == 0


def test_populate_nodes_callable_label_assignment(builder_with_graph):

    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n1', value=7)

    def generator(**kwargs):
        yield node

    def label_fn(value):
        return f"label-{value}"

    node_model = DummyNodeModel(generator=generator, label=label_fn)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    assert builder_with_graph._graph.nodes[node_id]['label'] == "label-7"


def test_populate_nodes_handles_invalid_parent_node_id(builder_with_graph):

    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n1', value=1)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator, parent_type=UniverseNode)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    # Should not create any edge since parent_type is UniverseNode
    assert len(builder_with_graph._graph.edges) == 0
    assert node_id in builder_with_graph._graph


def test_populate_nodes_adds_new_node(builder_with_graph):
    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n2', value=99)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    assert node_id in builder_with_graph._graph
    assert builder_with_graph._graph.nodes[node_id]['value'] == [99]


def test_populate_nodes_updates_existing_node(builder_with_graph):
    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n3', value=5)

    def generator(**kwargs):
        yield node
        yield node

    node_model = DummyNodeModel(generator=generator, multiplicity=Multiplicity.ALL)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    assert builder_with_graph._graph.nodes[node_id]['value'] == [5, 5]
    assert builder_with_graph._graph.nodes[node_id]['magnitude'] == 2


def test_populate_nodes_adds_edge_for_non_universe_parent(builder_with_graph):

    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n4', value=8)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator, parent_type='not_universe')
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    parent_node_id = builder_with_graph._parent_node_id(node_type_absolute_id)
    node_id = (node.key,)
    assert (parent_node_id, node_id) in builder_with_graph._graph.edges


def test_populate_nodes_callable_label_returns_non_string(builder_with_graph):

    node_type_absolute_id = ('parent', 'child')
    node = DummyNode(key='n5', value=123)

    def generator(**kwargs):
        yield node

    def label_fn(value):
        return 999  # Non-string label

    node_model = DummyNodeModel(generator=generator, label=label_fn)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    assert builder_with_graph._graph.nodes[node_id]['label'] == 999


def test_populate_nodes_empty_generator(builder_with_graph):
    node_type_absolute_id = ('parent', 'child')

    def generator(**kwargs):
        return
        yield

    node_model = DummyNodeModel(generator=generator)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    assert len(builder_with_graph._graph.nodes) == 0


def test_populate_nodes_universe_node_parent_handling(builder_with_graph):

    node_type_absolute_id = (UniverseNode, 'child')
    node = DummyNode(key='n6', value=55)

    def generator(**kwargs):
        yield node

    node_model = DummyNodeModel(generator=generator, parent_type=UniverseNode)
    builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
    builder_with_graph._populate_nodes(node_type_absolute_id)
    node_id = (node.key,)
    # Lineage should be (node.key,) and no edge should be created
    assert builder_with_graph._graph.nodes[node_id]['lineage'] == [node.key]
    assert len(builder_with_graph._graph.edges) == 0


1		from collections import Counter
2
3		import networkx as nx
4		import pytest
5
6		import graphinate
7		import graphinate.builders
8		from graphinate import GraphType
9		from graphinate.modeling import GraphModel, Multiplicity
10		from graphinate.typing import UniverseNode
11
12		graph_types = [
13		(nx.Graph(), GraphType.Graph),
14		(nx.DiGraph(), GraphType.DiGraph),
15		(nx.MultiGraph(), GraphType.MultiGraph),
16		(nx.MultiDiGraph(), GraphType.MultiDiGraph)
17		]
18
19
20		@pytest.mark.parametrize(('graph', 'expected_graph_type'), graph_types)
21		def test_returns_graph_type_for_graph(graph, expected_graph_type):
22		# Act
23		actual_graph_type = GraphType.of(graph)
24
25		# Assert
26		assert actual_graph_type == expected_graph_type
27
28
29		def test_networkx_builder__empty_model():
30		# arrange
31		name = ""
32		graph_model = graphinate.model(name=name)
33
34		# act
35		builder = graphinate.builders.NetworkxBuilder(graph_model)
36		graph = builder.build()
37
38		# assert
39		assert isinstance(graph, nx.Graph)
40		assert graph.graph['name'] == name
41
42
43		@pytest.mark.parametrize('graph_type', list(graphinate.GraphType))
44		def test_networkx_builder__graph_type(graph_type):
45		# arrange
46		name = str(graph_type)
47		graph_model = graphinate.model(name=name)
48
49		@graph_model.edge()
50		def edge():
51		for i in range(5):
52		yield {'source': i, 'target': i + 1}
53		if graph_type in (graphinate.GraphType.DiGraph, graphinate.GraphType.MultiDiGraph):
54		yield {'source': i + 1, 'target': i}
55		if graph_type in (graphinate.GraphType.MultiGraph, graphinate.GraphType.MultiDiGraph):
56		yield {'source': i, 'target': i + 1}
57
58		# act
59		builder = graphinate.builders.NetworkxBuilder(graph_model, graph_type=graph_type)
60		graph = builder.build()
61
62		# assert
63		assert isinstance(graph, nx.Graph)
64		assert graph.graph['name'] == name
65
66
67		def test_networkx_builder_repeating_nodes():
68		# arrange
69		name = 'Repeating Nodes'
70		graph_model = graphinate.GraphModel(name=name)
71
72		@graph_model.node()
73		def node():
74		for i in range(5):
75		yield i
76		yield i
77
78		# act
79		builder = graphinate.builders.NetworkxBuilder(graph_model)
80		graph: nx.Graph = builder.build()
81
82		# assert
83		assert isinstance(graph, nx.Graph)
84		assert graph.graph['name'] == name
85		assert all(graph.nodes[n]['magnitude'] == 2 for n in graph)
86
87
88		@pytest.mark.parametrize('weight', [1.0, 1.5])
89		def test_networkx_builder_repeating_edges(weight):
90		# arrange
91		name = 'Repeating Edges'
92		graph_model = graphinate.GraphModel(name=name)
93
94		@graph_model.edge(weight=weight)
95		def edge():
96		for i in range(5):
97		e = {'source': i, 'target': i + 1}
98		yield e
99		yield e
100
101		# act
102		builder = graphinate.builders.NetworkxBuilder(graph_model)
103		graph = builder.build()
104
105		# assert
106		assert isinstance(graph, nx.Graph)
107		assert graph.graph['name'] == name
108		assert all(m == weight * 2 for *_, m in graph.edges.data('weight'))
109
110
111		def test_networkx_builder_simple_tuple():
112		# arrange
113		name = 'Simple Tuple'
114		graph_model = graphinate.GraphModel(name=name)
115
116		@graph_model.edge()
117		def edge():
118		for i in range(5):
119		yield {'source': (i,), 'target': (i + 1,)}
120
121		# act
122		builder = graphinate.builders.NetworkxBuilder(graph_model)
123		graph = builder.build()
124
125		# assert
126		assert isinstance(graph, nx.Graph)
127		assert graph.graph['name'] == name
128
129
130		@pytest.mark.parametrize('execution_number', range(5))
131		def test_networkx_builder__map_graph_model(execution_number, map_graph_model):
132		# arrange
133		country_count, city_count, graph_model = map_graph_model
134		person_count = city_count
135
136		# act
137		builder = graphinate.builders.NetworkxBuilder(graph_model)
138		graph = builder.build()
139
140		# assert
141		assert graph.order() == country_count + city_count + person_count
142		assert graph.graph['node_types']['country'] == country_count
143		assert graph.graph['node_types']['city'] == city_count
144
145
146		@pytest.mark.parametrize('execution_number', range(5))
147		def test_d3_builder__map_graph_model(execution_number, map_graph_model):
148		# arrange
149		country_count, city_count, graph_model = map_graph_model
150		person_count = city_count
151
152		# act
153		builder = graphinate.builders.D3Builder(graph_model)
154		actual_graph = builder.build()
155
156		# assert
157		assert actual_graph['directed'] is False
158		assert actual_graph['multigraph'] is False
159		assert actual_graph['graph']['name'] == 'Map'
160		assert actual_graph['graph']['node_types']['country'] == country_count
161		assert actual_graph['graph']['node_types']['city'] == city_count
162		assert len(actual_graph['nodes']) == country_count + city_count + person_count
163
164
165		def test_d3_builder__map_graph_model__both_specific_ids(map_graph_model):
166		# arrange
167		_, _, graph_model = map_graph_model
168
169		# act
170		builder = graphinate.builders.D3Builder(graph_model)
171		actual_graph = builder.build(country_id="1", city_id="1")
172
173		# assert
174		assert actual_graph['directed'] is False
175		assert actual_graph['multigraph'] is False
176		assert actual_graph['graph']['name'] == 'Map'
177		assert actual_graph['graph']['node_types'].get('city', 0) in (0, 1)
178		assert actual_graph['graph']['node_types']['country'] == 1
179		assert len(actual_graph['nodes']) in (1, 3)
180
181
182		@pytest.mark.parametrize('execution_number', range(5))
183		def test_graphql_builder__map_graph_model(execution_number, map_graph_model, graphql_query):
184		# arrange
185		expected_country_count, expected_city_count, graph_model = map_graph_model
186		expected_person_count = expected_city_count
187
188		# act
189		builder = graphinate.builders.GraphQLBuilder(graph_model)
190
191		import strawberry
192		schema: strawberry.Schema = builder.build()
193		execution_result = schema.execute_sync(graphql_query)
194		actual_graph = execution_result.data
195
196		node_ids: set = {v['id'] for v in actual_graph['nodes']}
197		edges = actual_graph['edges']
198		edges_source_ids: set = {v['source']['id'] for v in edges}
199		edges_targets_ids: set = {v['target']['id'] for v in edges}
200
201		# assert
202		assert actual_graph
203		assert actual_graph['graph']
204		assert actual_graph['nodes']
205		assert actual_graph['edges']
206		assert actual_graph['graph']['name'] == 'Map'
207		node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
208		assert node_types_counts['country'] == expected_country_count
209		assert node_types_counts['city'] == expected_city_count
210		assert len(actual_graph['nodes']) == expected_country_count + expected_city_count + expected_person_count
211		assert edges_source_ids.issubset(node_ids)
212		assert edges_targets_ids.issubset(node_ids)
213		assert node_ids.issuperset(edges_source_ids)
214		assert node_ids.issuperset(edges_targets_ids)
215
216
217		graphql_operations_cases = [
218		("""{
219		empty: measure(measure: is_empty){...Details}
220		directed: measure(measure: is_directed){...Details}
221		planar: measure(measure: is_planar){...Details}
222		connectivity: measure(measure: is_connected){...Details}
223		node_connectivity: measure(measure: node_connectivity){...Details}
224		threshold_graph: measure(measure: is_threshold_graph){...Details}
225		}
226		fragment Details on Measure {name value}
227		""", {
228		"empty": {
229		"name": "is_empty",
230		"value": 0
231		},
232		"directed": {
233		"name": "is_directed",
234		"value": 0
235		},
236		"planar": {
237		"name": "is_planar",
238		"value": 1
239		},
240		"connectivity": {
241		"name": "is_connected",
242		"value": 1
243		},
244		"node_connectivity": {
245		"name": "node_connectivity",
246		"value": 2
247		},
248		"threshold_graph": {
249		"name": "is_threshold_graph",
250		"value": 0
251		}
252		}),
253		((
254		'query Graph {\n'
255		'nodes(nodeId: "KDAsKQ==") {type label}\n'
256		'edges(edgeId: "KCdLREFzS1E9PScsICdLREVzS1E9PScp") {type label}\n'
257		'}'
258		),
259		# noqa: E501
260		{
261		"nodes": [
262		{
263		"type": "node",
264		"label": "0"
265		}
266		],
267		"edges": [
268		{
269		"type": "edge",
270		"label": "0 ⟹ 1"
271		}
272		]
273		}),
274		("mutation {refresh}", {'refresh': True})
275		]
276
277
278		@pytest.mark.parametrize(('graphql_query', 'expected_response'), graphql_operations_cases)
279		def test_graphql_builder_query(octagonal_graph_model, graphql_query, expected_response):
280		# act
281		builder = graphinate.builders.GraphQLBuilder(octagonal_graph_model)
282
283		import strawberry
284		schema: strawberry.Schema = builder.build(
285		default_node_attributes=graphinate.builders.Builder.default_node_attributes
286		)
287		execution_result = schema.execute_sync(graphql_query)
288		actual_response = execution_result.data
289
290		# assert
291		assert actual_response == expected_response
292
293
294		def test_graphql_builder__ast_model__graph_query(ast_graph_model, graphql_query):
295		# act
296		builder = graphinate.builders.GraphQLBuilder(ast_graph_model)
297		import strawberry
298		schema: strawberry.Schema = builder.build()
299		execution_result = schema.execute_sync(graphql_query)
300		actual_graph = execution_result.data
301
302		# assert
303		assert actual_graph
304		assert actual_graph['graph']
305		assert actual_graph['nodes']
306		assert actual_graph['edges']
307		assert actual_graph['graph']['name'] == 'AST Graph'
308		node_types_counts = {c['name']: c['value'] for c in actual_graph['graph']['nodeTypeCounts']}
309		assert node_types_counts
310
311
312		class DummyNode:
313		def __init__(self, key, value):
314		self.key = key
315		self.value = value
316
317
318		class DummyNodeModel:
319		def __init__(self, uniqueness=True, label=None, type_='dummy', multiplicity=Multiplicity.ALL,
320		parent_type=UniverseNode, generator=None):
321		self.uniqueness = uniqueness
322		self.label = label
323		self.type = type_
324		self.multiplicity = multiplicity
325		self.parent_type = parent_type
326		self.generator = generator or (lambda **kwargs: [])
327
328
329		@pytest.fixture
330		def empty_graph_model():
331		model = GraphModel(name="test")
332		model._node_models = {}
333		return model
334
335
336		@pytest.fixture
337		def builder_with_graph(empty_graph_model):
338		builder = graphinate.builders.NetworkxBuilder(empty_graph_model)
339		builder._graph = nx.Graph(name="test", node_types=Counter(), edge_types=Counter())
340		return builder
341
342
343		def test_populate_nodes_adds_new_nodes_with_attributes(builder_with_graph):
344		node_type_absolute_id = ('parent', 'child')
345		node = DummyNode(key='n1', value=42)
346
347		def generator(**kwargs):
348		yield node
349
350		node_model = DummyNodeModel(generator=generator)
351		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
352		builder_with_graph._populate_nodes(node_type_absolute_id)
353		node_id = (node.key,)
354		assert node_id in builder_with_graph._graph
355		attrs = builder_with_graph._graph.nodes[node_id]
356		assert attrs['label'] == node.key
357		assert attrs['type'] == 'dummynode'
358		assert attrs['value'] == [node.value]
359		assert attrs['magnitude'] == 1
360		assert attrs['lineage'] == [node.key]
361		assert 'created' in attrs
362
363
364	View Code Duplication	def test_populate_nodes_adds_edges_for_non_universe_parent(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
365		node_type_absolute_id = ('parent', 'child')
366		node = DummyNode(key='n1', value=1)
367
368		def generator(**kwargs):
369		yield node
370
371		node_model = DummyNodeModel(generator=generator, parent_type='not_universe')
372		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
373		builder_with_graph._populate_nodes(node_type_absolute_id)
374		parent_node_id = builder_with_graph._parent_node_id(node_type_absolute_id)
375		node_id = (node.key,)
376		assert (parent_node_id, node_id) in builder_with_graph._graph.edges
377
378
379		def test_populate_nodes_handles_empty_generator(builder_with_graph):
380		node_type_absolute_id = ('parent', 'child')
381
382		def generator(**kwargs):
383		if False:
384		yield
385
386		node_model = DummyNodeModel(generator=generator)
387		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
388		builder_with_graph._populate_nodes(node_type_absolute_id)
389		assert len(builder_with_graph._graph.nodes) == 0
390		assert len(builder_with_graph._graph.edges) == 0
391
392
393	View Code Duplication	def test_populate_nodes_callable_label_assignment(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
394		node_type_absolute_id = ('parent', 'child')
395		node = DummyNode(key='n1', value=7)
396
397		def generator(**kwargs):
398		yield node
399
400		def label_fn(value):
401		return f"label-{value}"
402
403		node_model = DummyNodeModel(generator=generator, label=label_fn)
404		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
405		builder_with_graph._populate_nodes(node_type_absolute_id)
406		node_id = (node.key,)
407		assert builder_with_graph._graph.nodes[node_id]['label'] == "label-7"
408
409
410	View Code Duplication	def test_populate_nodes_handles_invalid_parent_node_id(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
411		node_type_absolute_id = ('parent', 'child')
412		node = DummyNode(key='n1', value=1)
413
414		def generator(**kwargs):
415		yield node
416
417		node_model = DummyNodeModel(generator=generator, parent_type=UniverseNode)
418		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
419		builder_with_graph._populate_nodes(node_type_absolute_id)
420		node_id = (node.key,)
421		# Should not create any edge since parent_type is UniverseNode
422		assert len(builder_with_graph._graph.edges) == 0
423		assert node_id in builder_with_graph._graph
424
425
426		def test_populate_nodes_adds_new_node(builder_with_graph):
427		node_type_absolute_id = ('parent', 'child')
428		node = DummyNode(key='n2', value=99)
429
430		def generator(**kwargs):
431		yield node
432
433		node_model = DummyNodeModel(generator=generator)
434		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
435		builder_with_graph._populate_nodes(node_type_absolute_id)
436		node_id = (node.key,)
437		assert node_id in builder_with_graph._graph
438		assert builder_with_graph._graph.nodes[node_id]['value'] == [99]
439
440
441		def test_populate_nodes_updates_existing_node(builder_with_graph):
442		node_type_absolute_id = ('parent', 'child')
443		node = DummyNode(key='n3', value=5)
444
445		def generator(**kwargs):
446		yield node
447		yield node
448
449		node_model = DummyNodeModel(generator=generator, multiplicity=Multiplicity.ALL)
450		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
451		builder_with_graph._populate_nodes(node_type_absolute_id)
452		node_id = (node.key,)
453		assert builder_with_graph._graph.nodes[node_id]['value'] == [5, 5]
454		assert builder_with_graph._graph.nodes[node_id]['magnitude'] == 2
455
456
457	View Code Duplication	def test_populate_nodes_adds_edge_for_non_universe_parent(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
458		node_type_absolute_id = ('parent', 'child')
459		node = DummyNode(key='n4', value=8)
460
461		def generator(**kwargs):
462		yield node
463
464		node_model = DummyNodeModel(generator=generator, parent_type='not_universe')
465		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
466		builder_with_graph._populate_nodes(node_type_absolute_id)
467		parent_node_id = builder_with_graph._parent_node_id(node_type_absolute_id)
468		node_id = (node.key,)
469		assert (parent_node_id, node_id) in builder_with_graph._graph.edges
470
471
472	View Code Duplication	def test_populate_nodes_callable_label_returns_non_string(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
473		node_type_absolute_id = ('parent', 'child')
474		node = DummyNode(key='n5', value=123)
475
476		def generator(**kwargs):
477		yield node
478
479		def label_fn(value):
480		return 999 # Non-string label
481
482		node_model = DummyNodeModel(generator=generator, label=label_fn)
483		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
484		builder_with_graph._populate_nodes(node_type_absolute_id)
485		node_id = (node.key,)
486		assert builder_with_graph._graph.nodes[node_id]['label'] == 999
487
488
489		def test_populate_nodes_empty_generator(builder_with_graph):
490		node_type_absolute_id = ('parent', 'child')
491
492		def generator(**kwargs):
493		return
494		yield
495
496		node_model = DummyNodeModel(generator=generator)
497		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
498		builder_with_graph._populate_nodes(node_type_absolute_id)
499		assert len(builder_with_graph._graph.nodes) == 0
500
501
502	View Code Duplication	def test_populate_nodes_universe_node_parent_handling(builder_with_graph):
		0 ignored issues – show Duplication introduced 2025-08-30 23:22 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
503		node_type_absolute_id = (UniverseNode, 'child')
504		node = DummyNode(key='n6', value=55)
505
506		def generator(**kwargs):
507		yield node
508
509		node_model = DummyNodeModel(generator=generator, parent_type=UniverseNode)
510		builder_with_graph.model._node_models = {node_type_absolute_id: [node_model]}
511		builder_with_graph._populate_nodes(node_type_absolute_id)
512		node_id = (node.key,)
513		# Lineage should be (node.key,) and no edge should be created
514		assert builder_with_graph._graph.nodes[node_id]['lineage'] == [node.key]
515		assert len(builder_with_graph._graph.edges) == 0
516

erivlis / graphinate

test_networkx_builder_repeating_nodes() A last analyzed 2025-10-26 22:50 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

test_networkx_builder_repeating_nodes() A
last analyzed 2025-10-26 22:50 UTC