echo function
(self)
| 1634 | ) |
| 1635 | |
| 1636 | def test_echo(self): |
| 1637 | """echo function""" |
| 1638 | |
| 1639 | if self.multihop: |
| 1640 | self.skipTest( |
| 1641 | f"Skipping because echo functionality is not supported with multihop" |
| 1642 | ) |
| 1643 | |
| 1644 | stats_before = bfd_grab_stats_snapshot(self) |
| 1645 | bfd_session_up(self) |
| 1646 | self.test_session.update(required_min_echo_rx=150000) |
| 1647 | self.test_session.send_packet() |
| 1648 | detection_time = ( |
| 1649 | self.test_session.detect_mult |
| 1650 | * self.vpp_session.required_min_rx |
| 1651 | / USEC_IN_SEC |
| 1652 | ) |
| 1653 | # echo shouldn't work without echo source set |
| 1654 | for dummy in range(10): |
| 1655 | sleep = self.vpp_session.required_min_rx / USEC_IN_SEC |
| 1656 | self.sleep(sleep, "delay before sending bfd packet") |
| 1657 | self.test_session.send_packet() |
| 1658 | p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset) |
| 1659 | self.assert_equal( |
| 1660 | p[BFD].required_min_rx_interval, |
| 1661 | self.vpp_session.required_min_rx, |
| 1662 | "BFD required min rx interval", |
| 1663 | ) |
| 1664 | self.test_session.send_packet() |
| 1665 | self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index) |
| 1666 | echo_seen = False |
| 1667 | # should be turned on - loopback echo packets |
| 1668 | for dummy in range(3): |
| 1669 | loop_until = time.time() + 0.75 * detection_time |
| 1670 | while time.time() < loop_until: |
| 1671 | p = self.pg0.wait_for_packet(1) |
| 1672 | self.logger.debug(ppp("Got packet:", p)) |
| 1673 | if p[UDP].dport == BFD.udp_dport_echo: |
| 1674 | self.assert_equal(p[IP].dst, self.pg0.local_ip4, "BFD ECHO dst IP") |
| 1675 | self.assertNotEqual( |
| 1676 | p[IP].src, |
| 1677 | self.loopback0.local_ip4, |
| 1678 | "BFD ECHO src IP equal to loopback IP", |
| 1679 | ) |
| 1680 | self.logger.debug(ppp("Looping back packet:", p)) |
| 1681 | self.assert_equal( |
| 1682 | p[Ether].dst, |
| 1683 | self.pg0.remote_mac, |
| 1684 | "ECHO packet destination MAC address", |
| 1685 | ) |
| 1686 | p[Ether].dst = self.pg0.local_mac |
| 1687 | self.pg0.add_stream(p) |
| 1688 | self.test_session.rx_packets_echo += 1 |
| 1689 | self.test_session.tx_packets_echo += 1 |
| 1690 | self.pg_start() |
| 1691 | echo_seen = True |
| 1692 | elif p.haslayer(BFD): |
| 1693 | self.test_session.rx_packets += 1 |
nothing calls this directly
no test coverage detected