build.models.MaintenanceWindow.str_to_datetime() - Code Metrics - Inspection of "Updated tox.ini and removed `tests_require` and `s..." - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#53)

by Vinicius

created 2022-06-15 14:21 UTC

build.models.MaintenanceWindow.str_to_datetime() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	5
ccs	4
cts	4
cp	1
rs	10
c	0
b	0
f	0
cc	1
nop	1
crap	1

"""Models used by the maintenance NApp.

This module define models for the maintenance window itself and the
scheduler.
"""
import datetime
from enum import IntEnum
from uuid import uuid4

import pytz
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.background import BackgroundScheduler

from kytos.core import KytosEvent, log
from kytos.core.interface import TAG, UNI
from kytos.core.link import Link

TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"


class Status(IntEnum):
    """Maintenance windows status."""

    PENDING = 0
    RUNNING = 1
    FINISHED = 2


class MaintenanceWindow:
    """Class to store a maintenance window."""

    def __init__(self, start, end, controller, **kwargs):
        """Create an instance of MaintenanceWindow.

        Args:
            start(datetime): when the maintenance will begin
            end(datetime): when the maintenance will finish
            items: list of items that will be maintained;
                each item can be either a switch, a link or a client interface
        """
        # pylint: disable=invalid-name
        self.controller = controller
        items = kwargs.get('items')
        if items is None:
            items = []
        mw_id = kwargs.get('mw_id')
        self.id = mw_id if mw_id else uuid4().hex
        self.description = kwargs.get('description')
        self.start = start
        self.end = end
        self._switches = []
        self._links = []
        self._unis = []
        self.items = items
        self.status = kwargs.get('status', Status.PENDING)

    @property
    def items(self):
        """Items getter."""
        return self._switches + self._links + self._unis

    @items.setter
    def items(self, items):
        """Items setter."""
        self._switches = []
        self._unis = []
        self._links = []
        for i in items:
            if isinstance(i, UNI):
                self._unis.append(i)
            elif isinstance(i, Link):
                self._links.append(i)
            else:
                self._switches.append(i)

    def as_dict(self):
        """Return this maintenance window as a dictionary."""
        mw_dict = {}
        mw_dict['id'] = self.id
        mw_dict['description'] = self.description if self.description else ''
        mw_dict['status'] = self.status
        mw_dict['start'] = self.start.strftime(TIME_FMT)
        mw_dict['end'] = self.end.strftime(TIME_FMT)
        mw_dict['items'] = []
        for i in self.items:
            try:
                mw_dict['items'].append(i.as_dict())
            except (AttributeError, TypeError):
                mw_dict['items'].append(i)
        return mw_dict

    @classmethod
    def from_dict(cls, mw_dict, controller):
        """Create a maintenance window from a dictionary of attributes."""
        mw_id = mw_dict.get('id')

        start = cls.str_to_datetime(mw_dict['start'])
        end = cls.str_to_datetime(mw_dict['end'])
        try:
            items = mw_dict['items']
        except KeyError:
            raise ValueError('At least one item must be provided')
        if not items:
            raise ValueError('At least one item must be provided')
        description = mw_dict.get('description')
        status = mw_dict.get('status', Status.PENDING)
        return cls(start, end, controller, items=items, mw_id=mw_id,
                   description=description, status=status)

    def update(self, mw_dict):
        """Update a maintenance window with the data from a dictionary."""
        try:
            start = self.str_to_datetime(mw_dict['start'])
        except KeyError:
            start = self.start
        try:
            end = self.str_to_datetime(mw_dict['end'])
        except KeyError:
            end = self.end
        now = datetime.datetime.now(pytz.utc)
        if start < now:
            raise ValueError('Start in the past not allowed.')
        if end < start:
            raise ValueError('End before start not allowed.')
        if 'items' in mw_dict:
            if not mw_dict['items']:
                raise ValueError('At least one item must be provided')
            self.items = mw_dict['items']
        self.start = start
        self.end = end
        if 'description' in mw_dict:
            self.description = mw_dict['description']

    @staticmethod
    def intf_from_dict(intf_id, controller):
        """Get the Interface instance with intf_id."""
        intf = controller.get_interface_by_id(intf_id)
        return intf

    @staticmethod
    def uni_from_dict(uni_dict, controller):
        """Create UNI instance from a dictionary."""
        intf = MaintenanceWindow.intf_from_dict(uni_dict['interface_id'],
                                                controller)
        tag = TAG.from_dict(uni_dict['tag'])
        if intf and tag:
            return UNI(intf, tag)
        return None

    @staticmethod
    def link_from_dict(link_dict, controller):
        """Create a link instance from a dictionary."""
        endpoint_a = controller.get_interface_by_id(
            link_dict['endpoint_a']['id'])
        endpoint_b = controller.get_interface_by_id(
            link_dict['endpoint_b']['id'])

        link = Link(endpoint_a, endpoint_b)
        if 'metadata' in link_dict:
            link.extend_metadata(link_dict['metadata'])
        s_vlan = link.get_metadata('s_vlan')
        if s_vlan:
            tag = TAG.from_dict(s_vlan)
            link.update_metadata('s_vlan', tag)
        return link

    @staticmethod
    def str_to_datetime(str_date):
        """Convert a string representing a date and time to datetime."""
        date = datetime.datetime.strptime(str_date, TIME_FMT)
        return date.astimezone(pytz.utc)

    def maintenance_event(self, operation):
        """Create events to start/end a maintenance."""
        if self._switches:
            switches = []
            for dpid in self._switches:
                switch = self.controller.switches.get(dpid, None)
                if switch:
                    switches.append(switch)
            event = KytosEvent(name=f'kytos/maintenance.{operation}_switch',
                               content={'switches': switches})
            self.controller.buffers.app.put(event)
        if self._unis:
            event = KytosEvent(name=f'kytos/maintenance.{operation}_uni',
                               content={'unis': self._unis})
            self.controller.buffers.app.put(event)
        if self._links:
            event = KytosEvent(name=f'kytos/maintenance.{operation}_link',
                               content={'links': self._links})
            self.controller.buffers.app.put(event)

    def start_mw(self):
        """Actions taken when a maintenance window starts."""
        self.status = Status.RUNNING
        self.maintenance_event('start')

    def end_mw(self):
        """Actions taken when a maintenance window finishes."""
        self.status = Status.FINISHED
        self.maintenance_event('end')


