vlan_pool.update_database() - Code Metrics - Inspection of "Merge pull request #178 from kytos-ng/epic/vlan_ra..." - kytos-ng/topology - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5b207c...70402e )

by Vinicius

created 2023-11-29 20:25 UTC

vlan_pool.update_database() D

↳ Parent: vlan_pool

Complexity

Conditions

Size

Total Lines	67
Code Lines	51

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	51
nop	1
dl	0
loc	67
rs	4.8
c	0
b	0
f	0

How to fix Long Method Complexity

import datetime
import sys
import os
from collections import defaultdict
from kytos.core.db import Mongo

DEFAULT_TAG_RANGES = [[1, 4095]]

def get_range(vlans, avoid) -> list[list[int]]:
    """Convert available_vlans to available_tags.
    From list[int] to list[list[int]]"""
    result = []
    if not vlans:
        return result
    vlans.sort()
    i = 0
    while i < len(vlans):
        if vlans[i] in avoid:
            i += 1
        else:
            break
    if not vlans[i:]:
        return result

    start = end = vlans[i]
    for tag in vlans[i+1:]:
        if tag in avoid:
            continue
        if tag == end + 1:
            end = tag
        else:
            result.append([start, end])
            start = end = tag
    result.append([start, end])
    return result

def generate_ranges(avoid) -> [list[list[int]]]:
    """Generate available_tags only from avoid"""
    if not avoid:
        return DEFAULT_TAG_RANGES

    avoid.sort()
    ranges = []
    start = 1

    for num in avoid:
        if num > start:
            ranges.append([start, num - 1])
        start = num + 1

    if start <= 4095:
        ranges.append([start, 4095])
    return ranges

def update_database(mongo: Mongo):
    """Update database"""
    db = mongo.client[mongo.db_name]
    intf_documents = db.interface_details.find()
    evc_documents = db.evcs.find({"archived": False})

    evc_intf = defaultdict(str)

    evc_tags = defaultdict(set)

    for evc in evc_documents:
        tag_a = evc["uni_a"].get("tag")
        if tag_a:
            intf_id = evc["uni_a"]["interface_id"]
            if tag_a["value"] in evc_tags[intf_id] and isinstance(tag_a["value"], int):
                print(f"Error: Detected duplicated {tag_a['value']} TAG"
                      f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_a['value'])]}"
                      f" in interface {intf_id}")
                sys.exit(1)
            evc_tags[intf_id].add(tag_a["value"])
            evc_intf[intf_id+str(tag_a["value"])] = evc["id"]

        tag_z = evc["uni_z"].get("tag")
        if tag_z:
            intf_id = evc["uni_z"]["interface_id"]
            if tag_z["value"] in evc_tags[intf_id] and isinstance(tag_z["value"], int):
                print(f"Error: Detected duplicated {tag_z['value']} TAG"
                      f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_z['value'])]}"
                      f" in interface {intf_id}")
                sys.exit(1)
            evc_tags[intf_id].add(tag_z["value"])
            evc_intf[intf_id+str(tag_z["value"])] = evc["id"]

    intf_count = 0
    for document in intf_documents:
        avoid_tags = evc_tags.pop(document["id"], set())
        if document.get("available_vlans") is None:
            continue
        ranges = get_range(document["available_vlans"], avoid_tags)
        result = db.interface_details.update_one(
            {"id": document["id"]},
            {
                "$set": 
                {
                    "available_tags": {"vlan": ranges},
                    "tag_ranges": {"vlan": DEFAULT_TAG_RANGES}
                },
                "$unset": {"available_vlans": ""}
            }
        )
        intf_count += result.modified_count

    evc_intf_count = 0
    for intf_id, avoid_tags in evc_tags.items():
        available_tags = generate_ranges(list(avoid_tags))
        utc_now = datetime.datetime.utcnow()
        result = db.interface_details.insert_one({
            "_id": intf_id,
            "id": intf_id,
            "inserted_at": utc_now,
            "updated_at": utc_now,
            "available_tags": {"vlan": available_tags},
            "tag_ranges": {"vlan": DEFAULT_TAG_RANGES},
        })
        if result:
            evc_intf_count += 1

    print(f"{intf_count} documents modified. {evc_intf_count} documents inserted")

