|
@@ 172-206 (lines=35) @@
|
| 169 |
|
|
| 170 |
|
assert current_data == {"mw_id": "some_mw"} |
| 171 |
|
|
| 172 |
|
async def test_create_mw_case_4(self): |
| 173 |
|
"""Test a successful case of the REST to create.""" |
| 174 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 175 |
|
url = f"{self.base_endpoint}" |
| 176 |
|
start = datetime.now(pytz.utc) + timedelta(days=1) |
| 177 |
|
end = start + timedelta(hours=2) |
| 178 |
|
self.controller.links = MagicMock() |
| 179 |
|
links = {"cf0f4071be426b3f745027f5d22bc61f8312ae86293c9b28e7e66015607a9260": 1} |
| 180 |
|
self.controller.links.get.side_effect = links.get |
| 181 |
|
payload = { |
| 182 |
|
"start": start.strftime(TIME_FMT), |
| 183 |
|
"end": end.strftime(TIME_FMT), |
| 184 |
|
"links": [ |
| 185 |
|
"cf0f4071be426b3f745027f5d22bc61f8312ae86293c9b28e7e66015607a9260", |
| 186 |
|
], |
| 187 |
|
} |
| 188 |
|
response = await self.api.post(url, json=payload) |
| 189 |
|
current_data = response.json() |
| 190 |
|
assert response.status_code == 201, current_data |
| 191 |
|
self.controller.links.get.assert_called_with( |
| 192 |
|
"cf0f4071be426b3f745027f5d22bc61f8312ae86293c9b28e7e66015607a9260" |
| 193 |
|
) |
| 194 |
|
assert response.status_code == 201, current_data |
| 195 |
|
args, kwargs = self.scheduler.add.call_args |
| 196 |
|
window: MW = args[0] |
| 197 |
|
|
| 198 |
|
assert window.start == start.replace(microsecond=0) |
| 199 |
|
assert window.end == end.replace(microsecond=0) |
| 200 |
|
assert window.switches == [] |
| 201 |
|
assert window.interfaces == [] |
| 202 |
|
assert window.links == [ |
| 203 |
|
"cf0f4071be426b3f745027f5d22bc61f8312ae86293c9b28e7e66015607a9260" |
| 204 |
|
] |
| 205 |
|
|
| 206 |
|
assert current_data == {"mw_id": window.id} |
| 207 |
|
|
| 208 |
|
async def test_create_mw_case_5(self): |
| 209 |
|
"""Test a fail case of the REST to create a maintenance window.""" |
|
@@ 103-132 (lines=30) @@
|
| 100 |
|
|
| 101 |
|
assert current_data == {"mw_id": window.id} |
| 102 |
|
|
| 103 |
|
async def test_create_mw_case_2(self): |
| 104 |
|
"""Test a successful case of the REST to create.""" |
| 105 |
|
self.napp.controller.loop = asyncio.get_running_loop() |
| 106 |
|
url = f"{self.base_endpoint}" |
| 107 |
|
start = datetime.now(pytz.utc) + timedelta(days=1) |
| 108 |
|
end = start + timedelta(hours=2) |
| 109 |
|
self.controller.switches = MagicMock() |
| 110 |
|
switches = {"00:00:00:00:00:00:00:01": 1} |
| 111 |
|
self.controller.switches.get.side_effect = switches.get |
| 112 |
|
payload = { |
| 113 |
|
"start": start.strftime(TIME_FMT), |
| 114 |
|
"end": end.strftime(TIME_FMT), |
| 115 |
|
"switches": [ |
| 116 |
|
"00:00:00:00:00:00:00:01", |
| 117 |
|
], |
| 118 |
|
} |
| 119 |
|
response = await self.api.post(url, json=payload) |
| 120 |
|
current_data = response.json() |
| 121 |
|
assert response.status_code == 201, current_data |
| 122 |
|
self.controller.switches.get.assert_called_with("00:00:00:00:00:00:00:01") |
| 123 |
|
args, kwargs = self.scheduler.add.call_args |
| 124 |
|
window: MW = args[0] |
| 125 |
|
|
| 126 |
|
assert window.start == start.replace(microsecond=0) |
| 127 |
|
assert window.end == end.replace(microsecond=0) |
| 128 |
|
assert window.switches == ["00:00:00:00:00:00:00:01"] |
| 129 |
|
assert window.interfaces == [] |
| 130 |
|
assert window.links == [] |
| 131 |
|
|
| 132 |
|
assert current_data == {"mw_id": window.id} |
| 133 |
|
|
| 134 |
|
async def test_create_mw_case_3(self): |
| 135 |
|
"""Test a successful case of the REST to create.""" |
kytos-ng /
maintenance
