Verify captured input packet stream for defined interface. :param object pg_if: Interface to verify captured packet stream for. :param list capture: Captured packet stream. :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
(self, pg_if, capture, traffic_type=0, ip_type=0, etype=-1)
| 341 | return pkts |
| 342 | |
| 343 | def verify_capture(self, pg_if, capture, traffic_type=0, ip_type=0, etype=-1): |
| 344 | """ |
| 345 | Verify captured input packet stream for defined interface. |
| 346 | |
| 347 | :param object pg_if: Interface to verify captured packet stream for. |
| 348 | :param list capture: Captured packet stream. |
| 349 | :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise. |
| 350 | """ |
| 351 | last_info = dict() |
| 352 | for i in self.pg_interfaces: |
| 353 | last_info[i.sw_if_index] = None |
| 354 | dst_sw_if_index = pg_if.sw_if_index |
| 355 | for packet in capture: |
| 356 | if etype > 0: |
| 357 | if packet[Ether].type != etype: |
| 358 | self.logger.error(ppp("Unexpected ethertype in packet:", packet)) |
| 359 | else: |
| 360 | continue |
| 361 | try: |
| 362 | # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data |
| 363 | if traffic_type == self.ICMP and ip_type == self.IPV6: |
| 364 | payload_info = self.payload_to_info( |
| 365 | packet[ICMPv6EchoRequest], "data" |
| 366 | ) |
| 367 | payload = packet[ICMPv6EchoRequest] |
| 368 | else: |
| 369 | payload_info = self.payload_to_info(packet[Raw]) |
| 370 | payload = packet[self.proto_map[payload_info.proto]] |
| 371 | except: |
| 372 | self.logger.error( |
| 373 | ppp("Unexpected or invalid packet (outside network):", packet) |
| 374 | ) |
| 375 | raise |
| 376 | |
| 377 | if ip_type != 0: |
| 378 | self.assertEqual(payload_info.ip, ip_type) |
| 379 | if traffic_type == self.ICMP: |
| 380 | try: |
| 381 | if payload_info.ip == 0: |
| 382 | self.assertEqual(payload.type, self.icmp4_type) |
| 383 | self.assertEqual(payload.code, self.icmp4_code) |
| 384 | else: |
| 385 | self.assertEqual(payload.type, self.icmp6_type) |
| 386 | self.assertEqual(payload.code, self.icmp6_code) |
| 387 | except: |
| 388 | self.logger.error( |
| 389 | ppp("Unexpected or invalid packet (outside network):", packet) |
| 390 | ) |
| 391 | raise |
| 392 | else: |
| 393 | try: |
| 394 | ip_version = IPv6 if payload_info.ip == 1 else IP |
| 395 | |
| 396 | ip = packet[ip_version] |
| 397 | packet_index = payload_info.index |
| 398 | |
| 399 | self.assertEqual(payload_info.dst, dst_sw_if_index) |
| 400 | self.logger.debug( |
no test coverage detected