(self, src_if, dst_if, count)
| 12 | |
| 13 | |
| 14 | def create_stream(self, src_if, dst_if, count): |
| 15 | packets = [] |
| 16 | for i in range(count): |
| 17 | # create packet info stored in the test case instance |
| 18 | info = self.create_packet_info(src_if, dst_if) |
| 19 | # convert the info into packet payload |
| 20 | payload = self.info_to_payload(info) |
| 21 | # create the packet itself |
| 22 | p = ( |
| 23 | Ether(dst=src_if.local_mac, src=src_if.remote_mac) |
| 24 | / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) |
| 25 | / UDP(sport=randint(49152, 65535), dport=5678) |
| 26 | / Raw(payload) |
| 27 | ) |
| 28 | # store a copy of the packet in the packet info |
| 29 | info.data = p.copy() |
| 30 | # append the packet to the list |
| 31 | packets.append(p) |
| 32 | |
| 33 | # return the created packet list |
| 34 | return packets |
| 35 | |
| 36 | |
| 37 | def verify_capture(self, src_if, dst_if, capture, reply): |
no test coverage detected