Method create_stream

test/test_ip_ecmp.py:97–125

Create input packet stream for defined interfaces. :param VppInterface src_if: Source Interface for packet stream. :param str src_ip_start: Starting source IPv4 or IPv6 address. :param str dst_ip_start: Starting destination IPv4 or IPv6 address. :param int ip_prefix_

(
        self, src_if, src_ip_start, dst_ip_start, ip_prefix_len, packet_sizes, ip_l=IP
    )

        return str(ip_addr + random.randint(0, 2 ** (ip_max_len - ip_prefix_len) - 2))

    def create_stream(
        self, src_if, src_ip_start, dst_ip_start, ip_prefix_len, packet_sizes, ip_l=IP
    ):
        """Create input packet stream for defined interfaces.

        :param VppInterface src_if: Source Interface for packet stream.
        :param str src_ip_start: Starting source IPv4 or IPv6 address.
        :param str dst_ip_start: Starting destination IPv4 or IPv6 address.
        :param int ip_prefix_len: IP address prefix length.
        :param list packet_sizes: packet size to test.
        :param Scapy ip_l: Required IP layer - IP or IPv6. (Default is IP.)
        """
        pkts = []
        for i in range(0, N_PKTS_IN_STREAM):
            info = self.create_packet_info(src_if, src_if)
            payload = self.info_to_payload(info)
            src_ip = self.get_ip_address(src_ip_start, ip_prefix_len)
            dst_ip = self.get_ip_address(dst_ip_start, ip_prefix_len)
            p = (
                Ether(dst=src_if.local_mac, src=src_if.remote_mac)
                / ip_l(src=src_ip, dst=dst_ip)
                / UDP(sport=1234, dport=1234)
                / Raw(payload)
            )
            info.data = p.copy()
            size = random.choice(packet_sizes)
            self.extend_packet(p, size)
            pkts.append(p)
        return pkts

    def verify_capture(self, rx_if, capture, ip_l=IP):
        """Verify captured input packet stream for defined interface.