build.tests.unit.test_models.TestMW.test_update_items() - Code Metrics - Inspection of "Add in persistence to maintenance windows" - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#64)

unknown

created 2023-01-10 18:11 UTC

A

↳ Parent: build.tests.unit.test_models.TestMW.test_start_mw_case_1()

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""Tests for the models module."""

from unittest import TestCase
from unittest.mock import patch, MagicMock, call

from apscheduler.jobstores.base import JobLookupError
from apscheduler.triggers.date import DateTrigger
from datetime import datetime, timedelta
import pytz
from kytos.lib.helpers import get_controller_mock
from napps.kytos.maintenance.models import MaintenanceWindow as MW, Status, Scheduler
from napps.kytos.maintenance.models import MaintenanceStart, MaintenanceEnd
TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"


class TestMW(TestCase):
    """Test of the MaintenanceWindow class."""

    # pylint: disable=protected-access

    def setUp(self):
        """Initialize before tests are executed."""
        self.controller = get_controller_mock()
        self.start = datetime.now(pytz.utc)
        self.start += timedelta(days=1)
        self.end = self.start + timedelta(hours=6)
        self.switches = [
            "01:23:45:67:89:ab:cd:ef"
        ]
        self.maintenance = MW(
            start=self.start,
            end=self.end,
            switches=self.switches
        )

    def test_as_dict(self):
        """Test as_dict method."""
        mw_dict = self.maintenance.dict()
        expected_dict = {
            'description': '',
            'start': self.start,
            'end': self.end,
            'id': self.maintenance.id,
            'switches': self.switches,
            'interfaces': [],
            'links': [],
            'status': Status.PENDING,
            'inserted_at': None,
            'updated_at': None,
        }
        self.assertEqual(mw_dict, expected_dict)

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_start_mw_case_1(self, buffer_put_mock):
        """Test the method that starts a maintenance."""
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                    '01:23:45:67:89:ab:cd:ef',
                    '01:23:45:67:65:ab:cd:ef'
                ],
                'interfaces': [],
                'links': [],
            }
        )
        maintenance.start_mw(self.controller)
        buffer_put_mock.assert_called_once()

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_start_mw_case_2(self, buffer_put_mock):
        """Test the method that starts a maintenance."""
        interface_id = "interface_1"
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                    '01:23:45:67:89:ab:cd:ef',
                    '01:23:45:67:65:ab:cd:ef'
                ],
                'interfaces': [interface_id],
                'links': [],
            }
        )
        maintenance.start_mw(self.controller)
        self.assertEqual(buffer_put_mock.call_count, 2)

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_start_mw_case_3(self, buffer_put_mock):
        """Test the method that starts a maintenance."""
        interface_id = "interface_1"
        link1 = "link_1"
        link2 = "link_2"
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                ],
                'interfaces': [interface_id],
                'links': [link1, link2],
            }
        )
        maintenance.start_mw(self.controller)
        self.assertEqual(buffer_put_mock.call_count, 2)

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_end_mw_case_1(self, buffer_put_mock):
        """Test the method that ends a maintenance."""
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                    '01:23:45:67:89:ab:cd:ef',
                    '01:23:45:67:65:ab:cd:ef'
                ],
                'interfaces': [],
                'links': [],
                'status': Status.RUNNING,
            }
        )
        maintenance.end_mw(self.controller)
        buffer_put_mock.assert_called_once()

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_end_mw_case_2(self, buffer_put_mock):
        """Test the method that ends a maintenance."""
        interface_id = "interface_1"
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                    '01:23:45:67:89:ab:cd:ef',
                    '01:23:45:67:65:ab:cd:ef'
                ],
                'interfaces': [interface_id],
                'links': [],
                'status': Status.RUNNING,
            }
        )
        maintenance.end_mw(self.controller)
        self.assertEqual(buffer_put_mock.call_count, 2)

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_end_mw_case_3(self, buffer_put_mock):
        """Test the method that ends a maintenance."""
        interface_id = "interface_1"
        link1 = "link_1"
        link2 = "link_2"
        maintenance = self.maintenance.copy(
            update = {
                'switches': [
                ],
                'interfaces': [interface_id],
                'links': [link1, link2],
                'status': Status.RUNNING,
            }
        )
        maintenance.end_mw(self.controller)
        self.assertEqual(buffer_put_mock.call_count, 2)

    def test_start_in_past(self):
        start = datetime.now(pytz.utc) - timedelta(days=1)

        self.assertRaises(ValueError, MW,
            start=start,
            end=self.end,
            switches=self.switches,
        )

    def test_end_before_start(self):
        end = datetime.now(pytz.utc) - timedelta(days=1)

        self.assertRaises(ValueError, MW,
            start=self.start,
            end=end,
            switches=self.switches,
        )

    def test_items_empty(self):

        self.assertRaises(ValueError, MW,
            start=self.start,
            end=self.end,
        )


class TestScheduler(TestCase):
    """Test of the Scheduler Class"""
    
    def setUp(self) -> None:
        self.kytosController = MagicMock()
        self.db = MagicMock()
        self.task_scheduler = MagicMock()
        
        self.now = datetime.now(pytz.utc)

        self.window = MW.construct(
            id = 'Test Window',
            description = '',
            start = self.now + timedelta(hours=1),
            end = self.now + timedelta(hours=2),
            status = 'pending',
            switches = [],
            interfaces = [],
            links = [],
            updated_at = self.now - timedelta(days=1),
            inserted_at = self.now - timedelta(days=1),
        )

        self.scheduler = Scheduler(self.kytosController, self.db, self.task_scheduler)

    @patch.object(MW, 'start_mw')
    def test_start(self, start_mock):

        pending_window = self.window.copy(
            update={'id': 'pending window', 'status': 'pending'}
        )
        running_window = self.window.copy(
            update={'id': 'running window', 'status': 'running'}
        )
        finished_window = self.window.copy(
            update={'id': 'finished window', 'status': 'finished'}
        )

        expected_schedule_calls = [
            call(MaintenanceStart(self.scheduler, 'pending window'),
            'date', id='pending window-start',
            run_date = pending_window.start),
            call(MaintenanceEnd(self.scheduler, 'running window'),
            'date', id='running window-end',
            run_date = running_window.end),
        ]

        self.db.get_windows.return_value = [
            pending_window,
            running_window,
            finished_window,
        ]
        self.scheduler.start()

        resultant_schedule_calls = self.task_scheduler.add_job.call_args_list
        self.assertEqual(resultant_schedule_calls, expected_schedule_calls)

        # Couldn't mock individual start_mw functions
        # so they all share the same mock
        # pending_window.start_mw.assert_not_called()
        running_window.start_mw.assert_called_once_with(self.kytosController)
        # finished_window.start_mw.assert_not_called()

    @patch.object(MW, 'end_mw')
    def test_shutdown(self, end_mock):
        pending_window = self.window.copy(
            update={'id': 'pending window', 'status': 'pending'}
        )
        running_window = self.window.copy(
            update={'id': 'running window', 'status': 'running'}
        )
        finished_window = self.window.copy(
            update={'id': 'finished window', 'status': 'finished'}
        )

        self.db.get_windows.return_value = [
            pending_window,
            running_window,
            finished_window,
        ]

        remove_job_effects = {
            'pending window-start': False,
            'pending window-end': True,
            'running window-start': True,
            'running window-end': False,
            'finished window-start': True,
            'finished window-end': True,
        }

        def side_effect(job_id):
            effect = remove_job_effects[job_id]
            if effect:
                raise JobLookupError(job_id)
            else:
                return None

        self.task_scheduler.remove_job.side_effect = side_effect

        self.scheduler.shutdown()

        # Couldn't mock individual end_mw functions
        # so they all share the same mock
        # pending_window.end_mw.assert_not_called()
        running_window.end_mw.assert_called_once_with(self.kytosController)
        # finished_window.end_mw.assert_not_called()

    def test_update(self):
        pending_window = self.window.copy(
            update={'id': 'pending window', 'status': 'pending'}
        )
        running_window = self.window.copy(
            update={'id': 'running window', 'status': 'running'}
        )
        finished_window = self.window.copy(
            update={'id': 'finished window', 'status': 'finished'}
        )

        modify_job_effects = {
            'pending window-start': (False, DateTrigger(pending_window.start)),
            'pending window-end': (True, DateTrigger(pending_window.end)),
            'running window-start': (True, DateTrigger(running_window.start)),
            'running window-end': (False, DateTrigger(running_window.end)),
            'finished window-start': (True, DateTrigger(finished_window.start)),
            'finished window-end': (True, DateTrigger(finished_window.end)),
        }

        def side_effect(job_id, trigger):
            throw, expected_trigger = modify_job_effects[job_id]
            self.assertEqual(trigger.run_date, expected_trigger.run_date)
            if throw:
                raise JobLookupError(job_id)
            else:
                return None
        
        self.task_scheduler.modify_job.side_effect = side_effect

        self.scheduler.update(pending_window)
        self.scheduler.update(running_window)
        self.scheduler.update(finished_window)

    @patch.object(MW, 'start_mw')
    def test_maintenance_start(self, start_mock):

        pending_window = self.window.copy(
            update={'id': 'pending window', 'status': 'pending'}
        )
        next_window = self.window.copy(
            update={'id': 'pending window', 'status': 'running'}
        )

        self.db.start_window.return_value = next_window
        start = MaintenanceStart(self.scheduler, pending_window.id)
        start()
        next_window.start_mw.assert_called_once_with(self.kytosController)

        self.task_scheduler.add_job.assert_called_once_with(
            MaintenanceEnd(self.scheduler, pending_window.id),
            'date',
            id='pending window-end',
            run_date=pending_window.end
        )

    @patch.object(MW, 'end_mw')
    def test_maintenance_end(self, end_mock):

        running_window = self.window.copy(
            update={'id': 'running window', 'status': 'running'}
        )
        next_window = self.window.copy(
            update={'id': 'running window', 'status': 'finished'}
        )

        self.db.end_window.return_value = next_window
        end = MaintenanceEnd(self.scheduler, running_window.id)
        end()
        next_window.end_mw.assert_called_once_with(self.kytosController)



1			"""Tests for the models module."""
2
3			from unittest import TestCase
4			from unittest.mock import patch, MagicMock, call
5
6			from apscheduler.jobstores.base import JobLookupError
7			from apscheduler.triggers.date import DateTrigger
8			from datetime import datetime, timedelta
9			import pytz
10			from kytos.lib.helpers import get_controller_mock
11			from napps.kytos.maintenance.models import MaintenanceWindow as MW, Status, Scheduler
12			from napps.kytos.maintenance.models import MaintenanceStart, MaintenanceEnd
13			TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
14
15
16			class TestMW(TestCase):
17			"""Test of the MaintenanceWindow class."""
18
19			# pylint: disable=protected-access
20
21			def setUp(self):
22			"""Initialize before tests are executed."""
23			self.controller = get_controller_mock()
24			self.start = datetime.now(pytz.utc)
25			self.start += timedelta(days=1)
26			self.end = self.start + timedelta(hours=6)
27			self.switches = [
28			"01:23:45:67:89:ab:cd:ef"
29			]
30			self.maintenance = MW(
31			start=self.start,
32			end=self.end,
33			switches=self.switches
34			)
35
36			def test_as_dict(self):
37			"""Test as_dict method."""
38			mw_dict = self.maintenance.dict()
39			expected_dict = {
40			'description': '',
41			'start': self.start,
42			'end': self.end,
43			'id': self.maintenance.id,
44			'switches': self.switches,
45			'interfaces': [],
46			'links': [],
47			'status': Status.PENDING,
48			'inserted_at': None,
49			'updated_at': None,
50			}
51			self.assertEqual(mw_dict, expected_dict)
52
53			@patch('kytos.core.buffers.KytosEventBuffer.put')
54			def test_start_mw_case_1(self, buffer_put_mock):
55			"""Test the method that starts a maintenance."""
56			maintenance = self.maintenance.copy(
57			update = {
58			'switches': [
59			'01:23:45:67:89:ab:cd:ef',
60			'01:23:45:67:65:ab:cd:ef'
61			],
62			'interfaces': [],
63			'links': [],
64			}
65			)
66			maintenance.start_mw(self.controller)
67			buffer_put_mock.assert_called_once()
68
69			@patch('kytos.core.buffers.KytosEventBuffer.put')
70			def test_start_mw_case_2(self, buffer_put_mock):
71			"""Test the method that starts a maintenance."""
72			interface_id = "interface_1"
73			maintenance = self.maintenance.copy(
74			update = {
75			'switches': [
76			'01:23:45:67:89:ab:cd:ef',
77			'01:23:45:67:65:ab:cd:ef'
78			],
79			'interfaces': [interface_id],
80			'links': [],
81			}
82			)
83			maintenance.start_mw(self.controller)
84			self.assertEqual(buffer_put_mock.call_count, 2)
85
86			@patch('kytos.core.buffers.KytosEventBuffer.put')
87			def test_start_mw_case_3(self, buffer_put_mock):
88			"""Test the method that starts a maintenance."""
89			interface_id = "interface_1"
90			link1 = "link_1"
91			link2 = "link_2"
92			maintenance = self.maintenance.copy(
93			update = {
94			'switches': [
95			],
96			'interfaces': [interface_id],
97			'links': [link1, link2],
98			}
99			)
100			maintenance.start_mw(self.controller)
101			self.assertEqual(buffer_put_mock.call_count, 2)
102
103			@patch('kytos.core.buffers.KytosEventBuffer.put')
104			def test_end_mw_case_1(self, buffer_put_mock):
105			"""Test the method that ends a maintenance."""
106			maintenance = self.maintenance.copy(
107			update = {
108			'switches': [
109			'01:23:45:67:89:ab:cd:ef',
110			'01:23:45:67:65:ab:cd:ef'
111			],
112			'interfaces': [],
113			'links': [],
114			'status': Status.RUNNING,
115			}
116			)
117			maintenance.end_mw(self.controller)
118			buffer_put_mock.assert_called_once()
119
120			@patch('kytos.core.buffers.KytosEventBuffer.put')
121			def test_end_mw_case_2(self, buffer_put_mock):
122			"""Test the method that ends a maintenance."""
123			interface_id = "interface_1"
124			maintenance = self.maintenance.copy(
125			update = {
126			'switches': [
127			'01:23:45:67:89:ab:cd:ef',
128			'01:23:45:67:65:ab:cd:ef'
129			],
130			'interfaces': [interface_id],
131			'links': [],
132			'status': Status.RUNNING,
133			}
134			)
135			maintenance.end_mw(self.controller)
136			self.assertEqual(buffer_put_mock.call_count, 2)
137
138			@patch('kytos.core.buffers.KytosEventBuffer.put')
139			def test_end_mw_case_3(self, buffer_put_mock):
140			"""Test the method that ends a maintenance."""
141			interface_id = "interface_1"
142			link1 = "link_1"
143			link2 = "link_2"
144			maintenance = self.maintenance.copy(
145			update = {
146			'switches': [
147			],
148			'interfaces': [interface_id],
149			'links': [link1, link2],
150			'status': Status.RUNNING,
151			}
152			)
153			maintenance.end_mw(self.controller)
154			self.assertEqual(buffer_put_mock.call_count, 2)
155
156			def test_start_in_past(self):
157			start = datetime.now(pytz.utc) - timedelta(days=1)
158
159			self.assertRaises(ValueError, MW,
160			start=start,
161			end=self.end,
162			switches=self.switches,
163			)
164
165			def test_end_before_start(self):
166			end = datetime.now(pytz.utc) - timedelta(days=1)
167
168			self.assertRaises(ValueError, MW,
169			start=self.start,
170			end=end,
171			switches=self.switches,
172			)
173
174			def test_items_empty(self):
175
176			self.assertRaises(ValueError, MW,
177			start=self.start,
178			end=self.end,
179			)
180
181
182			class TestScheduler(TestCase):
183			"""Test of the Scheduler Class"""
184
185			def setUp(self) -> None:
186			self.kytosController = MagicMock()
187			self.db = MagicMock()
188			self.task_scheduler = MagicMock()
189
190			self.now = datetime.now(pytz.utc)
191
192			self.window = MW.construct(
193			id = 'Test Window',
194			description = '',
195			start = self.now + timedelta(hours=1),
196			end = self.now + timedelta(hours=2),
197			status = 'pending',
198			switches = [],
199			interfaces = [],
200			links = [],
201			updated_at = self.now - timedelta(days=1),
202			inserted_at = self.now - timedelta(days=1),
203			)
204
205			self.scheduler = Scheduler(self.kytosController, self.db, self.task_scheduler)
206
207			@patch.object(MW, 'start_mw')
208			def test_start(self, start_mock):
209
210			pending_window = self.window.copy(
211			update={'id': 'pending window', 'status': 'pending'}
212			)
213			running_window = self.window.copy(
214			update={'id': 'running window', 'status': 'running'}
215			)
216			finished_window = self.window.copy(
217			update={'id': 'finished window', 'status': 'finished'}
218			)
219
220			expected_schedule_calls = [
221			call(MaintenanceStart(self.scheduler, 'pending window'),
222			'date', id='pending window-start',
223			run_date = pending_window.start),
224			call(MaintenanceEnd(self.scheduler, 'running window'),
225			'date', id='running window-end',
226			run_date = running_window.end),
227			]
228
229			self.db.get_windows.return_value = [
230			pending_window,
231			running_window,
232			finished_window,
233			]
234			self.scheduler.start()
235
236			resultant_schedule_calls = self.task_scheduler.add_job.call_args_list
237			self.assertEqual(resultant_schedule_calls, expected_schedule_calls)
238
239			# Couldn't mock individual start_mw functions
240			# so they all share the same mock
241			# pending_window.start_mw.assert_not_called()
242			running_window.start_mw.assert_called_once_with(self.kytosController)
243			# finished_window.start_mw.assert_not_called()
244
245			@patch.object(MW, 'end_mw')
246			def test_shutdown(self, end_mock):
247			pending_window = self.window.copy(
248			update={'id': 'pending window', 'status': 'pending'}
249			)
250			running_window = self.window.copy(
251			update={'id': 'running window', 'status': 'running'}
252			)
253			finished_window = self.window.copy(
254			update={'id': 'finished window', 'status': 'finished'}
255			)
256
257			self.db.get_windows.return_value = [
258			pending_window,
259			running_window,
260			finished_window,
261			]
262
263			remove_job_effects = {
264			'pending window-start': False,
265			'pending window-end': True,
266			'running window-start': True,
267			'running window-end': False,
268			'finished window-start': True,
269			'finished window-end': True,
270			}
271
272			def side_effect(job_id):
273			effect = remove_job_effects[job_id]
274			if effect:
275			raise JobLookupError(job_id)
276			else:
277			return None
278
279			self.task_scheduler.remove_job.side_effect = side_effect
280
281			self.scheduler.shutdown()
282
283			# Couldn't mock individual end_mw functions
284			# so they all share the same mock
285			# pending_window.end_mw.assert_not_called()
286			running_window.end_mw.assert_called_once_with(self.kytosController)
287			# finished_window.end_mw.assert_not_called()
288
289			def test_update(self):
290			pending_window = self.window.copy(
291			update={'id': 'pending window', 'status': 'pending'}
292			)
293			running_window = self.window.copy(
294			update={'id': 'running window', 'status': 'running'}
295			)
296			finished_window = self.window.copy(
297			update={'id': 'finished window', 'status': 'finished'}
298			)
299
300			modify_job_effects = {
301			'pending window-start': (False, DateTrigger(pending_window.start)),
302			'pending window-end': (True, DateTrigger(pending_window.end)),
303			'running window-start': (True, DateTrigger(running_window.start)),
304			'running window-end': (False, DateTrigger(running_window.end)),
305			'finished window-start': (True, DateTrigger(finished_window.start)),
306			'finished window-end': (True, DateTrigger(finished_window.end)),
307			}
308
309			def side_effect(job_id, trigger):
310			throw, expected_trigger = modify_job_effects[job_id]
311			self.assertEqual(trigger.run_date, expected_trigger.run_date)
312			if throw:
313			raise JobLookupError(job_id)
314			else:
315			return None
316
317			self.task_scheduler.modify_job.side_effect = side_effect
318
319			self.scheduler.update(pending_window)
320			self.scheduler.update(running_window)
321			self.scheduler.update(finished_window)
322
323			@patch.object(MW, 'start_mw')
324			def test_maintenance_start(self, start_mock):
325
326			pending_window = self.window.copy(
327			update={'id': 'pending window', 'status': 'pending'}
328			)
329			next_window = self.window.copy(
330			update={'id': 'pending window', 'status': 'running'}
331			)
332
333			self.db.start_window.return_value = next_window
334			start = MaintenanceStart(self.scheduler, pending_window.id)
335			start()
336			next_window.start_mw.assert_called_once_with(self.kytosController)
337
338			self.task_scheduler.add_job.assert_called_once_with(
339			MaintenanceEnd(self.scheduler, pending_window.id),
340			'date',
341			id='pending window-end',
342			run_date=pending_window.end
343			)
344
345			@patch.object(MW, 'end_mw')
346			def test_maintenance_end(self, end_mock):
347
348			running_window = self.window.copy(
349			update={'id': 'running window', 'status': 'running'}
350			)
351			next_window = self.window.copy(
352			update={'id': 'running window', 'status': 'finished'}
353			)
354
355			self.db.end_window.return_value = next_window
356			end = MaintenanceEnd(self.scheduler, running_window.id)
357			end()
358			next_window.end_mw.assert_called_once_with(self.kytosController)
359
360

kytos-ng / maintenance

Pull Request — master (#64)

A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like