def aggregate_outdated_interfaces(mongo: Mongo):
    """Aggregate outdated inteface details"""
    db = mongo.client[mongo.db_name]
    document_ids = set()
    result = db.interface_details.aggregate(
        [
            {"$sort": {"_id": 1}},
            {"$project": {
                "_id": 0,
                "id": 1,
                "max_number": {"$max": "$available_vlans"},
                "min_number": {"$min": "$available_vlans"},
                "available_vlans": 1,
            }}
        ]
    )
    
    messages = ""
    for document in result:
        document_ids.add(document["id"])
        if document.get("available_vlans") is None:
            continue
        document.pop("available_vlans")
        messages += str(document) + "\n"


    if messages != "":
        print("Here are the outdated interfaces. 'available_vlans' have a massive"
          " amount of items, minimum and maximum items will be shown only")
        print(messages)
    
    evc_documents = db.evcs.find({"archived": False})
    evc_intf = defaultdict(str)
    evc_tags = defaultdict(set)

    for evc in evc_documents:
        tag_a = evc["uni_a"].get("tag")
        if tag_a:
            intf_id = evc["uni_a"]["interface_id"]
            if tag_a["value"] in evc_tags[intf_id] and isinstance(tag_a["value"], int):
                print(f"WARNING: Detected duplicated {tag_a['value']} TAG"
                      f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_a['value'])]}"
                      f" in interface {intf_id}")
                print()
            evc_tags[intf_id].add(tag_a["value"])
            evc_intf[intf_id+str(tag_a["value"])] = evc["id"]

        tag_z = evc["uni_z"].get("tag")
        if tag_z:
            intf_id = evc["uni_z"]["interface_id"]
            if tag_z["value"] in evc_tags[intf_id] and isinstance(tag_z["value"], int):
                print(f"WARNING: Detected duplicated {tag_z['value']} TAG"
                      f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_z['value'])]}"
                      f" in interface {intf_id}")
                print()
            evc_tags[intf_id].add(tag_z["value"])
            evc_intf[intf_id+str(tag_z["value"])] = evc["id"]

    for id_ in document_ids:
        evc_tags.pop(id_, None)

    if evc_tags:
        print("New documents are going to be created. From the next interfaces,"
              " these tags should be avoided")

    for intf, avoid_tags in evc_tags.items():
        if intf in document_ids:
            continue
        aux = {"id": intf, "avoid_tags": avoid_tags}
        print(aux)

    if not evc_tags and messages == "":
        print("There is nothing to update or add")


if __name__ == "__main__":

    mongo = Mongo()
    cmds = {
        "aggregate_outdated_interfaces": aggregate_outdated_interfaces,
        "update_database": update_database,
    }
    try:
        cmd = os.environ["CMD"]
        cmds[cmd](mongo)
    except KeyError:
        print(
            f"Please set the 'CMD' env var. \nIt has to be one of these: {list(cmds.keys())}"
        )
        sys.exit(1)


1		import datetime
2		import sys
3		import os
4		from collections import defaultdict
5		from kytos.core.db import Mongo
6
7		DEFAULT_TAG_RANGES = [[1, 4095]]
8
9		def get_range(vlans, avoid) -> list[list[int]]:
10		"""Convert available_vlans to available_tags.
11		From list[int] to list[list[int]]"""
12		result = []
13		if not vlans:
14		return result
15		vlans.sort()
16		i = 0
17		while i < len(vlans):
18		if vlans[i] in avoid:
19		i += 1
20		else:
21		break
22		if not vlans[i:]:
23		return result
24
25		start = end = vlans[i]
26		for tag in vlans[i+1:]:
27		if tag in avoid:
28		continue
29		if tag == end + 1:
30		end = tag
31		else:
32		result.append([start, end])
33		start = end = tag
34		result.append([start, end])
35		return result
36
37		def generate_ranges(avoid) -> [list[list[int]]]:
38		"""Generate available_tags only from avoid"""
39		if not avoid:
40		return DEFAULT_TAG_RANGES
41
42		avoid.sort()
43		ranges = []
44		start = 1
45
46		for num in avoid:
47		if num > start:
48		ranges.append([start, num - 1])
49		start = num + 1
50
51		if start <= 4095:
52		ranges.append([start, 4095])
53		return ranges
54
55		def update_database(mongo: Mongo):
56		"""Update database"""
57		db = mongo.client[mongo.db_name]
58		intf_documents = db.interface_details.find()
59		evc_documents = db.evcs.find({"archived": False})
60
61		evc_intf = defaultdict(str)
		0 ignored issues – show Comprehensibility Best Practice introduced 2023-11-28 16:53 UTC by Report Bug Copy Issue Report The variable `str` does not seem to be defined. Loading history...
