build.tests.unit.test_models.TestScheduler.setUp() - 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-03 01:46 UTC

build.tests.unit.test_models.TestScheduler.setUp() A

↳ Parent: build.tests.unit.test_models

Complexity

Conditions

Size

Total Lines	21
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	17
nop	1
dl	0
loc	21
rs	9.55
c	0
b	0
f	0

"""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 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': [],
                '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 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': [],
                '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 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],
                'status': Status.RUNNING,
            }
        )
        maintenance.end_mw(self.controller)
        self.assertEqual(buffer_put_mock.call_count, 2)


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)

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(start = self.start, end = self.end,
31			switches = self.switches)
32
33			def test_as_dict(self):
34			"""Test as_dict method."""
35			mw_dict = self.maintenance.dict()
36			expected_dict = {
37			'description': '',
38			'start': self.start,
39			'end': self.end,
40			'id': self.maintenance.id,
41			'switches': self.switches,
42			'interfaces': [],
43			'links': [],
44			'status': Status.PENDING,
45			'inserted_at': None,
46			'updated_at': None,
47			}
48			self.assertEqual(mw_dict, expected_dict)
49
50			@patch('kytos.core.buffers.KytosEventBuffer.put')
51			def test_start_mw_case_1(self, buffer_put_mock):
52			"""Test the method that starts a maintenance."""
53			maintenance = self.maintenance.copy(
54			update = {
55			'switches': [
56			'01:23:45:67:89:ab:cd:ef',
57			'01:23:45:67:65:ab:cd:ef'
58			],
59			'interfaces': [],
60			'links': [],
61			}
62			)
63			maintenance.start_mw(self.controller)
64			buffer_put_mock.assert_called_once()
65
66			@patch('kytos.core.buffers.KytosEventBuffer.put')
67			def test_start_mw_case_2(self, buffer_put_mock):
68			"""Test the method that starts a maintenance."""
69			interface_id = "interface_1"
70			maintenance = self.maintenance.copy(
71			update = {
72			'switches': [
73			'01:23:45:67:89:ab:cd:ef',
74			'01:23:45:67:65:ab:cd:ef'
75			],
76			'interfaces': [interface_id],
77			'links': [],
78			}
79			)
80			maintenance.start_mw(self.controller)
81			self.assertEqual(buffer_put_mock.call_count, 2)
82
83			@patch('kytos.core.buffers.KytosEventBuffer.put')
84			def test_start_mw_case_3(self, buffer_put_mock):
85			"""Test the method that starts a maintenance."""
86			interface_id = "interface_1"
87			link1 = "link_1"
88			link2 = "link_2"
89			maintenance = self.maintenance.copy(
90			update = {
91			'switches': [
92			],
93			'interfaces': [interface_id],
94			'links': [link1, link2],
95			}
96			)
97			maintenance.start_mw(self.controller)
98			self.assertEqual(buffer_put_mock.call_count, 2)
99
100			@patch('kytos.core.buffers.KytosEventBuffer.put')
101			def test_end_mw_case_1(self, buffer_put_mock):
102			"""Test the method that starts a maintenance."""
103			maintenance = self.maintenance.copy(
104			update = {
105			'switches': [
106			'01:23:45:67:89:ab:cd:ef',
107			'01:23:45:67:65:ab:cd:ef'
108			],
109			'interfaces': [],
110			'links': [],
111			'status': Status.RUNNING,
112			}
113			)
114			maintenance.end_mw(self.controller)
115			buffer_put_mock.assert_called_once()
116
117			@patch('kytos.core.buffers.KytosEventBuffer.put')
118			def test_end_mw_case_2(self, buffer_put_mock):
119			"""Test the method that starts a maintenance."""
120			interface_id = "interface_1"
121			maintenance = self.maintenance.copy(
122			update = {
123			'switches': [
124			'01:23:45:67:89:ab:cd:ef',
125			'01:23:45:67:65:ab:cd:ef'
126			],
127			'interfaces': [interface_id],
128			'links': [],
129			'status': Status.RUNNING,
130			}
131			)
132			maintenance.end_mw(self.controller)
133			self.assertEqual(buffer_put_mock.call_count, 2)
134
135			@patch('kytos.core.buffers.KytosEventBuffer.put')
136			def test_end_mw_case_3(self, buffer_put_mock):
137			"""Test the method that starts a maintenance."""
138			interface_id = "interface_1"
139			link1 = "link_1"
140			link2 = "link_2"
141			maintenance = self.maintenance.copy(
142			update = {
143			'switches': [
144			],
145			'interfaces': [interface_id],
146			'links': [link1, link2],
147			'status': Status.RUNNING,
148			}
149			)
150			maintenance.end_mw(self.controller)
151			self.assertEqual(buffer_put_mock.call_count, 2)
152
153
154			class TestScheduler(TestCase):
155			"""Test of the Scheduler Class"""
156
157			def setUp(self) -> None:
158			self.kytosController = MagicMock()
159			self.db = MagicMock()
160			self.task_scheduler = MagicMock()
161
162			self.now = datetime.now(pytz.utc)
163
164			self.window = MW.construct(
165			id = 'Test Window',
166			description = '',
167			start = self.now + timedelta(hours=1),
168			end = self.now + timedelta(hours=2),
169			status = 'pending',
170			switches = [],
171			interfaces = [],
172			links = [],
173			updated_at = self.now - timedelta(days=1),
174			inserted_at = self.now - timedelta(days=1),
175			)
176
177			self.scheduler = Scheduler(self.kytosController, self.db, self.task_scheduler)
178
179			@patch.object(MW, 'start_mw')
180			def test_start(self, start_mock):
181
182			pending_window = self.window.copy(
183			update={'id': 'pending window', 'status': 'pending'}
184			)
185			running_window = self.window.copy(
186			update={'id': 'running window', 'status': 'running'}
187			)
188			finished_window = self.window.copy(
189			update={'id': 'finished window', 'status': 'finished'}
190			)
191
192			expected_schedule_calls = [
193			call(MaintenanceStart(self.scheduler, 'pending window'),
194			'date', id='pending window-start',
195			run_date = pending_window.start),
196			call(MaintenanceEnd(self.scheduler, 'running window'),
197			'date', id='running window-end',
198			run_date = running_window.end),
199			]
200
201			self.db.get_windows.return_value = [
202			pending_window,
203			running_window,
204			finished_window,
205			]
206			self.scheduler.start()
207
208			resultant_schedule_calls = self.task_scheduler.add_job.call_args_list
209			self.assertEqual(resultant_schedule_calls, expected_schedule_calls)
210
211			# Couldn't mock individual start_mw functions
212			# so they all share the same mock
213			# pending_window.start_mw.assert_not_called()
214			running_window.start_mw.assert_called_once_with(self.kytosController)
215			# finished_window.start_mw.assert_not_called()
216
217			@patch.object(MW, 'end_mw')
218			def test_shutdown(self, end_mock):
219			pending_window = self.window.copy(
220			update={'id': 'pending window', 'status': 'pending'}
221			)
222			running_window = self.window.copy(
223			update={'id': 'running window', 'status': 'running'}
224			)
225			finished_window = self.window.copy(
226			update={'id': 'finished window', 'status': 'finished'}
227			)
228
229			self.db.get_windows.return_value = [
230			pending_window,
231			running_window,
232			finished_window,
233			]
234
235			remove_job_effects = {
236			'pending window-start': False,
237			'pending window-end': True,
238			'running window-start': True,
239			'running window-end': False,
240			'finished window-start': True,
241			'finished window-end': True,
242			}
243
244			def side_effect(job_id):
245			effect = remove_job_effects[job_id]
246			if effect:
247			raise JobLookupError(job_id)
248			else:
249			return None
250
251			self.task_scheduler.remove_job.side_effect = side_effect
252
253			self.scheduler.shutdown()
254
255			# Couldn't mock individual end_mw functions
256			# so they all share the same mock
257			# pending_window.end_mw.assert_not_called()
258			running_window.end_mw.assert_called_once_with(self.kytosController)
259			# finished_window.end_mw.assert_not_called()
260
261			def test_update(self):
262			pending_window = self.window.copy(
263			update={'id': 'pending window', 'status': 'pending'}
264			)
265			running_window = self.window.copy(
266			update={'id': 'running window', 'status': 'running'}
267			)
268			finished_window = self.window.copy(
269			update={'id': 'finished window', 'status': 'finished'}
270			)
271
272			modify_job_effects = {
273			'pending window-start': (False, DateTrigger(pending_window.start)),
274			'pending window-end': (True, DateTrigger(pending_window.end)),
275			'running window-start': (True, DateTrigger(running_window.start)),
276			'running window-end': (False, DateTrigger(running_window.end)),
277			'finished window-start': (True, DateTrigger(finished_window.start)),
278			'finished window-end': (True, DateTrigger(finished_window.end)),
279			}
280
281			def side_effect(job_id, trigger):
282			throw, expected_trigger = modify_job_effects[job_id]
283			self.assertEqual(trigger.run_date, expected_trigger.run_date)
284			if throw:
285			raise JobLookupError(job_id)
286			else:
287			return None
288
289			self.task_scheduler.modify_job.side_effect = side_effect
290
291			self.scheduler.update(pending_window)
292			self.scheduler.update(running_window)
293			self.scheduler.update(finished_window)

kytos-ng / maintenance

Pull Request — master (#64)

build.tests.unit.test_models.TestScheduler.setUp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like