kytos-ng /
telemetry_int
| @@ 507-529 (lines=23) @@ | ||
| 504 | assert pp.evc_ids.add.call_count == 2 |
|
| 505 | pp.evc_ids.add.assert_called_with(evc_id) |
|
| 506 | ||
| 507 | def test__discard_pps_evc_ids(self): |
|
| 508 | """test_discard_pps_evc_ids.""" |
|
| 509 | dpid_a = "00:00:00:00:00:00:00:01" |
|
| 510 | intf_id_a = f"{dpid_a}:1" |
|
| 511 | intf_id_z = f"{dpid_a}:2" |
|
| 512 | evc_id = "3766c105686749" |
|
| 513 | evcs = { |
|
| 514 | evc_id: { |
|
| 515 | "metadata": {"telemetry": {"enabled": True}}, |
|
| 516 | "uni_a": {"interface_id": intf_id_a}, |
|
| 517 | "uni_z": {"interface_id": intf_id_z}, |
|
| 518 | } |
|
| 519 | } |
|
| 520 | controller = get_controller_mock() |
|
| 521 | int_manager = INTManager(controller) |
|
| 522 | pp = MagicMock() |
|
| 523 | mock = MagicMock() |
|
| 524 | int_manager.get_proxy_port_or_raise = mock |
|
| 525 | mock.return_value = pp |
|
| 526 | int_manager._discard_pps_evc_ids(evcs) |
|
| 527 | assert int_manager.get_proxy_port_or_raise.call_count == 2 |
|
| 528 | assert pp.evc_ids.discard.call_count == 2 |
|
| 529 | pp.evc_ids.discard.assert_called_with(evc_id) |
|
| 530 | ||
| 531 | def test_validate_evc_stored_flows(self) -> None: |
|
| 532 | """Test validate evc stored flows.""" |
|
| @@ 483-505 (lines=23) @@ | ||
| 480 | assert controller._buffers.app.aput.call_count == 3 |
|
| 481 | assert sleep_mock.call_count == 0 |
|
| 482 | ||
| 483 | def test__add_pps_evc_ids(self): |
|
| 484 | """test_add_pps_evc_ids.""" |
|
| 485 | dpid_a = "00:00:00:00:00:00:00:01" |
|
| 486 | intf_id_a = f"{dpid_a}:1" |
|
| 487 | intf_id_z = f"{dpid_a}:2" |
|
| 488 | evc_id = "3766c105686749" |
|
| 489 | evcs = { |
|
| 490 | evc_id: { |
|
| 491 | "metadata": {"telemetry": {"enabled": True}}, |
|
| 492 | "uni_a": {"interface_id": intf_id_a}, |
|
| 493 | "uni_z": {"interface_id": intf_id_z}, |
|
| 494 | } |
|
| 495 | } |
|
| 496 | controller = get_controller_mock() |
|
| 497 | int_manager = INTManager(controller) |
|
| 498 | pp = MagicMock() |
|
| 499 | mock = MagicMock() |
|
| 500 | int_manager.get_proxy_port_or_raise = mock |
|
| 501 | mock.return_value = pp |
|
| 502 | int_manager._add_pps_evc_ids(evcs) |
|
| 503 | assert int_manager.get_proxy_port_or_raise.call_count == 2 |
|
| 504 | assert pp.evc_ids.add.call_count == 2 |
|
| 505 | pp.evc_ids.add.assert_called_with(evc_id) |
|
| 506 | ||
| 507 | def test__discard_pps_evc_ids(self): |
|
| 508 | """test_discard_pps_evc_ids.""" |
|
