|
@@ 118-138 (lines=21) @@
|
| 115 |
|
assert response.status_code == 400 |
| 116 |
|
assert "evc_ids" in response.json()["description"] |
| 117 |
|
|
| 118 |
|
async def test_disable_telemetry(self, monkeypatch) -> None: |
| 119 |
|
"""Test disable telemetry.""" |
| 120 |
|
api_mock, flow = AsyncMock(), MagicMock() |
| 121 |
|
flow.cookie = 0xA800000000000001 |
| 122 |
|
monkeypatch.setattr( |
| 123 |
|
"napps.kytos.telemetry_int.main.api", |
| 124 |
|
api_mock, |
| 125 |
|
) |
| 126 |
|
|
| 127 |
|
evc_id = utils.get_id_from_cookie(flow.cookie) |
| 128 |
|
api_mock.get_evcs.return_value = { |
| 129 |
|
evc_id: {"metadata": {"telemetry": {"enabled": True}}} |
| 130 |
|
} |
| 131 |
|
|
| 132 |
|
self.napp.int_manager = AsyncMock() |
| 133 |
|
|
| 134 |
|
endpoint = f"{self.base_endpoint}/evc/disable" |
| 135 |
|
response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]}) |
| 136 |
|
assert self.napp.int_manager.disable_int.call_count == 1 |
| 137 |
|
assert response.status_code == 200 |
| 138 |
|
assert response.json() == [evc_id] |
| 139 |
|
|
| 140 |
|
async def test_get_enabled_evcs(self, monkeypatch) -> None: |
| 141 |
|
"""Test get enabled evcs.""" |
|
@@ 61-81 (lines=21) @@
|
| 58 |
|
assert response.status_code == 201 |
| 59 |
|
assert response.json() == [evc_id] |
| 60 |
|
|
| 61 |
|
async def test_redeploy_telemetry_enabled(self, monkeypatch) -> None: |
| 62 |
|
"""Test redeploy telemetry enabled.""" |
| 63 |
|
api_mock, flow = AsyncMock(), MagicMock() |
| 64 |
|
flow.cookie = 0xA800000000000001 |
| 65 |
|
monkeypatch.setattr( |
| 66 |
|
"napps.kytos.telemetry_int.main.api", |
| 67 |
|
api_mock, |
| 68 |
|
) |
| 69 |
|
|
| 70 |
|
evc_id = utils.get_id_from_cookie(flow.cookie) |
| 71 |
|
api_mock.get_evc.return_value = { |
| 72 |
|
evc_id: {"metadata": {"telemetry": {"enabled": True}}} |
| 73 |
|
} |
| 74 |
|
|
| 75 |
|
self.napp.int_manager = AsyncMock() |
| 76 |
|
|
| 77 |
|
endpoint = f"{self.base_endpoint}/evc/redeploy" |
| 78 |
|
response = await self.api_client.patch(endpoint, json={"evc_ids": [evc_id]}) |
| 79 |
|
assert self.napp.int_manager.redeploy_int.call_count == 1 |
| 80 |
|
assert response.status_code == 201 |
| 81 |
|
assert response.json() == [evc_id] |
| 82 |
|
|
| 83 |
|
async def test_redeploy_telemetry_not_enabled(self, monkeypatch) -> None: |
| 84 |
|
"""Test redeploy telemetry not enabled.""" |
kytos-ng /
telemetry_int
