build.utils.compare_uni_out_trace() - Code Metrics - Inspection of "Redid handling of link down to ensure tear down of..." - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#583)

unknown

created 2025-02-04 18:32 UTC

build.utils.compare_uni_out_trace() A

↳ Parent: build.utils

Complexity

Conditions

Size

Total Lines	15
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	4

Importance

Changes

Metric	Value
cc	4
eloc	12
nop	3
dl	0
loc	15
ccs	7
cts	7
cp	1
crap	4
rs	9.8
c	0
b	0
f	0

"""Utility functions."""
from typing import Union

import httpx

from kytos.core.common import EntityStatus
from kytos.core.events import KytosEvent
from kytos.core.interface import UNI, Interface, TAGRange
from napps.kytos.mef_eline import settings
from napps.kytos.mef_eline.exceptions import DisabledSwitch, FlowModException


def map_evc_event_content(evc, **kwargs) -> dict:
    """Returns a set of values from evc to be used for content"""
    return kwargs | {"evc_id": evc.id,
                     "id": evc.id,
                     "name": evc.name,
                     "metadata": evc.metadata,
                     "active": evc._active,
                     "enabled": evc._enabled,
                     "uni_a": evc.uni_a.as_dict(),
                     "uni_z": evc.uni_z.as_dict()}


def emit_event(controller, name, context="kytos/mef_eline", content=None,
               timeout=None):
    """Send an event when something happens with an EVC."""
    event_name = f"{context}.{name}"
    event = KytosEvent(name=event_name, content=content)
    controller.buffers.app.put(event, timeout=timeout)


def merge_flow_dicts(
    dst: dict[str, list], *srcs: dict[str, list]
) -> dict[str, list]:
    """Merge srcs dict flows into dst."""
    for src in srcs:
        for k, v in src.items():
            if k not in dst:
                dst[k] = v
            else:
                dst[k].extend(v)
    return dst


async def aemit_event(controller, name, content):
    """Send an asynchronous event"""
    event = KytosEvent(name=name, content=content)
    await controller.buffers.app.aput(event)


def compare_endpoint_trace(endpoint, vlan, trace):
    """Compare and endpoint with a trace step."""
    if vlan and "vlan" in trace:
        return (
            endpoint.switch.dpid == trace["dpid"]
            and endpoint.port_number == trace["port"]
            and vlan == trace["vlan"]
        )
    return (
        endpoint.switch.dpid == trace["dpid"]
        and endpoint.port_number == trace["port"]
    )


def map_dl_vlan(value: Union[str, int]) -> bool:
    """Map dl_vlan value with the following criteria:
    dl_vlan = untagged or 0 -> None
    dl_vlan = any or "4096/4096" -> 1
    dl_vlan = "num1/num2" -> int in [1, 4095]"""
    special_untagged = {"untagged", 0}
    if value in special_untagged:
        return None
    special_any = {"any": 1, "4096/4096": 1}
    value = special_any.get(value, value)
    if isinstance(value, int):
        return value
    value, mask = map(int, value.split('/'))
    return value & (mask & 4095)


def compare_uni_out_trace(
    tag_value: Union[None, int, str],
    interface: Interface,
    trace: dict
) -> bool:
    """Check if the trace last step (output) matches the UNI attributes."""
    # keep compatibility for old versions of sdntrace-cp
    if "out" not in trace:
        return True
    if not isinstance(trace["out"], dict):
        return False
    uni_vlan = map_dl_vlan(tag_value) if tag_value else None
    return (
        interface.port_number == trace["out"].get("port")
        and uni_vlan == trace["out"].get("vlan")
    )


def max_power2_divisor(number: int, limit: int = 4096) -> int:
    """Get the max power of 2 that is divisor of number"""
    while number % limit > 0:
        limit //= 2
    return limit


def get_vlan_tags_and_masks(tag_ranges: list[list[int]]) -> list[int, str]:
    """Get a list of vlan/mask pairs for a given list of ranges."""
    masks_list = []
    for start, end in tag_ranges:
        limit = end + 1
        while start < limit:
            divisor = max_power2_divisor(start)
            while divisor > limit - start:
                divisor //= 2
            mask = 4096 - divisor
            if mask == 4095:
                masks_list.append(start)
            else:
                masks_list.append(f"{start}/{mask}")
            start += divisor
    return masks_list


