(self, src_if, dst_if, capture)
| 53 | return packets |
| 54 | |
| 55 | def verify_capture(self, src_if, dst_if, capture): |
| 56 | packet_info = None |
| 57 | for packet in capture: |
| 58 | try: |
| 59 | ip = packet[IP] |
| 60 | udp = packet[UDP] |
| 61 | self.logger.debug(f"Converting payload to info for {packet[Raw]}") |
| 62 | # convert the payload to packet info object |
| 63 | payload_info = self.payload_to_info(packet[Raw]) |
| 64 | # make sure the indexes match |
| 65 | self.assert_equal( |
| 66 | payload_info.src, src_if.sw_if_index, "source sw_if_index" |
| 67 | ) |
| 68 | self.assert_equal( |
| 69 | payload_info.dst, dst_if.sw_if_index, "destination sw_if_index" |
| 70 | ) |
| 71 | packet_info = self.get_next_packet_info_for_interface2( |
| 72 | src_if.sw_if_index, dst_if.sw_if_index, packet_info |
| 73 | ) |
| 74 | # make sure we didn't run out of saved packets |
| 75 | self.assertIsNotNone(packet_info) |
| 76 | self.assert_equal( |
| 77 | payload_info.index, packet_info.index, "packet info index" |
| 78 | ) |
| 79 | saved_packet = packet_info.data # fetch the saved packet |
| 80 | # assert the values match |
| 81 | self.assert_equal(ip.src, saved_packet[IP].src, "IP source address") |
| 82 | # ... more assertions here |
| 83 | self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port") |
| 84 | except Exception: |
| 85 | self.logger.error(ppp("Unexpected or invalid packet:", packet)) |
| 86 | raise |
| 87 | remaining_packet = self.get_next_packet_info_for_interface2( |
| 88 | src_if.sw_if_index, dst_if.sw_if_index, packet_info |
| 89 | ) |
| 90 | self.assertIsNone( |
| 91 | remaining_packet, |
| 92 | "Interface %s: Packet expected from interface " |
| 93 | "%s didn't arrive" % (dst_if.name, src_if.name), |
| 94 | ) |
| 95 | |
| 96 | def test_mdata_cli(self): |
| 97 | """turn on mdata tracking, send packets, verify, check CLI output""" |
no test coverage detected