Passed
Push — master ( a20a5a...0d5b5f )
by Vinicius
15:33 queued 13:51
created

build.tests.unit.test_main.TestMain.setup_method()   A

Complexity

Conditions 2

Size

Total Lines 11
Code Lines 10

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 10
CRAP Score 2

Importance

Changes 0
Metric Value
cc 2
eloc 10
nop 1
dl 0
loc 11
ccs 10
cts 10
cp 1
crap 2
rs 9.9
c 0
b 0
f 0
1
"""Tests for the main madule."""
2
3 1
from unittest.mock import patch, MagicMock
4 1
from datetime import datetime, timedelta
5
6 1
import pytz
7
8 1
from kytos.lib.helpers import get_controller_mock, get_test_client
9 1
from napps.kytos.maintenance.main import Main
10 1
from napps.kytos.maintenance.models import MaintenanceWindow as MW
11 1
from napps.kytos.maintenance.models import MaintenanceWindows
12
13 1
TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
14
15
16 1
class TestMain:
17
    """Test the Main class of this NApp."""
18
19
    # pylint: disable=too-many-public-methods
20
21 1
    def setup_method(self):
22
        """Initialize before tests are executed."""
23 1
        self.controller = get_controller_mock()
24 1
        self.scheduler = MagicMock()
25 1
        new_sched = 'napps.kytos.maintenance.models.Scheduler.new_scheduler'
26 1
        with patch(new_sched) as new_scheduler:
27 1
            new_scheduler.return_value = self.scheduler
28 1
            self.napp = Main(self.controller)
29 1
        self.api = get_test_client(self.controller, self.napp)
30 1
        self.maxDiff = None
31 1
        self.base_endpoint = "kytos/maintenance/v1"
32
33 1
    async def test_create_mw_case_1(self, event_loop):
34
        """Test a successful case of the REST to create."""
35 1
        self.napp.controller.loop = event_loop
36 1
        url = f'{self.base_endpoint}'
37 1
        start = datetime.now(pytz.utc) + timedelta(days=1)
38 1
        end = start + timedelta(hours=2)
39 1
        payload = {
40
            'id': '1234',
41
            "start": start.strftime(TIME_FMT),
42
            "end": end.strftime(TIME_FMT),
43
            "switches": [
44
                "00:00:00:00:00:00:02",
45
            ],
46
            'interfaces': [
47
                "00:00:00:00:00:00:00:03:3",
48
            ],
49
        }
50 1
        response = await self.api.post(url, json=payload)
51 1
        assert response.status_code == 201
52 1
        current_data = response.json()
53 1
        self.scheduler.add.assert_called_once_with(
54
            MW.construct(
55
                id='1234',
56
                start=start.replace(microsecond=0),
57
                end=end.replace(microsecond=0),
58
                switches=['00:00:00:00:00:00:02'],
59
                interfaces=['00:00:00:00:00:00:00:03:3']
60
            ),
61
            force=False
62
        )
63 1
        assert current_data == {'mw_id': '1234'}
64
65 1
    async def test_create_mw_case_2(self, event_loop):
66
        """Test a fail case of the REST to create a maintenance window."""
67 1
        self.napp.controller.loop = event_loop
68 1
        url = f'{self.base_endpoint}'
69 1
        payload = {
70
            "switches": [
71
                "00:00:00:00:00:00:02",
72
            ],
73
            'interfaces': [
74
                "00:00:00:00:00:00:00:03:3",
75
            ],
76
        }
77 1
        response = await self.api.post(url, json=payload)
78 1
        assert response.status_code == 400
79 1
        self.scheduler.add.assert_not_called()
80
81 1 View Code Duplication
    async def test_create_mw_case_3(self, event_loop):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
82
        """Test a fail case of the REST to create a maintenance window."""
83 1
        self.napp.controller.loop = event_loop
84 1
        url = f'{self.base_endpoint}'
85 1
        start = datetime.now(pytz.utc) - timedelta(days=1)
86 1
        end = start + timedelta(hours=2)
87 1
        payload = {
88
            "start": start.strftime(TIME_FMT),
89
            "end": end.strftime(TIME_FMT),
90
            "switches": [
91
                "00:00:00:00:00:00:02",
92
            ],
93
            'interfaces': [
94
                "00:00:00:00:00:00:00:03:3",
95
            ],
96
        }