def check_disabled_component(uni_a: UNI, uni_z: UNI):
    """Check if a switch or an interface is disabled"""
    if uni_a.interface.switch != uni_z.interface.switch:
        return
    if uni_a.interface.switch.status == EntityStatus.DISABLED:
        dpid = uni_a.interface.switch.dpid
        raise DisabledSwitch(f"Switch {dpid} is disabled")
    if uni_a.interface.status == EntityStatus.DISABLED:
        id_ = uni_a.interface.id
        raise DisabledSwitch(f"Interface {id_} is disabled")
    if uni_z.interface.status == EntityStatus.DISABLED:
        id_ = uni_z.interface.id
        raise DisabledSwitch(f"Interface {id_} is disabled")


def make_uni_list(list_circuits: list) -> list:
    """Make uni list to be sent to sdntrace"""
    uni_list = []
    for circuit in list_circuits:
        if isinstance(circuit.uni_a.user_tag, TAGRange):
            # TAGRange value from uni_a and uni_z are currently mirrored
            mask_list = (circuit.uni_a.user_tag.mask_list or
                         circuit.uni_z.user_tag.mask_list)
            for mask in mask_list:
                uni_list.append((circuit.uni_a.interface, mask))
                uni_list.append((circuit.uni_z.interface, mask))
        else:
            tag_a = None
            if circuit.uni_a.user_tag:
                tag_a = circuit.uni_a.user_tag.value
            uni_list.append(
                (circuit.uni_a.interface, tag_a)
            )
            tag_z = None
            if circuit.uni_z.user_tag:
                tag_z = circuit.uni_z.user_tag.value
            uni_list.append(
                (circuit.uni_z.interface, tag_z)
            )
    return uni_list


def send_flow_mods_event(
    controller, flow_dict: dict[str, list], action: str, force=True
):
    """Send flow mods to be deleted or install to flow_manager
     through an event"""
    for dpid, flows in flow_dict.items():
        emit_event(
            controller,
            context="kytos.flow_manager",
            name=f"flows.{action}",
            content={
                "dpid": dpid,
                "flow_dict": {"flows": flows},
                "force": force,
            },
        )


def send_flow_mods_http(
    flow_dict: dict[str, list],
    action: str, force=True
):
    """
    Send a flow_mod list to a specific switch.

    Args:
        dpid(str): The target of flows (i.e. Switch.id).
        flow_mods(dict): Python dictionary with flow_mods.
        command(str): By default is 'flows'. To remove a flow is 'remove'.
        force(bool): True to send via consistency check in case of errors.
        by_switch(bool): True to send to 'flows_by_switch' request instead.
    """
    endpoint = f"{settings.MANAGER_URL}/flows_by_switch/?force={force}"

    formatted_dict = {
        dpid: {"flows": flows}
        for (dpid, flows) in flow_dict.items()
    }

    try:
        if action == "install":
            res = httpx.post(endpoint, json=formatted_dict, timeout=30)
        elif action == "delete":
            res = httpx.request(
                "DELETE", endpoint, json=formatted_dict, timeout=30
            )
    except httpx.RequestError as err:
        raise FlowModException(str(err)) from err
    if res.is_server_error or res.status_code >= 400:

        raise FlowModException(res.text)


def prepare_delete_flow(evc_flows: dict[str, list[dict]]):
    """Create flow mods suited for flow deletion."""
    dpid_flows: dict[str, list[dict]] = {}

    if not evc_flows:
        return dpid_flows

    for dpid, flows in evc_flows.items():
        dpid_flows.setdefault(dpid, [])
        for flow in flows:
            dpid_flows[dpid].append({
                "cookie": flow["cookie"],
                "match": flow["match"],
                "owner": "mef_eline",
                "cookie_mask": int(0xffffffffffffffff)
            })
    return dpid_flows


