build.managers.deployer.MaintenanceDeployer._get_affected_ids() - Code Metrics - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

MaintenanceDeployer._get_affected_ids() C
last analyzed 2025-06-23 13:22 UTC

↳ Parent: build.managers.deployer

Complexity

Conditions

Size

Total Lines	86
Code Lines	54

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	10

Importance

Changes

Metric	Value
cc	10
eloc	54
nop	2
dl	0
loc	86
ccs	13
cts	13
cp	1
crap	10
rs	5.7054
c	0
b	0
f	0

How to fix Long Method Complexity

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like build.managers.deployer.MaintenanceDeployer._get_affected_ids() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

"""Module for handling the deployment of maintenance windows."""
from collections import Counter
from dataclasses import dataclass
from itertools import chain
from threading import Lock

from kytos.core.common import EntityStatus
from kytos.core.controller import Controller
from kytos.core.events import KytosEvent
from kytos.core.switch import Switch
from kytos.core.interface import Interface
from kytos.core.link import Link

from ..models import MaintenanceWindow


@dataclass
class MaintenanceDeployer:
    """Class for deploying maintenances"""
    controller: Controller
    maintenance_switches: Counter
    maintenance_interfaces: Counter
    maintenance_links: Counter
    lock: Lock

    @classmethod
    def new_deployer(cls, controller: Controller):
        """
        Creates a new MaintenanceDeployer from the given Kytos Controller
        """
        instance = cls(controller, Counter(), Counter(), Counter(), Lock())
        Switch.register_status_func(
            'maintenance_status',
            instance.switch_status_func
        )
        Switch.register_status_reason_func(
            'maintenance_status',
            instance.switch_status_reason_func
        )
        Interface.register_status_func(
            'maintenance_status',
            instance.interface_status_func
        )
        Interface.register_status_reason_func(
            'maintenance_status',
            instance.interface_status_reason_func
        )
        Link.register_status_func(
            'maintenance_status',
            instance.link_status_func
        )
        Link.register_status_reason_func(
            'maintenance_status',
            instance.link_status_reason_func
        )

        return instance

    def _maintenance_event(self, window_devices: dict, operation: str):
        """Create events to start/end a maintenance."""
        event = KytosEvent(
            f'topology.interruption.{operation}',
            content={
                'type': 'maintenance',
                **window_devices
            }
        )
        self.controller.buffers.app.put(event)

    def _get_affected_ids(
        self,
        window: MaintenanceWindow
    ) -> dict[str, list[str]]:
        explicit_switches = filter(
            lambda switch: switch is not None,
            map(
                self.controller.switches.get,
                window.switches
            )
        )

        tot_switches = list(filter(
            self.switch_not_in_maintenance,
            explicit_switches
        ))

        implicit_interfaces = chain.from_iterable(
            map(
                lambda switch: switch.interfaces.values(),
                tot_switches
            )
        )

        explicit_interfaces = filter(
            lambda interface: interface is not None,
            map(
                self.controller.get_interface_by_id,
                window.interfaces
            )
        )

        tot_interfaces = list(
            filter(
                self.interface_not_in_maintenance,
                chain(implicit_interfaces, explicit_interfaces)
            )
        )

        implicit_links = filter(
            lambda link: link is not None,
            map(
                lambda interface: interface.link,
                tot_interfaces
            )
        )

        explicit_links = filter(
            lambda link: link is not None,
            map(
                self.controller.links.get,
                window.links
            )
        )

        tot_links = list(
            filter(
                self.link_not_in_maintenance,
                chain(implicit_links, explicit_links)
            )
        )

        affected_switch_ids = frozenset(
            map(
                lambda switch: switch.id,
                tot_switches
            )
        )

        affected_interface_ids = frozenset(
            map(
                lambda interface: interface.id,
                tot_interfaces
            )
        )

        affected_link_ids = frozenset(
            map(
                lambda link: link.id,
                tot_links
            )
        )
        return {
            'switches': affected_switch_ids,
            'interfaces': affected_interface_ids,
            'links': affected_link_ids,
        }

    def start_mw(self, window: MaintenanceWindow):
        """Actions taken when a maintenance window starts."""
        with self.lock:
            affected_ids = self._get_affected_ids(window)

            self.maintenance_switches.update(window.switches)
            self.maintenance_interfaces.update(window.interfaces)
            self.maintenance_links.update(window.links)

            self._maintenance_event(
                affected_ids,
                'start'
            )

    def end_mw(self, window: MaintenanceWindow):
        """Actions taken when a maintenance window finishes."""
        with self.lock:
            self.maintenance_switches.subtract(window.switches)
            self.maintenance_interfaces.subtract(window.interfaces)
            self.maintenance_links.subtract(window.links)

            affected_ids = self._get_affected_ids(window)

            self._maintenance_event(
                affected_ids,
                'end'
            )

    def switch_not_in_maintenance(self, dev: Switch) -> bool:
        """Checks if a switch is not undergoing maintenance"""
        return not self.maintenance_switches[dev.id]

    def interface_not_in_maintenance(self, dev: Interface) -> bool:
        """Checks if a interface is not undergoing maintenance"""
        return (
            not self.maintenance_interfaces[dev.id] and
            self.switch_not_in_maintenance(dev.switch)
        )

    def link_not_in_maintenance(self, dev: Link) -> bool:
        """Checks if a link is not undergoing maintenance"""
        return (
            not self.maintenance_links[dev.id] and
            all(
                map(
                    self.interface_not_in_maintenance,
                    (dev.endpoint_a, dev.endpoint_b)
                )
            )
        )

    def switch_status_func(self, dev: Switch) -> EntityStatus:
        """Checks if a given device is undergoing maintenance"""
        if self.switch_not_in_maintenance(dev):
            return EntityStatus.UP
        return EntityStatus.DOWN

    def switch_status_reason_func(self, dev: Switch) -> frozenset:
        """Checks if a given device is undergoing maintenance"""
        if self.switch_not_in_maintenance(dev):
            return frozenset()
        return frozenset({'maintenance'})

    def interface_status_func(self, dev: Interface) -> EntityStatus:
        """Checks if a given device is undergoing maintenance"""
        if self.interface_not_in_maintenance(dev):
            return EntityStatus.UP
        return EntityStatus.DOWN

    def interface_status_reason_func(self, dev: Interface) -> frozenset:
        """Checks if a given device is undergoing maintenance"""
        if self.interface_not_in_maintenance(dev):
            return frozenset()
        return frozenset({'maintenance'})

    def link_status_func(self, dev: Link) -> EntityStatus:
        """Checks if a given device is undergoing maintenance"""
        if self.link_not_in_maintenance(dev):
            return EntityStatus.UP
        return EntityStatus.DOWN

    def link_status_reason_func(self, dev: Link) -> frozenset:
        """Checks if a given device is undergoing maintenance"""
        if self.link_not_in_maintenance(dev):
            return frozenset()
        return frozenset({'maintenance'})

