import logging from core libs helpers import TemplateConfig TemplatePa

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import logging
from core.libs.helpers import TemplateConfig, TemplatePacket
class TestLag(object):
"""
@brief Suite for LAG testing.
"""
def test_setup_lag1(self, dut1, tg1):
"""
@brief Verify that LAG basic configuration can be set up to DUT.
"""
# Restore startup configuration on device.
logging.info("Restore startup configuration on device.")
dut1.restore_defaults()
# Create empty configuration object.
config = TemplateConfig()
# Add LACP template "one_lag_two_ports" into config object.
config.add_template("lacp", "one_lag_two_ports")
# Set ports 1, 2 to LAG 5.
config.lacp.lag.ports = [1, 2]
config.lacp.lag.id = 5
# Add VLAN template "one_lag_two_ports" into config object.
config.add_template("vlan", "one_vlan_two_ports")
# Set port 3 and LAG 5 to VLAN 25.
config.vlan.conf.ports = [3]
config.vlan.conf.lags = [5]
config.vlan.conf.member.vlan2.id = 25
# Apply LACP preconfiguration to device.
assert dut1.lacp.apply_config(config), "Configuration cannot be applied."
# Apply VLAN preconfiguration to device.
assert dut1.vlan.apply_config(config), "Configuration cannot be applied."
# Verify that LAG ID for port 1 is 5.
assert dut1.lacp.get_param("id", port=1) == 5, "LAG ID is not '5'."
# Verify that VLAN ID for port 3 is 25.
assert dut1.vlan.get_param("id", port=3) == 25, "VLAN ID is not '5'."
# Set port_admin_timeout for ports in LAG 5 to "SHORT".
assert dut1.lacp.set_param("actor.port_admin_timeout", "SHORT", ports=[1, 2]), "Ports timeout was not set to 'SHORT'."
# Add stream
my_stream = tg1.add_stream(port=1, num_packets=16)
# Set frame length:
my_stream.control.frame_length.length = 128
# Set packet per second:
my_stream.control.rate.packet_per_sec = 100
# Add "ethernet" layer with custom dst_mac:
my_stream.packet.l2.eth2.dst_mac = "11:22:33:44:55:66"
# Add "ip4" level with custom src_ip:
my_stream.packet.l3.ip4.src_ip = "1.2.3.4"
# Method for sort layers sort() :
my_stream.packet.sort()
# Capturing packets:
packets_l4 = tg1.get_captured_buffer([1, 2], filter="l4") # show only l4 packages
packets_ip = tg1.get_captured_buffer([1, 2], filter="l3.ip4") # show only IP packeges
packets_src_ip = tg1.get_captured_buffer([1, 2], filter="l3.ip4.src_ip == '1.2.3.4'")