1		"""Utility functions."""
2	1	from typing import Union
3
4	1	import httpx
5
6	1	from kytos.core.common import EntityStatus
7	1	from kytos.core.events import KytosEvent
8	1	from kytos.core.interface import UNI, Interface, TAGRange
9	1	from napps.kytos.mef_eline import settings
10	1	from napps.kytos.mef_eline.exceptions import DisabledSwitch, FlowModException
11
12
13	1	def map_evc_event_content(evc, **kwargs) -> dict:
14		"""Returns a set of values from evc to be used for content"""
15	1	return kwargs \| {"evc_id": evc.id,
16		"id": evc.id,
17		"name": evc.name,
18		"metadata": evc.metadata,
19		"active": evc._active,
20		"enabled": evc._enabled,
21		"uni_a": evc.uni_a.as_dict(),
22		"uni_z": evc.uni_z.as_dict()}
23
24
25	1	def emit_event(controller, name, context="kytos/mef_eline", content=None,
26		timeout=None):
27		"""Send an event when something happens with an EVC."""
28	1	event_name = f"{context}.{name}"
29	1	event = KytosEvent(name=event_name, content=content)
30	1	controller.buffers.app.put(event, timeout=timeout)
31
32
33	1	def merge_flow_dicts(
34		dst: dict[str, list], *srcs: dict[str, list]
35		) -> dict[str, list]:
36		"""Merge srcs dict flows into dst."""
37	1	for src in srcs:
38	1	for k, v in src.items():
39	1	if k not in dst:
40	1	dst[k] = v
41		else:
42	1	dst[k].extend(v)
43	1	return dst
44
45
46	1	async def aemit_event(controller, name, content):
47		"""Send an asynchronous event"""
48	1	event = KytosEvent(name=name, content=content)
49	1	await controller.buffers.app.aput(event)
50
51
52	1	def compare_endpoint_trace(endpoint, vlan, trace):
53		"""Compare and endpoint with a trace step."""
54	1	if vlan and "vlan" in trace:
55	1	return (
56		endpoint.switch.dpid == trace["dpid"]
57		and endpoint.port_number == trace["port"]
58		and vlan == trace["vlan"]
59		)
60	1	return (
61		endpoint.switch.dpid == trace["dpid"]
62		and endpoint.port_number == trace["port"]
63		)
64
65
66	1	def map_dl_vlan(value: Union[str, int]) -> bool:
67		"""Map dl_vlan value with the following criteria:
68		dl_vlan = untagged or 0 -> None
69		dl_vlan = any or "4096/4096" -> 1
70		dl_vlan = "num1/num2" -> int in [1, 4095]"""
71	1	special_untagged = {"untagged", 0}
72	1	if value in special_untagged:
73	1	return None
74	1	special_any = {"any": 1, "4096/4096": 1}
75	1	value = special_any.get(value, value)
76	1	if isinstance(value, int):
77	1	return value
78	1	value, mask = map(int, value.split('/'))
79	1	return value & (mask & 4095)
80
81
82	1	def compare_uni_out_trace(
83		tag_value: Union[None, int, str],
84		interface: Interface,
85		trace: dict
86		) -> bool:
87		"""Check if the trace last step (output) matches the UNI attributes."""
88		# keep compatibility for old versions of sdntrace-cp
89	1	if "out" not in trace:
90	1	return True
91	1	if not isinstance(trace["out"], dict):
92	1	return False
93	1	uni_vlan = map_dl_vlan(tag_value) if tag_value else None
94	1	return (
95		interface.port_number == trace["out"].get("port")
96		and uni_vlan == trace["out"].get("vlan")
97		)
98
99
100	1	def max_power2_divisor(number: int, limit: int = 4096) -> int:
101		"""Get the max power of 2 that is divisor of number"""
102	1	while number % limit > 0:
103	1	limit //= 2
104	1	return limit
105
106
107	1	def get_vlan_tags_and_masks(tag_ranges: list[list[int]]) -> list[int, str]:
108		"""Get a list of vlan/mask pairs for a given list of ranges."""
109	1	masks_list = []
110	1	for start, end in tag_ranges:
111	1	limit = end + 1
112	1	while start < limit:
113	1	divisor = max_power2_divisor(start)
114	1	while divisor > limit - start:
115	1	divisor //= 2
116	1	mask = 4096 - divisor
117	1	if mask == 4095:
118	1	masks_list.append(start)
119		else:
120	1	masks_list.append(f"{start}/{mask}")
121	1	start += divisor
122	1	return masks_list
123
124
125	1	def check_disabled_component(uni_a: UNI, uni_z: UNI):
126		"""Check if a switch or an interface is disabled"""
127	1	if uni_a.interface.switch != uni_z.interface.switch:
128	1	return
129	1	if uni_a.interface.switch.status == EntityStatus.DISABLED:
130	1	dpid = uni_a.interface.switch.dpid
131	1	raise DisabledSwitch(f"Switch {dpid} is disabled")
132	1	if uni_a.interface.status == EntityStatus.DISABLED:
133	1	id_ = uni_a.interface.id
134	1	raise DisabledSwitch(f"Interface {id_} is disabled")
135	1	if uni_z.interface.status == EntityStatus.DISABLED:
136	1	id_ = uni_z.interface.id
137	1	raise DisabledSwitch(f"Interface {id_} is disabled")
138
139
140	1	def make_uni_list(list_circuits: list) -> list:
141		"""Make uni list to be sent to sdntrace"""
142	1	uni_list = []
143	1	for circuit in list_circuits:
144	1	if isinstance(circuit.uni_a.user_tag, TAGRange):
145		# TAGRange value from uni_a and uni_z are currently mirrored
146	1	mask_list = (circuit.uni_a.user_tag.mask_list or
147		circuit.uni_z.user_tag.mask_list)
148	1	for mask in mask_list:
149	1	uni_list.append((circuit.uni_a.interface, mask))
150	1	uni_list.append((circuit.uni_z.interface, mask))
151		else:
152	1	tag_a = None
153	1	if circuit.uni_a.user_tag:
154	1	tag_a = circuit.uni_a.user_tag.value
155	1	uni_list.append(
156		(circuit.uni_a.interface, tag_a)
157		)
158	1	tag_z = None
159	1	if circuit.uni_z.user_tag:
160	1	tag_z = circuit.uni_z.user_tag.value
161	1	uni_list.append(
162		(circuit.uni_z.interface, tag_z)
163		)
164	1	return uni_list
165
166
167	1	def send_flow_mods_event(
168		controller, flow_dict: dict[str, list], action: str, force=True
169		):
170		"""Send flow mods to be deleted or install to flow_manager
171		through an event"""
172		for dpid, flows in flow_dict.items():
173		emit_event(
174		controller,
175		context="kytos.flow_manager",
176		name=f"flows.{action}",
177		content={
178		"dpid": dpid,
179		"flow_dict": {"flows": flows},
180		"force": force,
181		},
182		)
183
184
185	1	def send_flow_mods_http(
186		flow_dict: dict[str, list],
187		action: str, force=True
188		):
189		"""
190		Send a flow_mod list to a specific switch.
191
192		Args:
193		dpid(str): The target of flows (i.e. Switch.id).
194		flow_mods(dict): Python dictionary with flow_mods.
195		command(str): By default is 'flows'. To remove a flow is 'remove'.
196		force(bool): True to send via consistency check in case of errors.
197		by_switch(bool): True to send to 'flows_by_switch' request instead.
198		"""
199		endpoint = f"{settings.MANAGER_URL}/flows_by_switch/?force={force}"
200
201		formatted_dict = {
202		dpid: {"flows": flows}
203		for (dpid, flows) in flow_dict.items()
204		}
205
206		try:
207		if action == "install":
208		res = httpx.post(endpoint, json=formatted_dict, timeout=30)
209		elif action == "delete":
210		res = httpx.request(
211		"DELETE", endpoint, json=formatted_dict, timeout=30
212		)
213		except httpx.RequestError as err:
214		raise FlowModException(str(err)) from err
215		if res.is_server_error or res.status_code >= 400:
		0 ignored issues – show introduced 2024-12-17 14:21 UTC by Report Bug Copy Issue Report The variable `res` does not seem to be defined for all execution paths. Loading history...
216		raise FlowModException(res.text)
217
218
219	1	def prepare_delete_flow(evc_flows: dict[str, list[dict]]):
220		"""Create flow mods suited for flow deletion."""
221	1	dpid_flows: dict[str, list[dict]] = {}
222
223	1	if not evc_flows:
224		return dpid_flows
225
226	1	for dpid, flows in evc_flows.items():
227	1	dpid_flows.setdefault(dpid, [])
228	1	for flow in flows:
229	1	dpid_flows[dpid].append({
230		"cookie": flow["cookie"],
231		"match": flow["match"],
232		"owner": "mef_eline",
233		"cookie_mask": int(0xffffffffffffffff)
234		})
235		return dpid_flows
236

kytos-ng / mef_eline

Pull Request — master (#583)

build.utils.compare_uni_out_trace() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like