97 1
        response = await self.api.post(url, json=payload)
98 1
        current_data = response.json()
99 1
        assert response.status_code == 400
100 1
        assert current_data['description'] == \
101
                         'Start in the past not allowed'
102 1
        self.scheduler.add.assert_not_called()
103
104 1 View Code Duplication
    async def test_create_mw_case_4(self, event_loop):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
105
        """Test a fail case of the REST to create a maintenance window."""
106 1
        self.napp.controller.loop = event_loop
107 1
        url = f'{self.base_endpoint}'
108 1
        start = datetime.now(pytz.utc) + timedelta(days=1)
109 1
        end = start - timedelta(hours=2)
110 1
        payload = {
111
            "start": start.strftime(TIME_FMT),
112
            "end": end.strftime(TIME_FMT),
113
            "switches": [
114
                "00:00:00:00:00:00:02",
115
            ],
116
            'interfaces': [
117
                "00:00:00:00:00:00:00:03:3",
118
            ],
119
        }
120 1
        response = await self.api.post(url, json=payload)
121 1
        current_data = response.json()
122
123 1
        assert response.status_code == 400
124 1
        assert current_data['description'] == \
125
                         'End before start not allowed'
126 1
        self.scheduler.add.assert_not_called()
127
128 1
    async def test_get_mw_case_1(self):
129
        """Test get all maintenance windows, empty list."""
130 1
        self.scheduler.list_maintenances.return_value = MaintenanceWindows.construct(__root__ = [])
131 1
        url = f'{self.base_endpoint}'
132 1
        response = await self.api.get(url)
133 1
        current_data = response.json()
134 1
        assert response.status_code == 200
135 1
        assert current_data == []
136 1
        self.scheduler.list_maintenances.assert_called_once()
137
138 1
    async def test_get_mw_case_2(self):
139
        """Test get all maintenance windows."""
140 1
        now = datetime.now(pytz.utc)
141 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
142 1
        end1 = start1 + timedelta(hours=6)
143 1
        start2 = datetime.now(pytz.utc) + timedelta(hours=5)
144 1
        end2 = start2 + timedelta(hours=1, minutes=30)
145 1
        self.scheduler.list_maintenances.return_value = MaintenanceWindows.construct(
146
            __root__ = [
147
            MW.construct(
148
                id = '1234',
149
                start = start1.replace(microsecond=0),
150
                end = end1.replace(microsecond=0),
151
                switches = [
152
                    '00:00:00:00:00:00:12:23'
153
                ],
154
                description = '',
155
                links = [],
156
                interfaces = [],
157
                status = 'pending',
158
                updated_at = now.replace(microsecond=0),
159
                inserted_at = now.replace(microsecond=0),
160
            ),
161
            MW.construct(
162
                id = '4567',
163
                start = start2.replace(microsecond=0),
164
                end = end2.replace(microsecond=0),
165
                switches = [
166
                    '12:34:56:78:90:ab:cd:ef'
167
                ],
168
                description = '',
169
                links = [],
170
                interfaces = [],
171
                status = 'pending',
172
                updated_at = now.replace(microsecond=0),
173
                inserted_at = now.replace(microsecond=0),
174
            ),
175
        ])
176 1
        mw_dict = [
177
            {
178
                'id': '1234',
179
                'start': start1.strftime(TIME_FMT),
180
                'end': end1.strftime(TIME_FMT),
181
                'switches': [
182
                    '00:00:00:00:00:00:12:23'
183
                ],
184
                'description': '',
185
                'links': [],
186
                'interfaces': [],
187
                'status': 'pending',
188
                'updated_at': now.strftime(TIME_FMT),
189
                'inserted_at': now.strftime(TIME_FMT),
190
            },
191
            {
192
                'id': '4567',
193
                'start': start2.strftime(TIME_FMT),
194
                'end': end2.strftime(TIME_FMT),
195
                'switches': [
196
                    '12:34:56:78:90:ab:cd:ef'
197
                ],
198
                'description': '',
199
                'links': [],
200
                'interfaces': [],
201
                'status': 'pending',
202
                'updated_at': now.strftime(TIME_FMT),
203
                'inserted_at': now.strftime(TIME_FMT),
204
            }
205
        ]