62		evc_tags = defaultdict(set)
63
64		for evc in evc_documents:
65		tag_a = evc["uni_a"].get("tag")
66		if tag_a:
67		intf_id = evc["uni_a"]["interface_id"]
68		if tag_a["value"] in evc_tags[intf_id] and isinstance(tag_a["value"], int):
69		print(f"Error: Detected duplicated {tag_a['value']} TAG"
70		f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_a['value'])]}"
71		f" in interface {intf_id}")
72		sys.exit(1)
73		evc_tags[intf_id].add(tag_a["value"])
74		evc_intf[intf_id+str(tag_a["value"])] = evc["id"]
75
76		tag_z = evc["uni_z"].get("tag")
77		if tag_z:
78		intf_id = evc["uni_z"]["interface_id"]
79		if tag_z["value"] in evc_tags[intf_id] and isinstance(tag_z["value"], int):
80		print(f"Error: Detected duplicated {tag_z['value']} TAG"
81		f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_z['value'])]}"
82		f" in interface {intf_id}")
83		sys.exit(1)
84		evc_tags[intf_id].add(tag_z["value"])
85		evc_intf[intf_id+str(tag_z["value"])] = evc["id"]
86
87		intf_count = 0
88		for document in intf_documents:
89		avoid_tags = evc_tags.pop(document["id"], set())
90		if document.get("available_vlans") is None:
91		continue
92		ranges = get_range(document["available_vlans"], avoid_tags)
93		result = db.interface_details.update_one(
94		{"id": document["id"]},
95		{
96		"$set":
97		{
98		"available_tags": {"vlan": ranges},
99		"tag_ranges": {"vlan": DEFAULT_TAG_RANGES}
100		},
101		"$unset": {"available_vlans": ""}
102		}
103		)
104		intf_count += result.modified_count
105
106		evc_intf_count = 0
107		for intf_id, avoid_tags in evc_tags.items():
108		available_tags = generate_ranges(list(avoid_tags))
109		utc_now = datetime.datetime.utcnow()
110		result = db.interface_details.insert_one({
111		"_id": intf_id,
112		"id": intf_id,
113		"inserted_at": utc_now,
114		"updated_at": utc_now,
115		"available_tags": {"vlan": available_tags},
116		"tag_ranges": {"vlan": DEFAULT_TAG_RANGES},
117		})
118		if result:
119		evc_intf_count += 1
120
121		print(f"{intf_count} documents modified. {evc_intf_count} documents inserted")
122
123		def aggregate_outdated_interfaces(mongo: Mongo):
124		"""Aggregate outdated inteface details"""
125		db = mongo.client[mongo.db_name]
126		document_ids = set()
127		result = db.interface_details.aggregate(
128		[
129		{"$sort": {"_id": 1}},
130		{"$project": {
131		"_id": 0,
132		"id": 1,
133		"max_number": {"$max": "$available_vlans"},
134		"min_number": {"$min": "$available_vlans"},
135		"available_vlans": 1,
136		}}
137		]
138		)
139
140		messages = ""
141		for document in result:
142		document_ids.add(document["id"])
143		if document.get("available_vlans") is None:
144		continue
145		document.pop("available_vlans")
146		messages += str(document) + "\n"
		0 ignored issues – show Comprehensibility Best Practice introduced 2023-11-28 16:53 UTC by Report Bug Copy Issue Report The variable `str` does not seem to be defined. Loading history...
147
148		if messages != "":
149		print("Here are the outdated interfaces. 'available_vlans' have a massive"
150		" amount of items, minimum and maximum items will be shown only")
151		print(messages)
152
153		evc_documents = db.evcs.find({"archived": False})
154		evc_intf = defaultdict(str)
155		evc_tags = defaultdict(set)
156
157		for evc in evc_documents:
158		tag_a = evc["uni_a"].get("tag")
159		if tag_a:
160		intf_id = evc["uni_a"]["interface_id"]
161		if tag_a["value"] in evc_tags[intf_id] and isinstance(tag_a["value"], int):
162		print(f"WARNING: Detected duplicated {tag_a['value']} TAG"
163		f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_a['value'])]}"
164		f" in interface {intf_id}")
165		print()
166		evc_tags[intf_id].add(tag_a["value"])
167		evc_intf[intf_id+str(tag_a["value"])] = evc["id"]
168
169		tag_z = evc["uni_z"].get("tag")
170		if tag_z:
171		intf_id = evc["uni_z"]["interface_id"]
172		if tag_z["value"] in evc_tags[intf_id] and isinstance(tag_z["value"], int):
173		print(f"WARNING: Detected duplicated {tag_z['value']} TAG"
174		f" in EVCs {evc['id']} and {evc_intf[intf_id+str(tag_z['value'])]}"
175		f" in interface {intf_id}")
176		print()
177		evc_tags[intf_id].add(tag_z["value"])
178		evc_intf[intf_id+str(tag_z["value"])] = evc["id"]
179
180		for id_ in document_ids:
181		evc_tags.pop(id_, None)
182
183		if evc_tags:
184		print("New documents are going to be created. From the next interfaces,"
185		" these tags should be avoided")
186
187		for intf, avoid_tags in evc_tags.items():
188		if intf in document_ids:
189		continue
190		aux = {"id": intf, "avoid_tags": avoid_tags}
191		print(aux)
192
193		if not evc_tags and messages == "":
194		print("There is nothing to update or add")
195
196
197	View Code Duplication	if __name__ == "__main__":
		0 ignored issues – show Duplication introduced 2023-11-28 16:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
198		mongo = Mongo()
199		cmds = {
200		"aggregate_outdated_interfaces": aggregate_outdated_interfaces,
201		"update_database": update_database,
202		}
203		try:
204		cmd = os.environ["CMD"]
205		cmds[cmd](mongo)
206		except KeyError:
207		print(
208		f"Please set the 'CMD' env var. \nIt has to be one of these: {list(cmds.keys())}"
209		)
210		sys.exit(1)
211

kytos-ng / topology

Push — master ( 5b207c...70402e )

vlan_pool.update_database() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like