class Scheduler:
    """Scheduler for a maintenance window."""

    def __init__(self):
        """Initialize a new scheduler."""
        self.scheduler = BackgroundScheduler(timezone=pytz.utc)
        self.scheduler.start()

    def add(self, maintenance):
        """Add jobs to start and end a maintenance window."""
        self.scheduler.add_job(maintenance.start_mw, 'date',
                               id=f'{maintenance.id}-start',
                               run_date=maintenance.start)
        self.scheduler.add_job(maintenance.end_mw, 'date',
                               id=f'{maintenance.id}-end',
                               run_date=maintenance.end)

    def remove(self, maintenance):
        """Remove jobs that start and end a maintenance window."""
        try:
            self.scheduler.remove_job(f'{maintenance.id}-start')
        except JobLookupError:
            log.info(f'Job to start {maintenance.id} already removed.')
        try:
            self.scheduler.remove_job(f'{maintenance.id}-end')
        except JobLookupError:
            log.info(f'Job to end {maintenance.id} already removed.')


1		"""Models used by the maintenance NApp.
2
3		This module define models for the maintenance window itself and the
4		scheduler.
5		"""
6	1	import datetime
7	1	from enum import IntEnum
8	1	from uuid import uuid4
9
10	1	import pytz
11	1	from apscheduler.jobstores.base import JobLookupError
12	1	from apscheduler.schedulers.background import BackgroundScheduler
13
14	1	from kytos.core import KytosEvent, log
15	1	from kytos.core.interface import TAG, UNI
16	1	from kytos.core.link import Link
17
18	1	TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
19
20
21	1	class Status(IntEnum):
22		"""Maintenance windows status."""
23
24	1	PENDING = 0
25	1	RUNNING = 1
26	1	FINISHED = 2
27
28
29	1	class MaintenanceWindow:
30		"""Class to store a maintenance window."""
31
32	1	def __init__(self, start, end, controller, **kwargs):
33		"""Create an instance of MaintenanceWindow.
34
35		Args:
36		start(datetime): when the maintenance will begin
37		end(datetime): when the maintenance will finish
38		items: list of items that will be maintained;
39		each item can be either a switch, a link or a client interface
40		"""
41		# pylint: disable=invalid-name
42	1	self.controller = controller
43	1	items = kwargs.get('items')
44	1	if items is None:
45		items = []
46	1	mw_id = kwargs.get('mw_id')
47	1	self.id = mw_id if mw_id else uuid4().hex
48	1	self.description = kwargs.get('description')
49	1	self.start = start
50	1	self.end = end
51	1	self._switches = []
52	1	self._links = []
53	1	self._unis = []
54	1	self.items = items
55	1	self.status = kwargs.get('status', Status.PENDING)
56
57	1	@property
58	1	def items(self):
59		"""Items getter."""
60	1	return self._switches + self._links + self._unis
61
62	1	@items.setter
63	1	def items(self, items):
64		"""Items setter."""
65	1	self._switches = []
66	1	self._unis = []
67	1	self._links = []
68	1	for i in items:
69	1	if isinstance(i, UNI):
70		self._unis.append(i)
71	1	elif isinstance(i, Link):
72		self._links.append(i)
73		else:
74	1	self._switches.append(i)
75
76	1	def as_dict(self):
77		"""Return this maintenance window as a dictionary."""
78	1	mw_dict = {}
79	1	mw_dict['id'] = self.id
80	1	mw_dict['description'] = self.description if self.description else ''
81	1	mw_dict['status'] = self.status
82	1	mw_dict['start'] = self.start.strftime(TIME_FMT)
83	1	mw_dict['end'] = self.end.strftime(TIME_FMT)
84	1	mw_dict['items'] = []
85	1	for i in self.items:
86	1	try:
87	1	mw_dict['items'].append(i.as_dict())
88	1	except (AttributeError, TypeError):
89	1	mw_dict['items'].append(i)
90	1	return mw_dict
91
92	1	@classmethod
93	1	def from_dict(cls, mw_dict, controller):
94		"""Create a maintenance window from a dictionary of attributes."""
95		mw_id = mw_dict.get('id')
96
97		start = cls.str_to_datetime(mw_dict['start'])
98		end = cls.str_to_datetime(mw_dict['end'])
99		try:
100		items = mw_dict['items']
101		except KeyError:
102		raise ValueError('At least one item must be provided')
103		if not items:
104		raise ValueError('At least one item must be provided')
105		description = mw_dict.get('description')
106		status = mw_dict.get('status', Status.PENDING)
107		return cls(start, end, controller, items=items, mw_id=mw_id,
108		description=description, status=status)
109
110	1	def update(self, mw_dict):
111		"""Update a maintenance window with the data from a dictionary."""
112	1	try:
113	1	start = self.str_to_datetime(mw_dict['start'])
114	1	except KeyError:
115	1	start = self.start
116	1	try:
117	1	end = self.str_to_datetime(mw_dict['end'])
118	1	except KeyError:
119	1	end = self.end
120	1	now = datetime.datetime.now(pytz.utc)
121	1	if start < now:
122		raise ValueError('Start in the past not allowed.')
123	1	if end < start:
124		raise ValueError('End before start not allowed.')
125	1	if 'items' in mw_dict:
126	1	if not mw_dict['items']:
127	1	raise ValueError('At least one item must be provided')
128	1	self.items = mw_dict['items']
129	1	self.start = start
130	1	self.end = end
131	1	if 'description' in mw_dict:
132		self.description = mw_dict['description']
133
134	1	@staticmethod
135	1	def intf_from_dict(intf_id, controller):
136		"""Get the Interface instance with intf_id."""
137		intf = controller.get_interface_by_id(intf_id)
138		return intf
139
140	1	@staticmethod
141	1	def uni_from_dict(uni_dict, controller):
142		"""Create UNI instance from a dictionary."""
143		intf = MaintenanceWindow.intf_from_dict(uni_dict['interface_id'],
144		controller)
145		tag = TAG.from_dict(uni_dict['tag'])
146		if intf and tag:
147		return UNI(intf, tag)
148		return None
149
150	1	@staticmethod
151	1	def link_from_dict(link_dict, controller):
152		"""Create a link instance from a dictionary."""
153		endpoint_a = controller.get_interface_by_id(
154		link_dict['endpoint_a']['id'])
155		endpoint_b = controller.get_interface_by_id(
156		link_dict['endpoint_b']['id'])
157
158		link = Link(endpoint_a, endpoint_b)
159		if 'metadata' in link_dict:
160		link.extend_metadata(link_dict['metadata'])
161		s_vlan = link.get_metadata('s_vlan')
162		if s_vlan:
163		tag = TAG.from_dict(s_vlan)
164		link.update_metadata('s_vlan', tag)
165		return link
166
167	1	@staticmethod
168	1	def str_to_datetime(str_date):
169		"""Convert a string representing a date and time to datetime."""
170	1	date = datetime.datetime.strptime(str_date, TIME_FMT)
171	1	return date.astimezone(pytz.utc)
172
173	1	def maintenance_event(self, operation):
174		"""Create events to start/end a maintenance."""
175	1	if self._switches:
176	1	switches = []
177	1	for dpid in self._switches:
178	1	switch = self.controller.switches.get(dpid, None)
179	1	if switch:
180	1	switches.append(switch)
181	1	event = KytosEvent(name=f'kytos/maintenance.{operation}_switch',
182		content={'switches': switches})
183	1	self.controller.buffers.app.put(event)
184	1	if self._unis:
185	1	event = KytosEvent(name=f'kytos/maintenance.{operation}_uni',
186		content={'unis': self._unis})
187	1	self.controller.buffers.app.put(event)
188	1	if self._links:
189	1	event = KytosEvent(name=f'kytos/maintenance.{operation}_link',
190		content={'links': self._links})
191	1	self.controller.buffers.app.put(event)
192
193	1	def start_mw(self):
194		"""Actions taken when a maintenance window starts."""
195	1	self.status = Status.RUNNING
196	1	self.maintenance_event('start')
197
198	1	def end_mw(self):
199		"""Actions taken when a maintenance window finishes."""
200	1	self.status = Status.FINISHED
201	1	self.maintenance_event('end')
202
203
204	1	class Scheduler:
205		"""Scheduler for a maintenance window."""
206
207	1	def __init__(self):
208		"""Initialize a new scheduler."""
209	1	self.scheduler = BackgroundScheduler(timezone=pytz.utc)
210	1	self.scheduler.start()
211
212	1	def add(self, maintenance):
213		"""Add jobs to start and end a maintenance window."""
214	1	self.scheduler.add_job(maintenance.start_mw, 'date',
215		id=f'{maintenance.id}-start',
216		run_date=maintenance.start)
217	1	self.scheduler.add_job(maintenance.end_mw, 'date',
218		id=f'{maintenance.id}-end',
219		run_date=maintenance.end)
220
221	1	def remove(self, maintenance):
222		"""Remove jobs that start and end a maintenance window."""
223	1	try:
224	1	self.scheduler.remove_job(f'{maintenance.id}-start')
225	1	except JobLookupError:
226	1	log.info(f'Job to start {maintenance.id} already removed.')
227	1	try:
228	1	self.scheduler.remove_job(f'{maintenance.id}-end')
229	1	except JobLookupError:
230		log.info(f'Job to end {maintenance.id} already removed.')
231

kytos-ng / maintenance

Pull Request — master (#53)

build.models.MaintenanceWindow.str_to_datetime() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like