206
207 1
        url = f'{self.base_endpoint}'
208 1
        response = await self.api.get(url)
209 1
        current_data = response.json()
210 1
        assert response.status_code == 200
211 1
        assert current_data == mw_dict
212 1
        self.scheduler.list_maintenances.assert_called_once()
213
214 1
    async def test_get_mw_case_3(self):
215
        """Test get non-existent id."""
216 1
        self.scheduler.get_maintenance.return_value = None
217 1
        url = f'{self.base_endpoint}/2345'
218 1
        response = await self.api.get(url)
219 1
        current_data = response.json()
220 1
        assert response.status_code == 404
221 1
        assert current_data['description'] == \
222
                         'Maintenance with id 2345 not found'
223 1
        self.scheduler.get_maintenance.assert_called_once_with('2345')
224
225 1
    async def test_get_mw_case_4(self):
226
        """Test get existent id."""
227 1
        now = datetime.now(pytz.utc)
228 1
        start2 = datetime.now(pytz.utc) + timedelta(hours=5)
229 1
        end2 = start2 + timedelta(hours=1, minutes=30)
230 1
        self.scheduler.get_maintenance.return_value = MW.construct(
231
            id = '4567',
232
            start = start2.replace(microsecond=0),
233
            end = end2.replace(microsecond=0),
234
            switches = [
235
                '12:34:56:78:90:ab:cd:ef'
236
            ],
237
            updated_at = now.replace(microsecond=0),
238
            inserted_at = now.replace(microsecond=0),
239
        )
240 1
        mw_dict = {
241
            'id': '4567',
242
            'start': start2.strftime(TIME_FMT),
243
            'end': end2.strftime(TIME_FMT),
244
            'switches': [
245
                '12:34:56:78:90:ab:cd:ef'
246
            ],
247
            'description': '',
248
            'links': [],
249
            'interfaces': [],
250
            'status': 'pending',
251
            'updated_at': now.strftime(TIME_FMT),
252
            'inserted_at': now.strftime(TIME_FMT),
253
        }
254 1
        url = f'{self.base_endpoint}/4567'
255 1
        response = await self.api.get(url)
256 1
        current_data = response.json()
257 1
        assert response.status_code == 200
258 1
        assert current_data == mw_dict
259 1
        self.scheduler.get_maintenance.assert_called_once_with('4567')
260
261 1
    async def test_remove_mw_case_1(self):
262
        """Test remove non-existent id."""
263 1
        self.scheduler.get_maintenance.return_value = None
264 1
        url = f'{self.base_endpoint}/2345'
265 1
        response = await self.api.delete(url)
266 1
        current_data = response.json()
267 1
        assert response.status_code == 404
268 1
        assert current_data['description'] == \
269
                         'Maintenance with id 2345 not found'
270 1
        self.scheduler.get_maintenance.assert_called_once_with('2345')
271 1
        self.scheduler.remove.assert_not_called()
272
273 1
    async def test_remove_mw_case_2(self):
274
        """Test remove existent id."""
275 1
        start1 = datetime.now(pytz.utc) + timedelta(hours=1)
276 1
        end1 = start1 + timedelta(hours=6)
277 1
        self.scheduler.get_maintenance.return_value = MW.construct(
278
            id = '1234',
279
            start = start1.replace(microsecond=0),
280
            end = end1.replace(microsecond=0),
281
            switches = [
282
                '00:00:00:00:00:00:12:23'
283
            ],
284
        )
285 1
        url = f'{self.base_endpoint}/1234'
286 1
        response = await self.api.delete(url)
287 1
        current_data = response.json()
288 1
        assert response.status_code == 200
289 1
        assert current_data == {'response': 'Maintenance with id 1234 ' \
290
                                                    'successfully removed'}
291 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
292 1
        self.scheduler.remove.assert_called_once_with('1234')
