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

conftest.map_graph_model() D
last analyzed 2025-11-29 23:13 UTC

↳ Parent: conftest

Complexity

Conditions

Size

Total Lines	48
Code Lines	34

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	13
eloc	34
nop	2
dl	0
loc	48
rs	4.2
c	0
b	0
f	0

How to fix Complexity

import ast
import hashlib
import inspect
import operator
import pickle
import random
from _ast import AST
from collections.abc import Iterable

import faker
import pytest

from graphinate import model
from graphinate.builders import D3Builder


@pytest.fixture
def country_count():
    return random.randint(1, 10)


@pytest.fixture
def city_count():
    return random.randint(20, 40)


def _ast_nodes(parsed_asts: Iterable[AST]):
    for item in parsed_asts:
        if not isinstance(item, ast.Load):
            yield item
            yield from _ast_nodes(ast.iter_child_nodes(item))


def _ast_edge(parsed_ast: AST):

    for child_ast in ast.iter_child_nodes(parsed_ast):
        if not isinstance(child_ast, ast.Load):
            edge = {'source': parsed_ast, 'target': child_ast}
            edge_types = (field_name for field_name, value in ast.iter_fields(parsed_ast) if
                          child_ast == value or (child_ast in value if isinstance(value, list) else False))
            edge_type = next(edge_types, None)
            if edge_type:
                edge['type'] = edge_type
            yield edge
            yield from _ast_edge(child_ast)


@pytest.fixture

def ast_graph_model():
    graph_model = model(name='AST Graph')

    root_ast_node = ast.parse(inspect.getsource(D3Builder))

    def node_type(ast_node):
        return ast_node.__class__.__name__

    def node_label(ast_node) -> str:
        label = ast_node.__class__.__name__

        for field_name in ('name', 'id'):
            if field_name in ast_node._fields:
                label = f"{label}\n{field_name}: {operator.attrgetter(field_name)(ast_node)}"

        return label

    def key(value):
        # noinspection InsecureHash
        return hashlib.shake_128(pickle.dumps(value)).hexdigest(20)

    def endpoint(value, endpoint_name):
        return key(value[endpoint_name])

    def source(value):
        return endpoint(value, 'source')

    def target(value):
        return endpoint(value, 'target')

    @graph_model.node(type_=node_type, key=key, label=node_label, unique=True)
    def ast_node(**kwargs):
        yield from _ast_nodes([root_ast_node])

    @graph_model.edge(type_='edge', source=source, target=target, label=operator.itemgetter('type'))
    def ast_edge(**kwargs):
        yield from _ast_edge(root_ast_node)

    return graph_model


@pytest.fixture
def map_graph_model(country_count, city_count):
    country_ids = {str(c): None for c in range(1, country_count + 1)}
    city_ids = {str(c): random.choice(list(country_ids.keys())) for c in range(1, city_count + 1)}

    graph_model = model(name='Map')

    faker.Faker.seed(0)
    fake = faker.Faker()

    def country_node_label(value):
        return fake.country()

    def city_node_label(value):
        return fake.city()

    @graph_model.node(label=country_node_label, unique=False)
    def country(country_id=None, **kwargs):

        if country_id and country_id in country_ids:
            yield country_id
        else:
            yield from country_ids

    @graph_model.node(parent_type='country', label=city_node_label, unique=False)
    def city(country_id=None, city_id=None, **kwargs):

        if country_id is None and city_id is None:
            yield from city_ids.keys()

        if country_id is None and city_id is not None and city_id in city_ids:
            yield city_id

        if city_id is not None and country_id is not None and city_ids.get(city_id) == country_id:
            yield city_id

        if country_id is not None and city_id is None:
            yield from (k for k, v in city_ids.items() if v == country_id)

    @graph_model.node(type_=operator.itemgetter('sex'),
                      parent_type='city',
                      unique= False,
                      key=operator.itemgetter('username'),
                      label=operator.itemgetter('name'))
    def person(country_id=None, city_id=None, person_id=None, **kwargs):
        yield fake.profile()

    return country_count, city_count, graph_model


@pytest.fixture
def octagonal_graph_model():
    graph_model = model(name="Octagonal Graph")
    number_of_sides = 8

    # Register edges supplier function
    @graph_model.edge()
    def edge():
        for i in range(number_of_sides):
            yield {'source': i, 'target': i + 1}
        yield {'source': number_of_sides, 'target': 0}

    return graph_model


@pytest.fixture
def graphql_query():
    return """
    query Graph {
      graph {
        name
        nodeTypeCounts {
          name
          value
        }
        edgeTypeCounts {
          name
          value
        }
        created
        nodeCount
        edgeCount
        size
        order
        radius
        diameter
        averageDegree
        hash
      }
      nodes {
        id
        ...ElementDetails
        neighbors {id type label}
        children: neighbors(children: true) {id type label}
        edges {id type label}
      }
      edges {
        source {id ...ElementDetails}
        target {id ...ElementDetails}
        ...ElementDetails
        weight
      }
    }

    fragment ElementDetails on GraphElement {
      label
      type
      label
      color
      created
      updated
    }
    """


