(self, src_if, dst_if, count)
| 53 | self.assertEqual(ret.drop_M, 1) |
| 54 | |
| 55 | def create_stream(self, src_if, dst_if, count): |
| 56 | packets = [] |
| 57 | for i in range(count): |
| 58 | # create packet info stored in the test case instance |
| 59 | info = self.create_packet_info(src_if, dst_if) |
| 60 | # convert the info into packet payload |
| 61 | payload = self.info_to_payload(info) |
| 62 | # create the packet itself |
| 63 | p = ( |
| 64 | Ether(dst=src_if.local_mac, src=src_if.remote_mac) |
| 65 | / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4, ttl=i + 1) |
| 66 | / UDP(sport=randint(49152, 65535), dport=5678) |
| 67 | / Raw(payload) |
| 68 | ) |
| 69 | # store a copy of the packet in the packet info |
| 70 | info.data = p.copy() |
| 71 | # append the packet to the list |
| 72 | packets.append(p) |
| 73 | # return the created packet list |
| 74 | return packets |
| 75 | |
| 76 | def get_sflow_counter(self, counter): |
| 77 | counters = self.vapi.cli("sh errors").split("\n") |
no test coverage detected