293
294 1 View Code Duplication
    async def test_remove_mw_case_3(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
295
        """Test remove existent id."""
296 1
        start1 = datetime.now(pytz.utc) - timedelta(days=1)
297 1
        end1 = start1 + timedelta(hours=6)
298 1
        self.scheduler.get_maintenance.return_value = MW.construct(
299
            id = '1234',
300
            start = start1.replace(microsecond=0),
301
            end = end1.replace(microsecond=0),
302
            switches = [
303
                '00:00:00:00:00:00:12:23'
304
            ],
305
            status = 'running',
306
        )
307 1
        url = f'{self.base_endpoint}/1234'
308 1
        response = await self.api.delete(url)
309 1
        current_data = response.json()
310 1
        assert response.status_code == 400
311 1
        assert current_data['description'] == \
312
                         'Deleting a running maintenance is not allowed'
313 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
314 1
        self.scheduler.remove.assert_not_called()
315
316 1
    async def test_update_mw_case_1(self, event_loop):
317
        """Test update non-existent id."""
318 1
        self.napp.controller.loop = event_loop
319 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
320 1
        self.scheduler.get_maintenance.return_value = None
321 1
        payload = {
322
            "start": start1.strftime(TIME_FMT),
323
        }
324 1
        url = f'{self.base_endpoint}/2345'
325 1
        response = await self.api.patch(url, json=payload)
326 1
        current_data = response.json()
327 1
        assert response.status_code == 404
328 1
        assert current_data['description'] == \
329
                         'Maintenance with id 2345 not found'
330 1
        self.scheduler.update.assert_not_called()
331
332 1
    async def test_update_mw_case_2(self, event_loop):
333
        """Test update no data."""
334 1
        self.napp.controller.loop = event_loop
335 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
336 1
        payload = {
337
            "start": start1.strftime(TIME_FMT),
338
        }
339 1
        url = f'{self.base_endpoint}/1234'
340 1
        response = await self.api.patch(url, json=payload)
341 1
        current_data = response.json()
342 1
        assert response.status_code == 400
343 1
        assert "field required" in current_data['description']
344
345 1
    async def test_update_mw_case_3(self, event_loop):
346
        """Test successful update."""
347 1
        self.napp.controller.loop = event_loop
348 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
349 1
        end1 = start1 + timedelta(hours=6)
350 1
        self.scheduler.get_maintenance.return_value = MW.construct(
351
            id = '1234',
352
            start = start1.replace(microsecond=0),
353
            end = end1.replace(microsecond=0),
354
            switches = [
355
                '00:00:00:00:00:00:12:23'
356
            ],
357
        )
358 1
        start_new = datetime.now(pytz.utc) + timedelta(days=1, hours=3)
359 1
        payload = {
360
            "start": start_new.strftime(TIME_FMT),
361
        }
362 1
        url = f'{self.base_endpoint}/1234'
363 1
        response = await self.api.patch(url, json=payload)
364 1
        current_data = response.json()
365 1
        assert current_data == \
366
                         {'response': 'Maintenance 1234 updated'}
367 1
        assert response.status_code == 200
368 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
369 1
        self.scheduler.update.assert_called_once_with(
370
            MW.construct(
371
                id = '1234',
372
                start = start_new.replace(microsecond=0),
373
                end = end1.replace(microsecond=0),
374
                switches = [
375
                    '00:00:00:00:00:00:12:23'
376
                ],
377
            )
378
        )
379
380 1
    async def test_update_mw_case_4(self, event_loop):
381
        """Test successful update."""
382 1
        self.napp.controller.loop = event_loop
383 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
384 1
        end1 = start1 + timedelta(hours=6)
385 1
        self.scheduler.get_maintenance.return_value = MW.construct(
386
            id = '1234',
387
            start = start1.replace(microsecond=0),
388
            end = end1.replace(microsecond=0),
389
            switches = [
390
                '00:00:00:00:00:00:12:23'
391
            ],
392
        )
393 1
        start_new = datetime.now(pytz.utc) - timedelta(days=1, hours=3)
394 1
        payload = {
395
            "start": start_new.strftime(TIME_FMT),
396
        }
397 1
        url = f'{self.base_endpoint}/1234'
398 1
        response = await self.api.patch(url, json=payload)
399 1
        current_data = response.json()
400 1
        assert response.status_code == 400
401 1
        assert current_data['description'] == \
402
                         'Start in the past not allowed'
403 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
404 1
        self.scheduler.update.assert_not_called()
405
406 1
    async def test_update_mw_case_5(self, event_loop):
407
        """Test successful update."""
408 1
        self.napp.controller.loop = event_loop
409 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
410 1
        end1 = start1 + timedelta(hours=6)
411 1
        self.scheduler.get_maintenance.return_value = MW.construct(
412
            id = '1234',
413
            start = start1.replace(microsecond=0),
414
            end = end1.replace(microsecond=0),
415
            switches = [
416
                '00:00:00:00:00:00:12:23'
417
            ],
418
        )
419 1
        start_new = datetime.now(pytz.utc) + timedelta(days=1, hours=3)
420 1
        end_new = start_new - timedelta(hours=5)
421 1
        payload = {
422
            "start": start_new.strftime(TIME_FMT),
423
            "end": end_new.strftime(TIME_FMT)
424
        }
425 1
        url = f'{self.base_endpoint}/1234'
426 1
        response = await self.api.patch(url, json=payload)
427 1
        current_data = response.json()
428 1
        assert response.status_code == 400
429 1
        assert current_data['description'] == \
430
                         'End before start not allowed'
431 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
432 1
        self.scheduler.update.assert_not_called()
433
434 1
    async def test_update_mw_case_6(self, event_loop):
435
        """Test successful update."""
436 1
        self.napp.controller.loop = event_loop
437 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
438 1
        end1 = start1 + timedelta(hours=6)
439 1
        self.scheduler.get_maintenance.return_value = MW.construct(
440
            id = '1234',
441
            start = start1.replace(microsecond=0),
442
            end = end1.replace(microsecond=0),
443
            switches = [
444
                '00:00:00:00:00:00:12:23'
445
            ],
446
        )
447 1
        start_new = datetime.now(pytz.utc) + timedelta(days=1, hours=3)
448 1
        payload = {
449
            "start": start_new.strftime(TIME_FMT),
450
            "switches": [],
451
            'interfaces': [],
452
            'links': [],
453
        }
454
455 1
        url = f'{self.base_endpoint}/1234'
456 1
        response = await self.api.patch(url, json=payload)
457 1
        current_data = response.json()
458 1
        assert response.status_code == 400
459 1
        assert current_data['description'] == \
460
                         'At least one item must be provided'
461 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
462 1
        self.scheduler.update.assert_not_called()
463
464 1 View Code Duplication
    async def test_update_mw_case_7(self, event_loop):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
465
        """Test successful update."""
466 1
        self.napp.controller.loop = event_loop
467 1
        start1 = datetime.now(pytz.utc) + timedelta(days=1)
468 1
        end1 = start1 + timedelta(hours=6)
469 1
        self.scheduler.get_maintenance.return_value = MW.construct(
470
            id = '1234',
471
            start = start1.replace(microsecond=0),
472
            end = end1.replace(microsecond=0),
473
            switches = [
474
                '00:00:00:00:00:00:12:23'
475
            ],
476
        )
477 1
        payload = {
478
            'status': 'running',
479
        }
480 1
        url = f'{self.base_endpoint}/1234'
481 1
        response = await self.api.patch(url, json=payload)
482 1
        current_data = response.json()
483 1
        assert response.status_code == 400
484 1
        assert current_data['description'] == \
485
                         'Updating a maintenance status is not allowed'
486 1
        self.scheduler.update.assert_not_called()
487
488 1
    async def test_end_mw_case_1(self):
489
        """Test method that finishes the maintenance now."""
490 1
        self.scheduler.get_maintenance.return_value = None
491 1
        url = f'{self.base_endpoint}/2345/end'
492 1
        response = await self.api.patch(url)
493 1
        current_data = response.json()
494 1
        assert response.status_code == 404
495 1
        assert current_data['description'] == \
496
                         'Maintenance with id 2345 not found'
497
498 1 View Code Duplication
    async def test_end_mw_case_2(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
499
        """Test method that finishes the maintenance now."""
500 1
        start1 = datetime.now(pytz.utc) - timedelta(hours=1)
501 1
        end1 = start1 + timedelta(hours=6)
502 1
        self.scheduler.get_maintenance.return_value = MW.construct(
503
            id = '1234',
504
            start = start1.replace(microsecond=0),
505
            end = end1.replace(microsecond=0),
506
            switches = [
507
                '00:00:00:00:00:00:12:23'
508
            ],
509
            status = 'running',
510
        )
511 1
        url = f'{self.base_endpoint}/1234/end'
512 1
        response = await self.api.patch(url)
513 1
        current_data = response.json()
514 1
        self.scheduler.get.asssert_called_once_with('1234')
515 1
        self.scheduler.end_maintenance_early.assert_called_once_with('1234')
516 1
        assert response.status_code == 200
517 1
        assert current_data == \
518
                         {'response': 'Maintenance window 1234 finished'}
519
520 1 View Code Duplication
    async def test_end_mw_case_3(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
521
        """Test method that finishes the maintenance now."""
522 1
        start1 = datetime.now(pytz.utc) + timedelta(hours=1)
523 1
        end1 = start1 + timedelta(hours=6)
524 1
        self.scheduler.get_maintenance.return_value = MW.construct(
525
            id = '1234',
526
            start = start1.replace(microsecond=0),
527
            end = end1.replace(microsecond=0),
528
            switches = [
529
                '00:00:00:00:00:00:12:23'
530
            ],
531
            status = 'pending',
532
        )
533 1
        url = f'{self.base_endpoint}/1234/end'
534 1
        response = await self.api.patch(url)
535 1
        current_data = response.json()
536 1
        self.scheduler.get.asssert_called_once_with('1234')
537 1
        self.scheduler.end_maintenance_early.assert_not_called()
538 1
        assert response.status_code == 400
539 1
        assert current_data['description'] == \
540
                         'Maintenance window 1234 has not yet started'
541
542 1 View Code Duplication
    async def test_end_mw_case_4(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
543
        """Test method that finishes the maintenance now."""
544 1
        start1 = datetime.now(pytz.utc) - timedelta(hours=5)
545 1
        end1 = start1 + timedelta(hours=4)
546 1
        self.scheduler.get_maintenance.return_value = MW.construct(
547
            id = '1234',
548
            start = start1.replace(microsecond=0),
549
            end = end1.replace(microsecond=0),
550
            switches = [
551
                '00:00:00:00:00:00:12:23'
552
            ],
553
            status = 'finished',
554
        )
555 1
        url = f'{self.base_endpoint}/1234/end'
556 1
        response = await self.api.patch(url)
557 1
        current_data = response.json()
558 1
        self.scheduler.get.asssert_called_once_with('1234')
559 1
        self.scheduler.end_maintenance_early.assert_not_called()
560 1
        assert response.status_code == 400
561 1
        assert current_data['description'] == \
562
                         'Maintenance window 1234 has already finished'
563
564 1
    async def test_extend_case_1(self, event_loop):
565
        """Test successful extension."""
566 1
        self.napp.controller.loop = event_loop
567 1
        start1 = datetime.now(pytz.utc) - timedelta(hours=3)
568 1
        end1 = start1 + timedelta(hours=4)
569 1
        self.scheduler.get_maintenance.return_value = MW.construct(
570
            id = '1234',
571
            start = start1.replace(microsecond=0),
572
            end = end1.replace(microsecond=0),
573
            switches = [
574
                '00:00:00:00:00:00:12:23'
575
            ],
576
            status = 'running',
577
        )
578 1
        url = f'{self.base_endpoint}/1234/extend'
579 1
        payload = {
580
            'minutes': 45
581
        }
582 1
        response = await self.api.patch(url, json=payload)
583 1
        assert response.status_code == 200
584 1
        self.scheduler.get_maintenance.called_with('1234')
585 1
        self.scheduler.update.assert_called_with(MW.construct(
586
            id = '1234',
587
            start = start1.replace(microsecond=0),
588
            end = end1.replace(microsecond=0) + timedelta(minutes=45),
589
            switches = [
590
                '00:00:00:00:00:00:12:23'
591
            ],
592
            status = 'running',
593
        ))
594
595 1
    async def test_extend_case_2(self, event_loop):
596
        """Test no payload error."""
597 1
        self.napp.controller.loop = event_loop
598 1
        url = f'{self.base_endpoint}/1234/extend'
599 1
        response = await self.api.patch(url)
600 1
        assert response.status_code == 400
601 1
        current_data = response.json()
602 1
        assert 'Invalid json' in current_data['description']
603 1
        self.scheduler.update.assert_not_called()
604
605 1
    async def test_extend_case_3(self, event_loop):
606
        """Test payload without minutes."""
607 1
        self.napp.controller.loop = event_loop
608 1
        url = f'{self.base_endpoint}/1234/extend'
609 1
        payload = {
610
            'seconds': 240
611
        }
612 1
        response = await self.api.patch(url, json=payload)
613 1
        assert response.status_code == 400
614 1
        current_data = response.json()
615 1
        assert current_data['description'] == \
616
                         'Minutes of extension must be sent'
617 1
        self.scheduler.update.assert_not_called()
618
619 1
    async def test_extend_case_4(self, event_loop):
620
        """Test no integer extension minutes."""
621 1
        self.napp.controller.loop = event_loop
622 1
        url = f'{self.base_endpoint}/1234/extend'
623 1
        payload = {
624
            'minutes': '240'
625
        }
626 1
        response = await self.api.patch(url, json=payload)
627 1
        assert response.status_code == 400
628 1
        current_data = response.json()
629 1
        assert current_data['description'] == \
630
                         'Minutes of extension must be integer'
631 1
        self.scheduler.update.assert_not_called()
632
633 1 View Code Duplication
    async def test_extend_case_5(self, event_loop):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
634
        """Test maintenance did not start."""
635 1
        self.napp.controller.loop = event_loop
636 1
        start1 = datetime.now(pytz.utc) + timedelta(hours=3)
637 1
        end1 = start1 + timedelta(hours=4)
638 1
        self.scheduler.get_maintenance.return_value = MW.construct(
639
            id = '1234',
640
            start = start1.replace(microsecond=0),
641
            end = end1.replace(microsecond=0),
642
            switches = [
643
                '00:00:00:00:00:00:12:23'
644
            ],
645
            status = 'pending',
646
        )
647 1
        url = f'{self.base_endpoint}/1234/extend'
648 1
        payload = {
649
            'minutes': 240
650
        }
651 1
        response = await self.api.patch(url, json=payload)
652 1
        assert response.status_code == 400
653 1
        current_data = response.json()
654 1
        assert current_data['description'] == \
655
                         'Maintenance window 1234 has not yet started'
656 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
657 1
        self.scheduler.update.assert_not_called()
658
659 1 View Code Duplication
    async def test_extend_case_6(self, event_loop):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
660
        """Test maintenance already finished."""
661 1
        self.napp.controller.loop = event_loop
662 1
        start1 = datetime.now(pytz.utc) - timedelta(hours=3)
663 1
        end1 = start1 + timedelta(hours=2)
664 1
        self.scheduler.get_maintenance.return_value = MW.construct(
665
            id = '1234',
666
            start = start1.replace(microsecond=0),
667
            end = end1.replace(microsecond=0),
668
            switches = [
669
                '00:00:00:00:00:00:12:23'
670
            ],
671
            status = 'finished',
672
        )
673 1
        url = f'{self.base_endpoint}/1234/extend'
674 1
        payload = {
675
            'minutes': 240
676
        }
677 1
        response = await self.api.patch(url, json=payload)
678 1
        assert response.status_code == 400
679 1
        current_data = response.json()
680 1
        assert current_data['description'] == \
681
                         'Maintenance window 1234 has already finished'
682 1
        self.scheduler.get_maintenance.assert_called_once_with('1234')
683 1
        self.scheduler.update.assert_not_called()
684
685 1
    async def test_extend_case_7(self, event_loop):
686
        """Test no maintenace found."""
687 1
        self.napp.controller.loop = event_loop
688 1
        self.scheduler.get_maintenance.return_value = None
689 1
        url = f'{self.base_endpoint}/1235/extend'
690 1
        payload = {
691
            'minutes': 240
692
        }
693 1
        response = await self.api.patch(url, json=payload)
694 1
        assert response.status_code == 404
695 1
        current_data = response.json()
696 1
        assert current_data['description'] == \
697
                         'Maintenance with id 1235 not found'
698 1
        self.scheduler.get_maintenance.assert_called_once_with('1235')
699
        self.scheduler.update.assert_not_called()
700