xenavalkyrie package

xenavalkyrie.xena_app module

Classes and utilities that represents Xena XenaManager-2G application and chassis.

author:yoram@ignissoft.com
class xenavalkyrie.xena_app.XenaApp(logger, owner, api_wrapper)[source]

Bases: trafficgenerator.tgn_app.TgnApp

XenaApp object, equivalent to XenaManager-2G application.

class xenavalkyrie.xena_app.XenaBaseModule(parent, index)[source]

Bases: xenavalkyrie.xena_object.XenaObject

Represents Xena module.

capabilities
cli_prefix = 'm'
get_name()[source]
inventory()[source]

Get module inventory.

is_chimera()[source]
is_loki()[source]
is_odin()[source]
is_thor()[source]
ports
Returns:dictionary {index: object} of all ports.
save_config(config_file_name, file_mode='w+')[source]

Save module configuration file (including all ports under module).

Parameters:
  • config_file_name – full path to the configuration file.
  • file_mode – w+ for module configuration file, a+ for chassis configuration.
set_timing_source_local()[source]
class xenavalkyrie.xena_app.XenaChassis(parent, ip, port=22611, password='xena')[source]

Bases: xenavalkyrie.xena_object.XenaObject

Represents single Xena chassis.

cli_prefix = 'c'
get_session_id()[source]

Get ID of the current automation session on the chassis.

Note that this ID can be different for different chassis on the same session.

Returns:chassis ID.
inventory(modules_inventory=False)[source]

Get chassis inventory.

Parameters:modules_inventory – True - read modules inventory, false - don’t read.
modules
Returns:dictionary {index: object} of all modules.
ports
Returns:dictionary {name: object} of all ports.
read_stats()[source]
Returns:dictionary {own: {stat name: value}}
release_modules()[source]

Release all ports that were reserved during the session.

XenaManager-2G -> Release Ports.

release_ports()[source]

Release all ports that were reserved during the session.

XenaManager-2G -> Release Ports.

reserve_modules(locations, force=False)[source]

Reserve modules.

XenaManager-2G -> Reserve/Relinquish module. XenaManager-2G -> Reset module.

Parameters:
  • locations – list of modules locations to reserve
  • force – True - take forcefully, False - fail if module is reserved by other user
Returns:

modules dictionary (index: object)

reserve_ports(locations, force=False, reset=True)[source]

Reserve ports and reset factory defaults.

XenaManager-2G -> Reserve/Relinquish Port. XenaManager-2G -> Reset port.

Parameters:
  • locations – list of ports locations in the form <module/port> to reserve
  • force – True - take forcefully, False - fail if port is reserved by other user
  • reset – True - reset port, False - leave port configuration
Returns:

ports dictionary (index: object)

save_config(config_file_name)[source]

Save entire chassis configuration file.

Parameters:config_file_name – full path to the configuration file.
shutdown(restart=False, wait=False)[source]

Shutdown chassis.

Limitations: shutdown to single chassis will disconnect all chassis so in multiple chassis environment the test should reconnect by calling api.add_chassis(chassis).

Parameters:
  • restart – True - restart, False - poweroff
  • wait – True - wait for chassis to come up after restart, False - return immediately
Todo:

fix limitation.

start_traffic(blocking=False, *ports)[source]

Start traffic on list of ports.

Parameters:
  • blocking – True - start traffic and wait until traffic ends, False - start traffic and return.
  • ports – list of ports to start traffic on. Default - all session ports.
stats_captions = ['ses', 'typ', 'adr', 'own', 'ops', 'req', 'rsp']
stop_traffic(*ports)[source]

Stop traffic on list of ports.

Parameters:ports – list of ports to stop traffic on. Default - all session ports.
wait_traffic(*ports)[source]

Wait until traffic stops on ports.

Parameters:ports – list of ports to wait for.
class xenavalkyrie.xena_app.XenaModule(parent, index)[source]

Bases: xenavalkyrie.xena_app.XenaBaseModule

class xenavalkyrie.xena_app.XenaModuleCapabilities[source]

Structure that provides the module capabilities

class xenavalkyrie.xena_app.XenaSession(logger, owner, api)[source]

Bases: xenavalkyrie.xena_object.XenaObject

Xena scripting object. Root object for the Xena objects tree.

add_chassis(chassis, port=22611, password='xena')[source]

Add chassis.

XenaManager-2G -> Add Chassis.

Parameters:
  • chassis – chassis IP address
  • port – chassis port number
  • password – chassis password
Returns:

newly created chassis

Return type:

xenavalkyrie.xena_app.XenaChassis

chassis_list
Returns:dictionary {name: object} of all chassis.
clear_stats(*ports)[source]

Clear stats (TX and RX) for list of ports.

Parameters:ports – list of ports to clear stats on. Default - all session ports.
disconnect(release=True)[source]

Release ports and disconnect from all chassis.

inventory()[source]

Get inventory for all chassis.

modules
Returns:dictionary {name: object} of all modules.
ports
Returns:dictionary {name: object} of all ports.
read_stats(*ports)[source]

Read statistics on list of ports.

Parameters:ports – list of ports to read statistics. Default - all session ports.
release_modules()[source]

Release modules.

XenaManager-2G -> Release Module.

release_ports()[source]

Release all ports that were reserved during the session.

XenaManager-2G -> Release Ports.

reserve_modules(locations, force=False)[source]

Reserve modules.

XenaManager-2G -> Reserve/Relinquish Module. XenaManager-2G -> Reserve Module.

Parameters:
  • locations – list of locations in the form <ip/slot/port> to reserve
  • force – True - take forcefully. False - fail if module is reserved by other user
Returns:

module dictionary (index: object)

reserve_ports(locations, force=False, reset=True)[source]

Reserve ports and reset factory defaults.

XenaManager-2G -> Reserve/Relinquish Port. XenaManager-2G -> Reserve Port.

Parameters:
  • locations – list of ports locations in the form <ip/slot/port> to reserve
  • force – True - take forcefully. False - fail if port is reserved by other user
  • reset – True - reset port, False - leave port configuration
Returns:

ports dictionary (index: object)

start_capture(*ports)[source]

Start capture on list of ports.

Parameters:ports – list of ports to start capture on. Default - all session ports.
start_traffic(blocking=False, *ports)[source]

Start traffic on list of ports.

Parameters:
  • blocking – True - start traffic and wait until traffic ends, False - start traffic and return.
  • ports – list of ports to start traffic on. Default - all session ports.
stop_capture(*ports)[source]

Stop capture on list of ports.

Parameters:ports – list of ports to stop capture on. Default - all session ports.
stop_traffic(*ports)[source]

Stop traffic on list of ports.

Parameters:ports – list of ports to stop traffic on. Default - all session ports.
xenavalkyrie.xena_app.init_xena(api, logger, owner, ip=None, port=57911)[source]

Create XenaApp object.

Parameters:
  • api – cli/rest
  • logger – python logger
  • owner – owner of the scripting session
  • ip – rest server IP
  • port – rest server TCP port
Returns:

Xena object

Return type:

XenaApp

xenavalkyrie.xena_object module

Base classes and utilities for all Xena Manager (Xena) objects.

author:yoram@ignissoft.com
exception xenavalkyrie.xena_object.XenaAttributeError[source]

Bases: trafficgenerator.tgn_utils.TgnError

class xenavalkyrie.xena_object.XenaObject(**data)[source]

Bases: trafficgenerator.tgn_object.TgnObject

Base class for all Xena objects.

get_attribute(attribute)[source]

Returns single object attribute.

Parameters:attribute – requested attribute to query.
Returns:returned value.
Return type:str
get_attributes()[source]

Returns all object’s attributes.

Returns:dictionary of <name, value> of all attributes.
Return type:dict of (str, str)
id
Returns:object ID.
index
Returns:object index.
obj_id()[source]
Returns:object ID.
obj_index()[source]
Returns:object index.
read_stat(captions, stat_name)[source]
release()[source]

Release object.

XenaManager-2G -> Release Chassis/Module/Port.

relinquish()[source]

Relinquish object.

XenaManager-2G -> Relinquish Chassis/Module/Port.

reserve(force=False)[source]

Reserve object.

XenaManager-2G -> [Relinquish]/Reserve Chassis/Module/Port.

Parameters:force – True - take forcefully, False - fail if port is reserved by other user
send_command(command, *arguments)[source]

Send command with no output.

Parameters:
  • command – command to send.
  • arguments – list of command arguments.
send_command_return(command, *arguments)[source]

Send command and wait for single line output.

send_command_return_multilines(command, *arguments)[source]

Send command and wait for multiple lines output.

set_attributes(**attributes)[source]

Sets list of attributes.

Parameters:attributes – dictionary of {attribute: value} to set.
wait_for_states(attribute, timeout=40, *states)[source]
class xenavalkyrie.xena_object.XenaObject21(**data)[source]

Bases: xenavalkyrie.xena_object.XenaObject

Base class for all Xena objects with index_len = 2 and command_len = 1.

class xenavalkyrie.xena_object.XenaObjectsDict(**kwds)[source]

Bases: trafficgenerator.tgn_object.TgnObjectsDict

xenavalkyrie.xena_port module

Classes and utilities that represents Xena XenaManager-2G port.

author:yoram@ignissoft.com
class xenavalkyrie.xena_port.XenaBasePort(parent, index)

Bases: xenavalkyrie.xena_object.XenaObject

Represents Xena port.

add_filter(comment=None)

Add filter.

We cannot set state before we set condition so it is the test responsibility.

Parameters:comment – filter description.
Returns:newly created filter.
Return type:xenavalkyrie.xena_filter.XenaFilter
add_length()

Add match.

Returns:newly created match.
Return type:xenavalkyrie.xena_filter.XenaMatch
add_match()

Add match.

Returns:newly created match.
Return type:xenavalkyrie.xena_filter.XenaMatch
add_stream(name=None, tpld_id=None, state=<XenaStreamState.enabled: 'ON'>)

Add stream.

Parameters:
Returns:

newly created stream.

Return type:

xenavalkyrie.xena_stream.XenaStream

capabilities
capture
Returns:capture object.
Return type:XenaCapture
clear_stats()

Clear att TX and RX statistics counter.

Port Statistics -> Clear TX Counters, Clear RX Counters

cli_prefix = 'p'
filters
Returns:dictionary {id: object} of all filters.
Return type:dict of (int, xenavalkyrie.xena_filter.XenaFilter)
inventory()
lengthes
Returns:dictionary {id: object} of all lengthes.
Return type:dict of (int, xenavalkyrie.xena_filter.XenaLength)
load_config(config_file_name)

Load configuration file from xpc file.

Parameters:config_file_name – full path to the configuration file.
matches
Returns:dictionary {id: object} of all matches.
Return type:dict of (int, xenavalkyrie.xena_filter.XenaMatch)
read_port_stats()
Returns:dictionary {group name {stat name: value}}. Sea XenaBasePort.stats_captions.
read_stream_stats()
Returns:dictionary {stream index {stat name: value}}. Sea XenaStream.stats_captions.
read_tpld_stats()
Returns:dictionary {tpld index {group name {stat name: value}}}. Sea XenaTpld.stats_captions.
remove_filter(index)

Remove filter.

Parameters:index – index of filter to remove.
remove_length(index)

Remove length.

Parameters:index – index of length to remove.
remove_match(index)

Remove match.

Parameters:index – index of match to remove.
remove_stream(index)

Remove stream.

Parameters:index – index of stream to remove.
reset()

Reset port-level parameters to standard values, and delete all streams, filters, capture, and dataset definitions.

save_config(config_file_name, file_mode='w+')

Save configuration file to xpc file.

Parameters:
  • config_file_name – full path to the configuration file.
  • file_mode – w+ for port configuration file, a+ for module configuration.
start_capture()

Start capture on port.

Capture -> Start Capture

start_traffic(blocking=False)

Start port traffic.

Port -> Start Traffic

Parameters:blocking – True - start traffic and wait until traffic ends, False - start traffic and return.
stats_captions = {'pr_extra': ['fcserrors', 'pauseframes', 'arprequests', 'arpreplies', 'pingrequests', 'pingreplies', 'gapcount', 'gapduration'], 'pr_notpld': ['bps', 'pps', 'bytes', 'packets'], 'pr_pfcstats': ['total', 'CoS 0', 'CoS 1', 'CoS 2', 'CoS 3', 'CoS 4', 'CoS 5', 'CoS 6', 'CoS 7'], 'pr_total': ['bps', 'pps', 'bytes', 'packets'], 'pt_extra': ['arprequests', 'arpreplies', 'pingrequests', 'pingreplies', 'injectedfcs', 'injectedseq', 'injectedmis', 'injectedint', 'injectedtid', 'training'], 'pt_notpld': ['bps', 'pps', 'bytes', 'packets'], 'pt_total': ['bps', 'pps', 'bytes', 'packets']}
stop_capture()

Stop capture on port.

Capture -> Stop Capture

stop_traffic()

Stop port traffic.

Port -> Stop Traffic

streams
Returns:dictionary {id: object} of all streams.
Return type:dict of (int, xenavalkyrie.xena_stream.XenaStream)
tplds
Returns:dictionary {id: object} of all current tplds.
Return type:dict of (int, xenavalkyrie.xena_port.XenaTpld)
wait_for_up(timeout=40)
class xenavalkyrie.xena_port.XenaCapture(parent)

Bases: xenavalkyrie.xena_object.XenaObject

Represents capture parameters, correspond to the Capture panel of the XenaManager, and deal with configuration of the capture criteria and inspection of the captured data from a port.

get_packets(from_index=0, to_index=None, cap_type=<XenaCaptureBufferType.text: 1>, file_name=None, tshark=None)

Get captured packets from chassis.

Parameters:
  • from_index – index of first packet to read.
  • to_index – index of last packet to read. If None - read all packets.
  • cap_type – returned capture format. If pcap then file name and tshark must be provided.
  • file_name – if specified, capture will be saved in file.
  • tshark – tshark object for pcap type only.
Type:

xenavalkyrie.xena_tshark.Tshark

Returns:

list of requested packets, None for pcap type.

packets
Returns:dictionary {id: object} of all packets.
Return type:dict of (int, xenavalkyrie.xena_port.XenaCapturePacket)
read_stats()
Returns:dictionary {stat name: value}. Sea XenaCapture.stats_captions.
stats_captions = ['status', 'packets', 'starttime']
class xenavalkyrie.xena_port.XenaCaptureBufferType

Bases: enum.Enum

pcap = 2
raw = 0
text = 1
class xenavalkyrie.xena_port.XenaCapturePacket(parent, index)

Bases: xenavalkyrie.xena_object.XenaObject21

Represents single captured packet.

class xenavalkyrie.xena_port.XenaPort(parent, index)

Bases: xenavalkyrie.xena_port.XenaBasePort

clear_rx_pcs_stats()
enable_pma_err_pulse(enable=True)
read_fec_stats()
Returns:list showing how many FEC blocks have been seen with [0, 1, 2, 3….N, > N] symbol errors
read_rx_total_stats()
Returns:
set_pma_err_pulse(duration, period, repetition, coeff, exp)

Sets the parameters for the PMA pulse error inject.

Period must be bigger than duration, BER will be calculated as coeff * power(10, exp)

:param duration : 0 ms – 5 s; increments of 1 ms; 0 = constant BER :param period : 10 ms – 50 s; number of ms – must be multiple of 10 ms :param repetition: 1 – 64K; 0 = continuous :param coeff : (0.01 < coeff < 9.99) * 100 :param exp : -3 < exp < -17

set_tx_error_rate(rate)
class xenavalkyrie.xena_port.XenaPortCapabilities

Structure that provides the port capabilities

class xenavalkyrie.xena_port.XenaTpld(parent, index)

Bases: xenavalkyrie.xena_object.XenaObject21

read_stats()
Returns:dictionary {group name {stat name: value}}. Sea XenaTpld.stats_captions.
stats_captions = {'pr_tplderrors': ['dummy', 'seq', 'mis', 'pld'], 'pr_tpldjitter': ['min', 'avg', 'max', 'avg1sec', 'min1sec', 'max1sec'], 'pr_tpldlatency': ['min', 'avg', 'max', 'avg1sec', 'min1sec', 'max1sec'], 'pr_tpldtraffic': ['bps', 'pps', 'byt', 'pac']}

xenavalkyrie.xena_statistics_view module

Classes and utilities that represents Xena statistics as tables like the GUI.

Statistics views represent statistics as tables. There are three different views - ports, streams and TPLD.

author:yoram@ignissoft.com
class xenavalkyrie.xena_statistics_view.XenaPortsStats(session)[source]

Bases: xenavalkyrie.xena_statistics_view.XenaStats

Ports statistics view.

Represents all ports statistics as multi table:

Port Name Group     Group    
  Name Name Name Name
IP/Module/Port value value Name Name
IP/Module/Port value value Name Name
read_stats()[source]

Read current ports statistics from chassis.

Returns:dictionary {port name {group name, {stat name: stat value}}}
class xenavalkyrie.xena_statistics_view.XenaStats(session)[source]

Bases: object

Base class for all statistics views.

get_flat_stats()[source]
Returns:statistics as flat table {port/strea,/tpld name {group_stat name: value}}
class xenavalkyrie.xena_statistics_view.XenaStreamsStats(session)[source]

Bases: xenavalkyrie.xena_statistics_view.XenaStats

Streams statistics view.

Represents all streams statistics as table:

Stream tx   rx              
      Port       Port      
      Group   Group   Group   Group  
  Name Name Name Name Name
Object value value value value value
Object value value value value value
get_flat_stats()[source]
Returns:statistics as flat table {port/strea,/tpld name {group_stat name: value}}
read_stats()[source]

Read current statistics from chassis.

Returns:dictionary {stream: {tx: {stat name: stat value}} rx: {tpld: {stat group {stat name: value}}}}
class xenavalkyrie.xena_statistics_view.XenaTpldsStats(session)[source]

Bases: xenavalkyrie.xena_statistics_view.XenaStats

TPLDs statistics view.

Represents all TPLDs statistics as multi column table:

TPLD Full Index Group     Group    
  Name Name Name Name
Module/Port/Index value value Name Name
Module/Port/Index value value Name Name
read_stats()[source]

Read current statistics from chassis.

Returns:dictionary {tpld full index {group name {stat name: stat value}}}

xenavalkyrie.xena_stream module

Classes and utilities that represents Xena XenaManager-2G stream.

author:yoram@ignissoft.com
class xenavalkyrie.xena_stream.XenaModifier(parent, index)[source]

Bases: xenavalkyrie.xena_stream._XenaModifierBase

class xenavalkyrie.xena_stream.XenaModifierAction[source]

Bases: enum.Enum

decrement = 'DEC'
increment = 'INC'
random = 'RANDOM'
class xenavalkyrie.xena_stream.XenaModifierType[source]

Bases: enum.Enum

extended = 1
standard = 0
class xenavalkyrie.xena_stream.XenaStream(parent, index, name='')[source]

Bases: xenavalkyrie.xena_object.XenaObject21

add_modifier(m_type=<XenaModifierType.standard: 0>, **kwargs)[source]

Add modifier.

Parameters:m_type – modifier type - standard or extended.
Type:xenavalkyrie.xena_stram.ModifierType
Returns:newly created modifier.
Return type:xenavalkyrie.xena_stream.XenaModifier
create_command = 'ps_create'
del_object_from_parent()[source]

Delete object from parent object.

get_packet_headers()[source]
Returns:current packet headers
Return type:pypacker.layer12.ethernet.Ethernet
modifiers
Returns:dictionary {index: object} of standard modifiers.
next_tpld_id = 0
read_stats()[source]
Returns:dictionary {stat name: value} See XenaStream.stats_captions
remove_modifier(index, m_type=<XenaModifierType.standard: 0>)[source]

Remove modifier.

Parameters:
  • m_type – modifier type - standard or extended.
  • index – index of modifier to remove.
set_packet_headers(headers, l4_checksum=False)[source]

Set packet header.

The method will try to set ps_headerprotocol to inform the Xena GUI and tester how to interpret the packet header byte sequence specified with PS_PACKETHEADER. This is mainly for information purposes, and the stream will transmit the packet header bytes even if no protocol segments are specified. If the method fails to set some segment it will log a warning and skip setup.

Parameters:
  • headers (pypacker.layer12.ethernet.Ethernet) – current packet headers
  • l4_checksum – True - set tcp/udp checksum flag, False - do not set
set_state(state)[source]

Set stream state.

Parameters:state – new stream state.
stats_captions = ['bps', 'pps', 'bytes', 'packets']
xmodifiers
Returns:dictionary {index: object} of extended modifiers.
class xenavalkyrie.xena_stream.XenaStreamState[source]

Bases: enum.Enum

disabled = 'OFF'
enabled = 'ON'
suspended = 'SUPPRESS'
class xenavalkyrie.xena_stream.XenaXModifier(parent, index)[source]

Bases: xenavalkyrie.xena_stream._XenaModifierBase