open-traffic-generator / conformance

Open Traffic Generator Conformance Tests
MIT License
4 stars 1 forks source link

Capture packet is not working #34

Closed desaimg1 closed 1 year ago

desaimg1 commented 1 year ago

Hi Team,

I tried using following sample for packet capture but is is throwing Error.

Sample Code : https://github.com/open-traffic-generator/conformance/blob/main/features/flows/headers/tcp/test_tcp_header.py#L48

My code :

def init(self):
    # Configure a new API instance where the location points to controller
    if self.mode == 'ixnet':
        self.api = snappi.api(location=self.exec_params['controller'], verify=False, ext='ixnetwork')
    elif self.mode == 'trex':
        self.api = snappi.api(location=self.exec_params['controller'], verify=False, ext='trex')
    else:
        self.api = snappi.api(location=self.exec_params['controller'], verify=False)
    logging.info("%s Starting connection to controller... " % time.strftime("%s"))
     self.configuration = self.api.config()

    # Configure two ports where the location points to the port location:

    for pid, port in enumerate(self.exec_params['port_groups']):
        print("pid and port is : {} {}".format(pid,port))
        location = port.get('location', f"localhost:{BASE_TENGINE_PORT+pid}")
        print("location, port is {}, {}".format(location,port[port['init']]))
        self.configuration.ports.port(name=port[port['init']], location=location)

    self.configuration.options.port_options.location_preemption = True

    ca = self.configuration.captures.add(name="ca", port_names=[p.name for p in self.configuration.ports])
    ca.format = ca.PCAP

In script file:
==========
def test_bgp_as_number_with_packet_capture(self):
    self.dp.stop_protocols()
    self.dp.start_capture()
    time.sleep(5)
    self.dp.start_protocols()
    time.sleep(5)
    self.dp.stop_capture()

    captured_packets = self.dp.api.get_capture(self.dp.configuration.ports[0].name)

Error:

  captured_packets = self.dp.api.get_capture(self.dp.configuration.ports[0].name)

test_hero_8_24k_ips_b2b.py:299:


self = <snappi_ixnetwork.snappi_api.Api object at 0x7fbb950482b0>, request = 'VNET-1'

def get_capture(self, request):
    """Gets capture file and returns it as a byte stream"""
    try:
        if (
            isinstance(request, (type(self._capture_request), str))
            is False
        ):
            raise TypeError(
                "The content must be of type Union[CaptureRequest, str]"
            )
        if isinstance(request, str) is True:
            request = self._capture_request.deserialize(request)
        self._connect()
    except Exception as err:
      raise SnappiIxnException(err)

E snappi_ixnetwork.exceptions.SnappiIxnException: File "/usr/local/lib/python3.7/dist-packages/snappi_ixnetwork/snappi_api.py", line 446, in get_capture E request = self._capture_request.deserialize(request) E File "/usr/local/lib/python3.7/dist-packages/snappi/snappi.py", line 244, in deserialize E self._decode(serialized_object) E File "/usr/local/lib/python3.7/dist-packages/snappi/snappi.py", line 532, in _decode E for property_name, property_value in obj.items(): E 'str' object has no attribute 'items'

/usr/local/lib/python3.7/dist-packages/snappi_ixnetwork/snappi_api.py:449: SnappiIxnException

ANISH-GOTTAPU commented 1 year ago

captured_packets = self.dp.api.get_capture(self.dp.configuration.ports[0].name) don't work

You have to create capture_request body and send it as input to get_capture

Correct way of API usage:

request = self.dp.api.capture_request()
request.port_name = self.dp.configuration.ports[0].name
captured_packets = self.dp.api.get_capture(request)