kytos-ng /
telemetry_int
| @@ 472-494 (lines=23) @@ | ||
| 469 | assert pp.evc_ids.add.call_count == 2 |
|
| 470 | pp.evc_ids.add.assert_called_with(evc_id) |
|
| 471 | ||
| 472 | def test__discard_pps_evc_ids(self): |
|
| 473 | """test_discard_pps_evc_ids.""" |
|
| 474 | dpid_a = "00:00:00:00:00:00:00:01" |
|
| 475 | intf_id_a = f"{dpid_a}:1" |
|
| 476 | intf_id_z = f"{dpid_a}:2" |
|
| 477 | evc_id = "3766c105686749" |
|
| 478 | evcs = { |
|
| 479 | evc_id: { |
|
| 480 | "metadata": {"telemetry": {"enabled": True}}, |
|
| 481 | "uni_a": {"interface_id": intf_id_a}, |
|
| 482 | "uni_z": {"interface_id": intf_id_z}, |
|
| 483 | } |
|
| 484 | } |
|
| 485 | controller = get_controller_mock() |
|
| 486 | int_manager = INTManager(controller) |
|
| 487 | pp = MagicMock() |
|
| 488 | mock = MagicMock() |
|
| 489 | int_manager.get_proxy_port_or_raise = mock |
|
| 490 | mock.return_value = pp |
|
| 491 | int_manager._discard_pps_evc_ids(evcs) |
|
| 492 | assert int_manager.get_proxy_port_or_raise.call_count == 2 |
|
| 493 | assert pp.evc_ids.discard.call_count == 2 |
|
| 494 | pp.evc_ids.discard.assert_called_with(evc_id) |
|
| 495 | ||
| @@ 448-470 (lines=23) @@ | ||
| 445 | assert controller._buffers.app.aput.call_count == 3 |
|
| 446 | assert sleep_mock.call_count == 0 |
|
| 447 | ||
| 448 | def test__add_pps_evc_ids(self): |
|
| 449 | """test_add_pps_evc_ids.""" |
|
| 450 | dpid_a = "00:00:00:00:00:00:00:01" |
|
| 451 | intf_id_a = f"{dpid_a}:1" |
|
| 452 | intf_id_z = f"{dpid_a}:2" |
|
| 453 | evc_id = "3766c105686749" |
|
| 454 | evcs = { |
|
| 455 | evc_id: { |
|
| 456 | "metadata": {"telemetry": {"enabled": True}}, |
|
| 457 | "uni_a": {"interface_id": intf_id_a}, |
|
| 458 | "uni_z": {"interface_id": intf_id_z}, |
|
| 459 | } |
|
| 460 | } |
|
| 461 | controller = get_controller_mock() |
|
| 462 | int_manager = INTManager(controller) |
|
| 463 | pp = MagicMock() |
|
| 464 | mock = MagicMock() |
|
| 465 | int_manager.get_proxy_port_or_raise = mock |
|
| 466 | mock.return_value = pp |
|
| 467 | int_manager._add_pps_evc_ids(evcs) |
|
| 468 | assert int_manager.get_proxy_port_or_raise.call_count == 2 |
|
| 469 | assert pp.evc_ids.add.call_count == 2 |
|
| 470 | pp.evc_ids.add.assert_called_with(evc_id) |
|
| 471 | ||
| 472 | def test__discard_pps_evc_ids(self): |
|
| 473 | """test_discard_pps_evc_ids.""" |
|
