|
@@ 24-45 (lines=22) @@
|
| 21 |
|
self.api_client = get_test_client(controller, self.napp) |
| 22 |
|
self.base_endpoint = "kytos/telemetry_int/v1" |
| 23 |
|
|
| 24 |
|
async def test_enable_telemetry(self, monkeypatch) -> None: |
| 25 |
|
"""Test enable telemetry.""" |
| 26 |
|
api_mock, flow = AsyncMock(), MagicMock() |
| 27 |
|
flow.cookie = 0xA800000000000001 |
| 28 |
|
monkeypatch.setattr( |
| 29 |
|
"napps.kytos.telemetry_int.main.api", |
| 30 |
|
api_mock, |
| 31 |
|
) |
| 32 |
|
|
| 33 |
|
evc_id = utils.get_id_from_cookie(flow.cookie) |
| 34 |
|
api_mock.get_evcs.return_value = { |
| 35 |
|
evc_id: {"metadata": {"telemetry": {"enabled": False}}} |
| 36 |
|
} |
| 37 |
|
|
| 38 |
|
self.napp.int_manager = AsyncMock() |
| 39 |
|
|
| 40 |
|
endpoint = f"{self.base_endpoint}/evc/enable" |
| 41 |
|
response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]}) |
| 42 |
|
assert self.napp.int_manager.enable_int.call_count == 1 |
| 43 |
|
assert self.napp.int_manager._remove_int_flows.call_count == 1 |
| 44 |
|
assert response.status_code == 201 |
| 45 |
|
assert response.json() == [evc_id] |
| 46 |
|
|
| 47 |
|
async def test_redeploy_telemetry_enabled(self, monkeypatch) -> None: |
| 48 |
|
"""Test redeploy telemetry enabled.""" |
|
@@ 126-146 (lines=21) @@
|
| 123 |
|
assert response.status_code == 200 |
| 124 |
|
assert response.json() == [evc_id] |
| 125 |
|
|
| 126 |
|
async def test_get_enabled_evcs(self, monkeypatch) -> None: |
| 127 |
|
"""Test get enabled evcs.""" |
| 128 |
|
api_mock, flow = AsyncMock(), MagicMock() |
| 129 |
|
flow.cookie = 0xA800000000000001 |
| 130 |
|
monkeypatch.setattr( |
| 131 |
|
"napps.kytos.telemetry_int.main.api", |
| 132 |
|
api_mock, |
| 133 |
|
) |
| 134 |
|
|
| 135 |
|
evc_id = utils.get_id_from_cookie(flow.cookie) |
| 136 |
|
api_mock.get_evcs.return_value = { |
| 137 |
|
evc_id: {"metadata": {"telemetry": {"enabled": True}}}, |
| 138 |
|
} |
| 139 |
|
|
| 140 |
|
endpoint = f"{self.base_endpoint}/evc" |
| 141 |
|
response = await self.api_client.get(endpoint) |
| 142 |
|
assert api_mock.get_evcs.call_args[1] == {"metadata.telemetry.enabled": "true"} |
| 143 |
|
assert response.status_code == 200 |
| 144 |
|
data = response.json() |
| 145 |
|
assert len(data) == 1 |
| 146 |
|
assert evc_id in data |
| 147 |
|
|
| 148 |
|
async def test_get_evc_compare(self, monkeypatch) -> None: |
| 149 |
|
"""Test get evc compre ok case.""" |