def table_9_test_preparation(self, env, tg_port, sw_port, packet_num, ipgap, offset=0, arp_packet=None):
"""Prepare ports, packets for table 9 related tests.
"""
# Configure ARP packet and stream
if arp_packet:
srcmac = arp_packet[0]['Ether']['src']
arp_packet[0]['Ether']['src'] = str(EUI(EUI(srcmac).value + offset, dialect=mac_unix_expanded))
srcip = arp_packet[1]['ARP']['psrc']
arp_packet[1]['ARP']['psrc'] = str(IPAddress(srcip) + offset)
else:
raise UIException("ARP packet not supplied")
# set admin status of host and switch to Up, wait till operational status is up
sw_port_id = int(sw_port.split()[1])
sw_instances = [switch for switch in env.switch[1].node.values() if switch.id != sw_port.split()[0]]
sw_instance = [switch for switch in env.switch[1].node.values() if switch.id == sw_port.split()[0]]
if not sw_instance:
raise UIException("Not found switch id and port pair connection in configuration")
env.tg[1].connect_port(tg_port)
sw_instance[0].ui.modify_ports(ports=[sw_port_id], adminMode='Up')
sw_instance[0].ui.wait_for_port_value_to_change([sw_port_id], 'operationalStatus', "Up")
# Prepare and send stream
arp_stream = env.tg[1].set_stream(arp_packet, count=packet_num, inter=ipgap, iface=tg_port,
arp_sa_increment=(1, packet_num + offset),
arp_sip_increment=(1, packet_num + offset))
env.tg[1].start_streams([arp_stream])
time.sleep(ipgap * packet_num)
env.tg[1].stop_streams([arp_stream])
return sw_instances
评论列表
文章目录