1		import ast
2		import hashlib
3		import inspect
4		import operator
5		import pickle
6		import random
7		from _ast import AST
8		from collections.abc import Iterable
9
10		import faker
11		import pytest
12
13		from graphinate import model
14		from graphinate.builders import D3Builder
15
16
17		@pytest.fixture
18		def country_count():
19		return random.randint(1, 10)
20
21
22		@pytest.fixture
23		def city_count():
24		return random.randint(20, 40)
25
26
27		def _ast_nodes(parsed_asts: Iterable[AST]):
28		for item in parsed_asts:
29		if not isinstance(item, ast.Load):
30		yield item
31		yield from _ast_nodes(ast.iter_child_nodes(item))
32
33
34	View Code Duplication	def _ast_edge(parsed_ast: AST):
		0 ignored issues – show Duplication introduced 2023-09-26 21:28 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
35		for child_ast in ast.iter_child_nodes(parsed_ast):
36		if not isinstance(child_ast, ast.Load):
37		edge = {'source': parsed_ast, 'target': child_ast}
38		edge_types = (field_name for field_name, value in ast.iter_fields(parsed_ast) if
39		child_ast == value or (child_ast in value if isinstance(value, list) else False))
40		edge_type = next(edge_types, None)
41		if edge_type:
42		edge['type'] = edge_type
43		yield edge
44		yield from _ast_edge(child_ast)
45
46
47	View Code Duplication	@pytest.fixture
		0 ignored issues – show Duplication introduced 2023-09-26 21:28 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
48		def ast_graph_model():
49		graph_model = model(name='AST Graph')
50
51		root_ast_node = ast.parse(inspect.getsource(D3Builder))
52
53		def node_type(ast_node):
54		return ast_node.__class__.__name__
55
56		def node_label(ast_node) -> str:
57		label = ast_node.__class__.__name__
58
59		for field_name in ('name', 'id'):
60		if field_name in ast_node._fields:
61		label = f"{label}\n{field_name}: {operator.attrgetter(field_name)(ast_node)}"
62
63		return label
64
65		def key(value):
66		# noinspection InsecureHash
67		return hashlib.shake_128(pickle.dumps(value)).hexdigest(20)
68
69		def endpoint(value, endpoint_name):
70		return key(value[endpoint_name])
71
72		def source(value):
73		return endpoint(value, 'source')
74
75		def target(value):
76		return endpoint(value, 'target')
77
78		@graph_model.node(type_=node_type, key=key, label=node_label, unique=True)
79		def ast_node(**kwargs):
80		yield from _ast_nodes([root_ast_node])
81
82		@graph_model.edge(type_='edge', source=source, target=target, label=operator.itemgetter('type'))
83		def ast_edge(**kwargs):
84		yield from _ast_edge(root_ast_node)
85
86		return graph_model
87
88
89		@pytest.fixture
90		def map_graph_model(country_count, city_count):
91		country_ids = {str(c): None for c in range(1, country_count + 1)}
92		city_ids = {str(c): random.choice(list(country_ids.keys())) for c in range(1, city_count + 1)}
93
94		graph_model = model(name='Map')
95
96		faker.Faker.seed(0)
97		fake = faker.Faker()
98
99		def country_node_label(value):
100		return fake.country()
101
102		def city_node_label(value):
103		return fake.city()
104
105		@graph_model.node(label=country_node_label, unique=False)
106		def country(country_id=None, **kwargs):
107
108		if country_id and country_id in country_ids:
109		yield country_id
110		else:
111		yield from country_ids
112
113		@graph_model.node(parent_type='country', label=city_node_label, unique=False)
114		def city(country_id=None, city_id=None, **kwargs):
115
116		if country_id is None and city_id is None:
117		yield from city_ids.keys()
118
119		if country_id is None and city_id is not None and city_id in city_ids:
120		yield city_id
121
122		if city_id is not None and country_id is not None and city_ids.get(city_id) == country_id:
123		yield city_id
124
125		if country_id is not None and city_id is None:
126		yield from (k for k, v in city_ids.items() if v == country_id)
127
128		@graph_model.node(type_=operator.itemgetter('sex'),
129		parent_type='city',
130		unique= False,
131		key=operator.itemgetter('username'),
132		label=operator.itemgetter('name'))
133		def person(country_id=None, city_id=None, person_id=None, **kwargs):
134		yield fake.profile()
135
136		return country_count, city_count, graph_model
137
138
139		@pytest.fixture
140		def octagonal_graph_model():
141		graph_model = model(name="Octagonal Graph")
142		number_of_sides = 8
143
144		# Register edges supplier function
145		@graph_model.edge()
146		def edge():
147		for i in range(number_of_sides):
148		yield {'source': i, 'target': i + 1}
149		yield {'source': number_of_sides, 'target': 0}
150
151		return graph_model
152
153
154		@pytest.fixture
155		def graphql_query():
156		return """
157		query Graph {
158		graph {
159		name
160		nodeTypeCounts {
161		name
162		value
163		}
164		edgeTypeCounts {
165		name
166		value
167		}
168		created
169		nodeCount
170		edgeCount
171		size
172		order
173		radius
174		diameter
175		averageDegree
176		hash
177		}
178		nodes {
179		id
180		...ElementDetails
181		neighbors {id type label}
182		children: neighbors(children: true) {id type label}
183		edges {id type label}
184		}
185		edges {
186		source {id ...ElementDetails}
187		target {id ...ElementDetails}
188		...ElementDetails
189		weight
190		}
191		}
192
193		fragment ElementDetails on GraphElement {
194		label
195		type
196		label
197		color
198		created
199		updated
200		}
201		"""
202

erivlis / graphinate

conftest.map_graph_model() D last analyzed 2025-11-29 23:13 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

conftest.map_graph_model() D
last analyzed 2025-11-29 23:13 UTC