Create input packet stream for defined interface. :param list packet_sizes: Required packet sizes.
(cls, packet_sizes, packet_count=test_packet_count)
| 1357 | |
| 1358 | @classmethod |
| 1359 | def create_stream(cls, packet_sizes, packet_count=test_packet_count): |
| 1360 | """Create input packet stream for defined interface. |
| 1361 | |
| 1362 | :param list packet_sizes: Required packet sizes. |
| 1363 | """ |
| 1364 | for i in range(0, packet_count): |
| 1365 | info = cls.create_packet_info(cls.src_if, cls.src_if) |
| 1366 | payload = cls.info_to_payload(info) |
| 1367 | p = ( |
| 1368 | Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) |
| 1369 | / IPv6(src=cls.src_if.remote_ip6, dst=cls.dst_if.remote_ip6) |
| 1370 | / UDP(sport=1234, dport=5678) |
| 1371 | / Raw(payload) |
| 1372 | ) |
| 1373 | size = packet_sizes[(i // 2) % len(packet_sizes)] |
| 1374 | cls.extend_packet(p, size, cls.padding) |
| 1375 | info.data = p |
| 1376 | |
| 1377 | @classmethod |
| 1378 | def create_fragments(cls): |
nothing calls this directly
no test coverage detected