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

conftest._ast_nodes() A
last analyzed 2025-11-02 22:30 UTC

↳ Parent: conftest

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	5
nop	1
dl	0
loc	5
rs	10
c	0
b	0
f	0

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

import graphinate


@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 = graphinate.model(name='AST Graph')

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

erivlis / graphinate

conftest._ast_nodes() A last analyzed 2025-11-02 22:30 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

conftest._ast_nodes() A
last analyzed 2025-11-02 22:30 UTC