echo packets with invalid checksum don't keep a session up
(self)
| 1916 | self.assertTrue(timeout_ok, "Expected timeout event didn't occur") |
| 1917 | |
| 1918 | def test_invalid_echo_checksum(self): |
| 1919 | """echo packets with invalid checksum don't keep a session up""" |
| 1920 | |
| 1921 | if self.multihop: |
| 1922 | self.skipTest( |
| 1923 | f"Skipping because echo functionality is not supported with multihop" |
| 1924 | ) |
| 1925 | |
| 1926 | bfd_session_up(self) |
| 1927 | self.test_session.update(required_min_echo_rx=150000) |
| 1928 | self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index) |
| 1929 | self.test_session.send_packet() |
| 1930 | # should be turned on - loopback echo packets |
| 1931 | timeout_at = None |
| 1932 | timeout_ok = False |
| 1933 | for dummy in range(10 * self.vpp_session.detect_mult): |
| 1934 | p = self.pg0.wait_for_packet(1) |
| 1935 | if p[UDP].dport == BFD.udp_dport_echo: |
| 1936 | self.logger.debug(ppp("Got echo packet:", p)) |
| 1937 | if timeout_at is None: |
| 1938 | timeout_at = ( |
| 1939 | time.time() |
| 1940 | + self.vpp_session.detect_mult |
| 1941 | * self.test_session.required_min_echo_rx |
| 1942 | / USEC_IN_SEC |
| 1943 | ) |
| 1944 | p[BFD_vpp_echo].checksum = getrandbits(64) |
| 1945 | p[Ether].dst = self.pg0.local_mac |
| 1946 | self.logger.debug(ppp("Looping back modified echo packet:", p)) |
| 1947 | self.pg0.add_stream(p) |
| 1948 | self.pg_start() |
| 1949 | elif p.haslayer(BFD): |
| 1950 | self.logger.debug(ppp("Got packet:", p)) |
| 1951 | if "P" in p.sprintf("%BFD.flags%"): |
| 1952 | final = self.test_session.create_packet() |
| 1953 | final[BFD].flags = "F" |
| 1954 | self.test_session.send_packet(final) |
| 1955 | if p[BFD].state == BFDState.down: |
| 1956 | self.assertIsNotNone( |
| 1957 | timeout_at, |
| 1958 | "Session went down before first echo packet received", |
| 1959 | ) |
| 1960 | now = time.time() |
| 1961 | self.assertGreaterEqual( |
| 1962 | now, |
| 1963 | timeout_at, |
| 1964 | "Session timeout at %s, but is expected at %s" |
| 1965 | % (now, timeout_at), |
| 1966 | ) |
| 1967 | self.assert_equal( |
| 1968 | p[BFD].diag, BFDDiagCode.echo_function_failed, BFDDiagCode |
| 1969 | ) |
| 1970 | events = self.vapi.collect_events() |
| 1971 | self.assert_equal(len(events), 1, "number of bfd events") |
| 1972 | self.assert_equal(events[0].state, BFDState.down, BFDState) |
| 1973 | timeout_ok = True |
| 1974 | break |
| 1975 | else: |
nothing calls this directly
no test coverage detected