Method create_stream

test/test_acl_plugin.py:269–341

Create input packet stream for defined interface using hosts or deleted_hosts list. :param object src_if: Interface to create packet stream for. :param list packet_sizes: List of required packet sizes. :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0:

(
        self,
        src_if,
        packet_sizes,
        traffic_type=0,
        ipv6=0,
        proto=-1,
        ports=0,
        fragments=False,
        pkt_raw=True,
        etype=-1,
    )

        return ""

    def create_stream(
        self,
        src_if,
        packet_sizes,
        traffic_type=0,
        ipv6=0,
        proto=-1,
        ports=0,
        fragments=False,
        pkt_raw=True,
        etype=-1,
    ):
        """
        Create input packet stream for defined interface using hosts or
        deleted_hosts list.

        :param object src_if: Interface to create packet stream for.
        :param list packet_sizes: List of required packet sizes.
        :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
        :return: Stream of packets.
        """
        pkts = []
        if self.flows.__contains__(src_if):
            src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
            for dst_if in self.flows[src_if]:
                dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
                n_int = len(dst_hosts) * len(src_hosts)
                for i in range(0, n_int):
                    dst_host = dst_hosts[int(i / len(src_hosts))]
                    src_host = src_hosts[i % len(src_hosts)]
                    pkt_info = self.create_packet_info(src_if, dst_if)
                    if ipv6 == 1:
                        pkt_info.ip = 1
                    elif ipv6 == 0:
                        pkt_info.ip = 0
                    else:
                        pkt_info.ip = random.choice([0, 1])
                    if proto == -1:
                        pkt_info.proto = random.choice(self.proto[self.IP])
                    else:
                        pkt_info.proto = proto
                    payload = self.info_to_payload(pkt_info)
                    p = Ether(dst=dst_host.mac, src=src_host.mac)
                    if etype > 0:
                        p = Ether(dst=dst_host.mac, src=src_host.mac, type=etype)
                    if pkt_info.ip:
                        p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
                        if fragments:
                            p /= IPv6ExtHdrFragment(offset=64, m=1)
                    else:
                        if fragments:
                            p /= IP(
                                src=src_host.ip4, dst=dst_host.ip4, flags=1, frag=64
                            )
                        else:
                            p /= IP(src=src_host.ip4, dst=dst_host.ip4)
                    if traffic_type == self.ICMP:
                        if pkt_info.ip: