NpduManager

class mtf.network_port.npdu_manager.NpduManager

Manager for configuring and transmitting NPDUs

__init__()
property vlan_assign

vlan_assign attribute

set_up(ecu_name: str, adapter_name: str, ecu_local_ip: str, brcf_path: str = None, drcf_path: str = None, drcf_channels: [<class 'str'>] = None, input_local_port: int = 0, listener_channel_name: str = None)

Setup the npdu manager to be configured with global properties.

Note

If DRCF file is passed into parameters, please pass accordingly the brcf_path, ecu_name and drcf_channels

Parameters:

  • ecu_name – ECU name from BRCF file.

  • adapter_name – The Adapter name from testBench config (yaml file).

  • ecu_local_ip – The local IP address of the ECU

  • brcf_path – If not none BRCF file will be paresed and all channels from this file will be configured. Defaults to None.

  • drcf_path – If not none DRCF file will be parsed and all channels from this drcf_channel will be configured. Defaults to None.

  • drcf_channels – The list of DRCF channels to be configured. Defaults to None.

  • listener_channel_name – The adapter name to listen over it: an optional parameter in case there is a need to listen over a specific adapter

Returns:

True if the setup was done, else False.

add_input_pdu_tunnel(channel_name: str, multi_cast_ip_v4_addresses: list, multi_cast_port: int, remote_ip_v4_address: str, local_port: int = 0, with_vlan_assign: bool = True, force_add: bool = False, adapter_name: str = '')

Add an input tunnel to the Npdu manager’s configuration.

Parameters:

  • channel_name – the Tunnel name from BRCF file.

  • multi_cast_ip_v4_addresses – list of the multi-cast Ipv4 addresses.

  • multi_cast_port – the multi-cast port identifier.

  • remote_ip_v4_address – the remote ipv4 address.

  • local_port – the local port identifier.

  • with_vlan_assign – Flag to allow adding the Vlan. Defaults to True.

  • force_add – Flag to indicate if to force adding the tunnel. Defaults to False.

  • adapter_name – the local adapter name. Defaults to empty string. will be returned.The tunnel will be added or an exception will be logged.

add_output_pdu_tunnel(channel_name: str, multi_cast_ip_v4_addresses: list, multi_cast_port: int, transmission_timeout: int, local_ip: str, local_port: int = 0, force_add: bool = False, adapter_name: str = '')

Add an output tunnel to the Npdu manager’s configuration.

Parameters:

  • channel_name – the Tunnel name from BRCF file.

  • multi_cast_ip_v4_addresses – list of the multi-cast Ipv4 addresses.

  • multi_cast_port – the multi-cast port identifier.

  • transmission_timeout – the transmission cycle time in second.

  • local_ip – The local Ip address.

  • local_port – the local port identifier.

  • force_add – Flag to indicate if to force adding the tunnel. Defaults to False.

  • adapter_name – the local adapter name. Defaults to empty string. will be returned. The tunnel will be added or an exception will be logged.

start_npdu_listener(channel_name: str, silent_mode: bool = False, npdu_filter: mtf_eth_npdu_filter | None = None) → bool

Start ndpu listener on tunnel with name channel_name.

Note

The tunnel must be configured before in the setup of npdu manager or a custom tunnel was added with Add Npdu Tunnel

Parameters:

channel_name – name of Npdu tunnel

Returns:

True if listener is started. False otherwise.

stop_npdu_listener(channel_name: str, npdu_filter: mtf_eth_npdu_filter | None = None) → bool

Stop Npdu listener on tunnel with name channel_name

Parameters:

channel_name – name of Npdu tunnel

Returns:

True if listener is started. False otherwise.

get_received_pdus(channel_name: str, pdu_id: int, reset: bool = True, npdu_filter: mtf_eth_npdu_filter | None = None)

Get a list of payloads of the received pdus with a specific pdu_id.

Warning

The tunnel should be already configured

Parameters:

  • channel_name – tunnel name.

  • pdu_id – pdu id

  • reset – Flag to allow reseting the message queue AFTER getting the received pdus if True. Defaults to True.

Returns:

list of npdu elements payloads.

get_received_pdus_elemets(channel_name: str, pdu_id: int, reset=True, npdu_filter: mtf_eth_npdu_filter | None = None)

Get a list of NpduElement from the received pdus elements objects with a specific pdu_id.

Warning

The tunnel should be already configured

Note

  • Each NpduElement has the following attributes:

    *pdu_id *pdu_index *payload *direction *tci (VlanTci object, optional)

Parameters:

  • channel_name – tunnel name.

  • pdu_id – pdu identifier.

  • reset – Flag to allow reseting the message queue AFTER getting the received pdus if True. Defaults to True.

Returns:

List of all received NpduElements if tunnel configured

get_received_pdus_timestamp(channel_name: str, pdu_id: int, reset=True, npdu_filter: mtf_eth_npdu_filter | None = None)

Get a list of Timestamp from the received pdus with a specific pdu_id.

Warning

The tunnel should be already configured

Parameters:

  • channel_name – tunnel name.

  • pdu_id – pdu identifier.

  • reset – Flag to allow reseting the message queue AFTER getting the received pdus if True. Defaults to True.

Returns:

List of all received Npdu Timestamp if tunnel configured

clean_up() → bool

Clean up Npdu manager.

Note

This method is already called in the Teardown of the Testcase. This method does the following cleanups for each configured tunnel: * Stop all listeners. * Clean up the queue. * Remove tunnel from configuration. * Remove all listener instances.

Returns:

True if the clean-up was done correctly. False otherwise.

transmit(channel_name, npdu, trig_mode=0)

Transmits npdu frame on tunnel with name channel_name.

Note

Needs tunnel configuration. Each NpduElement has the following attributes:

*pdu_id *pdu_index *payload *direction *tci (VlanTci object, optional)

Parameters:

  • channel_name – name of tunnel.

  • npdu – NpduElement to be transmitted.

  • trig_mode – Npdu trigger mode that could be 0= Buffered or 1= Direct. Defaults to 0 (Buffered).

get_all_received_pdus(channel_name: str, reset=True, npdu_filter: mtf_eth_npdu_filter | None = None)

Get a list of tuples (pdu_id, payload) of all the received pdus.

Note

The tunnel should be already configured

Parameters:

  • channel_name – tunnel name.

  • reset – Flag to allow reseting the message queue AFTER getting the received pdus if True. Defaults to True.

Returns:

list [(npdu_element.pdu_id, npdu_element.payload)] .

get_all_received_npdu_data(channel_name: str, reset=True, npdu_filter: mtf_eth_npdu_filter | None = None)

Get a list of all the received npdu elements of channel_name.

Note

  • The tunnel should be already configured

Parameters:

  • channel_name – channel name

  • reset – Flag to allow reseting the message queue AFTER getting the received pdus if True. Defaults to True.

Returns:

list [npdu_element_1, npdu_element_2,…].

transmit_npdus(channel_name, frames: list)

Transmits a list of npdu frames on tunnel with name channel_name.

Note

  • Needs tunnel configuration.

  • Trigger mode is by default buffered in this case.

  • Each NpduElement has the following attributes:

    *pdu_id *pdu_index *payload *direction *tci (VlanTci object, optional)

Parameters:

  • channel_name – name of tunnel.

  • frames – NpduElement to be transmitted.

set_npdu_listener_filter(channel_name: str, filter: dict)

set a filter on a certain npdu tunnel listener.

Note

  • Needs tunnel configuration.

Parameters:

  • channel_name – name of tunnel.

  • filter – a dictionary with the name of the element to be filtered on as keys and their corresponding values.

static __new__(cls, *args, **kwargs)

Initialize a new instance of the ControllerBase.

The instance will be automatically added to the list of controllers.

clear_npdu_listener_filter(channel_name: str)

clear a filter on a certain npdu tunnel listener.

Note

  • Needs tunnel configuration.

Parameters:

channel_name – name of tunnel.

classmethod controllers_cleanup()

Clean up all controller instances.

Calls the clean_up() method for each instance.