|
@@ 578-601 (lines=24) @@
|
| 575 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 576 |
|
self.scheduler.update.assert_not_called() |
| 577 |
|
|
| 578 |
|
async def test_update_mw_case_5(self): |
| 579 |
|
"""Test successful update.""" |
| 580 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 581 |
|
start1 = datetime.now(pytz.utc) + timedelta(days=1) |
| 582 |
|
end1 = start1 + timedelta(hours=6) |
| 583 |
|
self.scheduler.get_maintenance.return_value = MW.model_construct( |
| 584 |
|
id="1234", |
| 585 |
|
start=start1.replace(microsecond=0), |
| 586 |
|
end=end1.replace(microsecond=0), |
| 587 |
|
switches=["00:00:00:00:00:00:12:23"], |
| 588 |
|
) |
| 589 |
|
start_new = datetime.now(pytz.utc) + timedelta(days=1, hours=3) |
| 590 |
|
end_new = start_new - timedelta(hours=5) |
| 591 |
|
payload = { |
| 592 |
|
"start": start_new.strftime(TIME_FMT), |
| 593 |
|
"end": end_new.strftime(TIME_FMT), |
| 594 |
|
} |
| 595 |
|
url = f"{self.base_endpoint}/1234" |
| 596 |
|
response = await self.api.patch(url, json=payload) |
| 597 |
|
current_data = response.json() |
| 598 |
|
assert response.status_code == 400 |
| 599 |
|
assert current_data["description"] == "end: Value error, End before start not allowed" |
| 600 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 601 |
|
self.scheduler.update.assert_not_called() |
| 602 |
|
|
| 603 |
|
async def test_update_mw_case_6(self): |
| 604 |
|
"""Test successful update.""" |
|
@@ 810-832 (lines=23) @@
|
| 807 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 808 |
|
self.scheduler.update.assert_not_called() |
| 809 |
|
|
| 810 |
|
async def test_extend_case_6(self): |
| 811 |
|
"""Test maintenance already finished.""" |
| 812 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 813 |
|
start1 = datetime.now(pytz.utc) - timedelta(hours=3) |
| 814 |
|
end1 = start1 + timedelta(hours=2) |
| 815 |
|
self.scheduler.get_maintenance.return_value = MW.model_construct( |
| 816 |
|
id="1234", |
| 817 |
|
start=start1.replace(microsecond=0), |
| 818 |
|
end=end1.replace(microsecond=0), |
| 819 |
|
switches=["00:00:00:00:00:00:12:23"], |
| 820 |
|
status="finished", |
| 821 |
|
) |
| 822 |
|
url = f"{self.base_endpoint}/1234/extend" |
| 823 |
|
payload = {"minutes": 240} |
| 824 |
|
response = await self.api.patch(url, json=payload) |
| 825 |
|
assert response.status_code == 400 |
| 826 |
|
current_data = response.json() |
| 827 |
|
assert ( |
| 828 |
|
current_data["description"] |
| 829 |
|
== "Maintenance window 1234 has already finished" |
| 830 |
|
) |
| 831 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 832 |
|
self.scheduler.update.assert_not_called() |
| 833 |
|
|
| 834 |
|
async def test_extend_case_7(self): |
| 835 |
|
"""Test no maintenace found.""" |
|
@@ 787-808 (lines=22) @@
|
| 784 |
|
assert current_data["description"] == "Minutes of extension must be integer" |
| 785 |
|
self.scheduler.update.assert_not_called() |
| 786 |
|
|
| 787 |
|
async def test_extend_case_5(self): |
| 788 |
|
"""Test maintenance did not start.""" |
| 789 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 790 |
|
start1 = datetime.now(pytz.utc) + timedelta(hours=3) |
| 791 |
|
end1 = start1 + timedelta(hours=4) |
| 792 |
|
self.scheduler.get_maintenance.return_value = MW.model_construct( |
| 793 |
|
id="1234", |
| 794 |
|
start=start1.replace(microsecond=0), |
| 795 |
|
end=end1.replace(microsecond=0), |
| 796 |
|
switches=["00:00:00:00:00:00:12:23"], |
| 797 |
|
status="pending", |
| 798 |
|
) |
| 799 |
|
url = f"{self.base_endpoint}/1234/extend" |
| 800 |
|
payload = {"minutes": 240} |
| 801 |
|
response = await self.api.patch(url, json=payload) |
| 802 |
|
assert response.status_code == 400 |
| 803 |
|
current_data = response.json() |
| 804 |
|
assert ( |
| 805 |
|
current_data["description"] == "Maintenance window 1234 has not yet started" |
| 806 |
|
) |
| 807 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 808 |
|
self.scheduler.update.assert_not_called() |
| 809 |
|
|
| 810 |
|
async def test_extend_case_6(self): |
| 811 |
|
"""Test maintenance already finished.""" |
|
@@ 555-576 (lines=22) @@
|
| 552 |
|
) |
| 553 |
|
) |
| 554 |
|
|
| 555 |
|
async def test_update_mw_case_4(self): |
| 556 |
|
"""Test successful update.""" |
| 557 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 558 |
|
start1 = datetime.now(pytz.utc) + timedelta(days=1) |
| 559 |
|
end1 = start1 + timedelta(hours=6) |
| 560 |
|
self.scheduler.get_maintenance.return_value = MW.model_construct( |
| 561 |
|
id="1234", |
| 562 |
|
start=start1.replace(microsecond=0), |
| 563 |
|
end=end1.replace(microsecond=0), |
| 564 |
|
switches=["00:00:00:00:00:00:12:23"], |
| 565 |
|
) |
| 566 |
|
start_new = datetime.now(pytz.utc) - timedelta(days=1, hours=3) |
| 567 |
|
payload = { |
| 568 |
|
"start": start_new.strftime(TIME_FMT), |
| 569 |
|
} |
| 570 |
|
url = f"{self.base_endpoint}/1234" |
| 571 |
|
response = await self.api.patch(url, json=payload) |
| 572 |
|
current_data = response.json() |
| 573 |
|
assert response.status_code == 400 |
| 574 |
|
assert current_data["description"] == "start: Value error, Start in the past not allowed" |
| 575 |
|
self.scheduler.get_maintenance.assert_called_once_with("1234") |
| 576 |
|
self.scheduler.update.assert_not_called() |
| 577 |
|
|
| 578 |
|
async def test_update_mw_case_5(self): |
| 579 |
|
"""Test successful update.""" |
kytos-ng /
maintenance