1		"""Module for handling the deployment of maintenance windows."""
2	1	from collections import Counter
3	1	from dataclasses import dataclass
4	1	from itertools import chain
5	1	from threading import Lock
6
7	1	from kytos.core.common import EntityStatus
8	1	from kytos.core.controller import Controller
9	1	from kytos.core.events import KytosEvent
10	1	from kytos.core.switch import Switch
11	1	from kytos.core.interface import Interface
12	1	from kytos.core.link import Link
13
14	1	from ..models import MaintenanceWindow
15
16
17	1	@dataclass
18	1	class MaintenanceDeployer:
19		"""Class for deploying maintenances"""
20	1	controller: Controller
21	1	maintenance_switches: Counter
22	1	maintenance_interfaces: Counter
23	1	maintenance_links: Counter
24	1	lock: Lock
25
26	1	@classmethod
27	1	def new_deployer(cls, controller: Controller):
28		"""
29		Creates a new MaintenanceDeployer from the given Kytos Controller
30		"""
31	1	instance = cls(controller, Counter(), Counter(), Counter(), Lock())
32	1	Switch.register_status_func(
33		'maintenance_status',
34		instance.switch_status_func
35		)
36	1	Switch.register_status_reason_func(
37		'maintenance_status',
38		instance.switch_status_reason_func
39		)
40	1	Interface.register_status_func(
41		'maintenance_status',
42		instance.interface_status_func
43		)
44	1	Interface.register_status_reason_func(
45		'maintenance_status',
46		instance.interface_status_reason_func
47		)
48	1	Link.register_status_func(
49		'maintenance_status',
50		instance.link_status_func
51		)
52	1	Link.register_status_reason_func(
53		'maintenance_status',
54		instance.link_status_reason_func
55		)
56
57	1	return instance
58
59	1	def _maintenance_event(self, window_devices: dict, operation: str):
60		"""Create events to start/end a maintenance."""
61	1	event = KytosEvent(
62		f'topology.interruption.{operation}',
63		content={
64		'type': 'maintenance',
65		**window_devices
66		}
67		)
68	1	self.controller.buffers.app.put(event)
69
70	1	def _get_affected_ids(
71		self,
72		window: MaintenanceWindow
73		) -> dict[str, list[str]]:
74	1	explicit_switches = filter(
75		lambda switch: switch is not None,
76		map(
77		self.controller.switches.get,
78		window.switches
79		)
80		)
81
82	1	tot_switches = list(filter(
83		self.switch_not_in_maintenance,
84		explicit_switches
85		))
86
87	1	implicit_interfaces = chain.from_iterable(
88		map(
89		lambda switch: switch.interfaces.values(),
90		tot_switches
91		)
92		)
93
94	1	explicit_interfaces = filter(
95		lambda interface: interface is not None,
96		map(
97		self.controller.get_interface_by_id,
98		window.interfaces
99		)
100		)
101
102	1	tot_interfaces = list(
103		filter(
104		self.interface_not_in_maintenance,
105		chain(implicit_interfaces, explicit_interfaces)
106		)
107		)
108
109	1	implicit_links = filter(
110		lambda link: link is not None,
111		map(
112		lambda interface: interface.link,
113		tot_interfaces
114		)
115		)
116
117	1	explicit_links = filter(
118		lambda link: link is not None,
119		map(
120		self.controller.links.get,
121		window.links
122		)
123		)
124
125	1	tot_links = list(
126		filter(
127		self.link_not_in_maintenance,
128		chain(implicit_links, explicit_links)
129		)
130		)
131
132	1	affected_switch_ids = frozenset(
133		map(
134		lambda switch: switch.id,
135		tot_switches
136		)
137		)
138
139	1	affected_interface_ids = frozenset(
140		map(
141		lambda interface: interface.id,
142		tot_interfaces
143		)
144		)
145
146	1	affected_link_ids = frozenset(
147		map(
148		lambda link: link.id,
149		tot_links
150		)
151		)
152	1	return {
153		'switches': affected_switch_ids,
154		'interfaces': affected_interface_ids,
155		'links': affected_link_ids,
156		}
157
158	1	def start_mw(self, window: MaintenanceWindow):
159		"""Actions taken when a maintenance window starts."""
160	1	with self.lock:
161	1	affected_ids = self._get_affected_ids(window)
162
163	1	self.maintenance_switches.update(window.switches)
164	1	self.maintenance_interfaces.update(window.interfaces)
165	1	self.maintenance_links.update(window.links)
166
167	1	self._maintenance_event(
168		affected_ids,
169		'start'
170		)
171
172	1	def end_mw(self, window: MaintenanceWindow):
173		"""Actions taken when a maintenance window finishes."""
174	1	with self.lock:
175	1	self.maintenance_switches.subtract(window.switches)
176	1	self.maintenance_interfaces.subtract(window.interfaces)
177	1	self.maintenance_links.subtract(window.links)
178
179	1	affected_ids = self._get_affected_ids(window)
180
181	1	self._maintenance_event(
182		affected_ids,
183		'end'
184		)
185
186	1	def switch_not_in_maintenance(self, dev: Switch) -> bool:
187		"""Checks if a switch is not undergoing maintenance"""
188	1	return not self.maintenance_switches[dev.id]
189
190	1	def interface_not_in_maintenance(self, dev: Interface) -> bool:
191		"""Checks if a interface is not undergoing maintenance"""
192	1	return (
193		not self.maintenance_interfaces[dev.id] and
194		self.switch_not_in_maintenance(dev.switch)
195		)
196
197	1	def link_not_in_maintenance(self, dev: Link) -> bool:
198		"""Checks if a link is not undergoing maintenance"""
199	1	return (
200		not self.maintenance_links[dev.id] and
201		all(
202		map(
203		self.interface_not_in_maintenance,
204		(dev.endpoint_a, dev.endpoint_b)
205		)
206		)
207		)
208
209	1	def switch_status_func(self, dev: Switch) -> EntityStatus:
210		"""Checks if a given device is undergoing maintenance"""
211	1	if self.switch_not_in_maintenance(dev):
212	1	return EntityStatus.UP
213	1	return EntityStatus.DOWN
214
215	1	def switch_status_reason_func(self, dev: Switch) -> frozenset:
216		"""Checks if a given device is undergoing maintenance"""
217	1	if self.switch_not_in_maintenance(dev):
218	1	return frozenset()
219	1	return frozenset({'maintenance'})
220
221	1	def interface_status_func(self, dev: Interface) -> EntityStatus:
222		"""Checks if a given device is undergoing maintenance"""
223	1	if self.interface_not_in_maintenance(dev):
224	1	return EntityStatus.UP
225	1	return EntityStatus.DOWN
226
227	1	def interface_status_reason_func(self, dev: Interface) -> frozenset:
228		"""Checks if a given device is undergoing maintenance"""
229	1	if self.interface_not_in_maintenance(dev):
230	1	return frozenset()
231	1	return frozenset({'maintenance'})
232
233	1	def link_status_func(self, dev: Link) -> EntityStatus:
234		"""Checks if a given device is undergoing maintenance"""
235	1	if self.link_not_in_maintenance(dev):
236	1	return EntityStatus.UP
237	1	return EntityStatus.DOWN
238
239	1	def link_status_reason_func(self, dev: Link) -> frozenset:
240		"""Checks if a given device is undergoing maintenance"""
241	1	if self.link_not_in_maintenance(dev):
242		return frozenset()
243		return frozenset({'maintenance'})

kytos-ng / maintenance

MaintenanceDeployer._get_affected_ids() C last analyzed 2025-06-23 13:22 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

MaintenanceDeployer._get_affected_ids() C
last analyzed 2025-06-23 13:22 UTC