build.managers.deployer.MaintenanceDeployer._get_affected_ids() - Code Metrics - Inspection of "Add service interruptions to maintenance windows" - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#78)

unknown

created 2023-06-22 15:43 UTC

MaintenanceDeployer._get_affected_ids() C

↳ Parent: build.managers.deployer

Complexity

Conditions

Size

Total Lines	77
Code Lines	49

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	10

Importance

Changes

Metric	Value
cc	10
eloc	49
nop	2
dl	0
loc	77
ccs	13
cts	13
cp	1
crap	10
rs	5.869
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.


from collections import Counter
from dataclasses import dataclass
from itertools import chain

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

    @classmethod
    def new_deployer(cls, controller: Controller):
        """
        Creates a new MaintenanceDeployer from the given Kytos Controller
        """
        instance = cls(controller, Counter(), Counter(), Counter())
        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(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(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.napps[('kytos', 'topology')].links.get,
                window.links
            )
        )

        tot_links = list(chain(implicit_links, explicit_links))

        affected_switch_ids = list(set(map(
            lambda switch: switch.id,
            filter(
                self.switch_not_in_maintenance,
                tot_switches
            )
        )))

        affected_interface_ids = list(set(map(
            lambda interface: interface.id,
            filter(
                self.interface_not_in_maintenance,
                tot_interfaces
            )
        )))

        affected_link_ids = list(set(map(
            lambda link: link.id,
            filter(
                self.link_not_in_maintenance,
                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."""
        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."""

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

kytos-ng / maintenance

Pull Request — master (#78)

MaintenanceDeployer._get_affected_ids() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like