echo function
(self)
| 2338 | ) |
| 2339 | |
| 2340 | def test_echo(self): |
| 2341 | """echo function""" |
| 2342 | |
| 2343 | if self.multihop: |
| 2344 | self.skipTest( |
| 2345 | f"Skipping because echo functionality is not supported with multihop" |
| 2346 | ) |
| 2347 | |
| 2348 | stats_before = bfd_grab_stats_snapshot(self) |
| 2349 | bfd_session_up(self) |
| 2350 | self.test_session.update(required_min_echo_rx=150000) |
| 2351 | self.test_session.send_packet() |
| 2352 | detection_time = ( |
| 2353 | self.test_session.detect_mult |
| 2354 | * self.vpp_session.required_min_rx |
| 2355 | / USEC_IN_SEC |
| 2356 | ) |
| 2357 | # echo shouldn't work without echo source set |
| 2358 | for dummy in range(10): |
| 2359 | sleep = self.vpp_session.required_min_rx / USEC_IN_SEC |
| 2360 | self.sleep(sleep, "delay before sending bfd packet") |
| 2361 | self.test_session.send_packet() |
| 2362 | p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset) |
| 2363 | self.assert_equal( |
| 2364 | p[BFD].required_min_rx_interval, |
| 2365 | self.vpp_session.required_min_rx, |
| 2366 | "BFD required min rx interval", |
| 2367 | ) |
| 2368 | self.test_session.send_packet() |
| 2369 | self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index) |
| 2370 | echo_seen = False |
| 2371 | # should be turned on - loopback echo packets |
| 2372 | for dummy in range(3): |
| 2373 | loop_until = time.time() + 0.75 * detection_time |
| 2374 | while time.time() < loop_until: |
| 2375 | p = self.pg0.wait_for_packet(1) |
| 2376 | self.logger.debug(ppp("Got packet:", p)) |
| 2377 | if p[UDP].dport == BFD.udp_dport_echo: |
| 2378 | self.assert_equal( |
| 2379 | p[IPv6].dst, self.pg0.local_ip6, "BFD ECHO dst IP" |
| 2380 | ) |
| 2381 | self.assertNotEqual( |
| 2382 | p[IPv6].src, |
| 2383 | self.loopback0.local_ip6, |
| 2384 | "BFD ECHO src IP equal to loopback IP", |
| 2385 | ) |
| 2386 | self.logger.debug(ppp("Looping back packet:", p)) |
| 2387 | self.assert_equal( |
| 2388 | p[Ether].dst, |
| 2389 | self.pg0.remote_mac, |
| 2390 | "ECHO packet destination MAC address", |
| 2391 | ) |
| 2392 | self.test_session.rx_packets_echo += 1 |
| 2393 | self.test_session.tx_packets_echo += 1 |
| 2394 | p[Ether].dst = self.pg0.local_mac |
| 2395 | self.pg0.add_stream(p) |
| 2396 | self.pg_start() |
| 2397 | echo_seen = True |
nothing calls this directly
no test coverage detected