KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use wait_for_port_status method in localstack

Best Python code snippet using localstack_python

test_qos_samples.py

Source:test_qos_samples.py

...73        helpers.set_ports_admin_enabled(env.switch, ports)74        # Disable Flow Control functionality75        env.switch[1].ui.modify_ports(device_ports, flowControl='None')76        for port in device_ports:77            env.switch[1].ui.wait_for_port_status(port, 'flowControl', 'None', 10)78        # Disable Cut Through functionality79        env.switch[1].ui.modify_ports(device_ports, cutThrough='Disabled')80        for port in device_ports:81            env.switch[1].ui.wait_for_port_status(port, 'cutThrough', 'Disabled', 10)82        # SetUp multicast and broadcast rate limits83        env.switch[1].ui.modify_ports(device_ports, setPortAttr="mcast_rate", attrVal=0)84        env.switch[1].ui.modify_ports(device_ports, setPortAttr="bcast_rate", attrVal=0)85        # Define packets with different Dot1p tags.86        packet_1 = ({"Ethernet": {"dst": dst_3, "src": src_1, "type": 0x8100}},87                    {"Dot1Q": {"vlan": 1, "type": 0x0800, "prio": prio_higher}},88                    {"IP": {}})89        packet_2 = ({"Ethernet": {"dst": dst_3, "src": src_2, "type": 0x8100}},90                    {"Dot1Q": {"vlan": 1, "type": 0x0800, "prio": prio_lower}},91                    {"IP": {}})92        stream_1 = env.tg[1].set_stream(packet_1, iface=ports[('tg1', 'sw1')][1],93                                        rate=100, continuous=True)94        stream_2 = env.tg[1].set_stream(packet_2, iface=ports[('tg1', 'sw1')][2],95                                        rate=100, continuous=True)96        # Set Dot1p trust mode value on ingress port_1 and port_2.97        env.switch[1].ui.configure_port_cos(ports=[device_ports[0], device_ports[1]],98                                            trustMode="Dot1p")99        # Verify values have been changed100        for port in [device_ports[0], device_ports[1]]:101            port_qos_trust_mode = env.switch[1].ui.get_table_ports_qos_scheduling(port=port)102            assert port_qos_trust_mode['trustMode'] == "Dot1p", \103                'Trust mode value is not set to "Dot1p" for port %s' % port104        # Set Strict sched mode value on egress port_3.105        env.switch[1].ui.configure_port_cos(ports=[device_ports[2], ], schedMode="Strict")106        # Verify value has been changed107        port_qos_3_sched_mode = env.switch[1].ui.get_table_ports_qos_scheduling(port=device_ports[2])108        assert port_qos_3_sched_mode['schedMode'] == "Strict", \109            'Strict sched mode is not set on port %s' % ports[('sw1', 'tg1')][3]110        # Set Tagged value on egress port_3.111        env.switch[1].ui.modify_vlan_ports(ports=[device_ports[2], ],112                                           vlans=[1, ],113                                           tagged="Tagged")114        # Add Static MAC entry for egress port_3 definition.115        env.switch[1].ui.create_static_macs(device_ports[2], [1, ], [dst_3, ])116        # Set QoS statistic type for Dot1p definition on the TG port_3.117        env.tg[1].set_qos_stat_type(ports[('tg1', 'sw1')][3], "VLAN")118        # Need to wait until device generates all ICMP packets.119        time.sleep(5)120        # Send streams of packets with Dot1p priorities from the TG port_1 and port_2.121        env.tg[1].clear_statistics([ports[('tg1', 'sw1')][1],122                                    ports[('tg1', 'sw1')][2],123                                    ports[('tg1', 'sw1')][3], ])124        env.tg[1].start_sniff([ports[('tg1', 'sw1')][3], ],125                              filter_layer="Dot1Q.IP",126                              src_filter=src_1,127                              dst_filter=dst_3)128        env.tg[1].start_streams([stream_1, stream_2, ])129        time.sleep(10)130        # Stop sending streams from TG port_1 and port_2.131        env.tg[1].stop_streams([stream_1, stream_2, ])132        time.sleep(0.5)133        env.tg[1].stop_sniff([ports[('tg1', 'sw1')][3], ], drop_packets=True)134        # Get count of sent packets on the TG port_1 and port_2.135        sent_count_stream_1 = env.tg[1].get_sent_frames_count(ports[('tg1', 'sw1')][1], )136        # Get count of received packets with Dot1p higher priority on TG port_3.137        dot1p_count_stream_1 = env.tg[1].get_qos_frames_count(ports[('tg1', 'sw1')][3],138                                                              prio_higher)139        if dot1p_count_stream_1 not in list(range(sent_count_stream_1,140                                             sent_count_stream_1 * 101 // 100)):141            pytest.fail("Packets with Dot1p higher priority are lost when default mapping is configured.")142        # Verify that filtered count of packets on TG port_3 is correct.143        filtered_count_stream_1 = env.tg[1].get_filtered_frames_count(ports[('tg1', 'sw1')][3], )144        if filtered_count_stream_1 != dot1p_count_stream_1:145            pytest.fail("Filter packets statistic has incorrect count of packets!")146        # Verify that not used queues don't receive any packets.147        if not self.not_used_cos_statistics(tg_instance=env.tg[1],148                                            prio_value_1=prio_higher,149                                            prio_value_2=prio_lower,150                                            iface=ports[('tg1', 'sw1')][3]):151            pytest.fail("Not used CoS statistics received unexpected packets!")152    @pytest.mark.skip("Pypacker does not support Vlan prio field")153    def test_qos_dscp_strict_mode(self, env):154        """155        @brief  Verify that packets with DSCP higher priority displace156                packets with DSCP lower priority by Strict schedule mode.157        @steps158            -# Set DSCP trust mode value on ingress port 1 and port 2.159            -# Set Strict sched mode value on egress port 3.160            -# Send two streams with different IP.tos values.161            -# Verify that packets with DSCP higher priority displace packets with DSCP lower priority162        @endsteps163        """164        prio_higher = 96165        prio_lower = 32166        src_1 = "00:00:00:00:01:11"167        src_2 = "00:00:00:00:02:22"168        dst_3 = "00:00:00:00:03:33"169         # Define active ports170        ports = env.get_ports([['sw1', 'tg1', 3], ])171        device_ports = list(ports[('sw1', 'tg1')].values())172        # Disable all ports and enabling only necessary ones:173        helpers.set_all_ports_admin_disabled(env.switch)174        helpers.set_ports_admin_enabled(env.switch, ports)175        # Disable Flow Control functionality176        env.switch[1].ui.modify_ports(device_ports, flowControl='None')177        for port in device_ports:178            env.switch[1].ui.wait_for_port_status(port, 'flowControl', 'None', 10)179        # Disable Cut Through functionality180        env.switch[1].ui.modify_ports(device_ports, cutThrough='Disabled')181        for port in device_ports:182            env.switch[1].ui.wait_for_port_status(port, 'cutThrough', 'Disabled', 10)183        # SetUp multicast and broadcast rate limits184        env.switch[1].ui.modify_ports(device_ports, setPortAttr="mcast_rate", attrVal=0)185        env.switch[1].ui.modify_ports(device_ports, setPortAttr="bcast_rate", attrVal=0)186        # Define packets with different TOS tags and configure streams of defined packets.187        packet_1 = ({"Ethernet": {"dst": dst_3, "src": src_1, "type": 0x0800}},188                    {"IP": {"tos": prio_higher}},189                    {"UDP": {}})190        packet_2 = ({"Ethernet": {"dst": dst_3, "src": src_2, "type": 0x0800}},191                    {"IP": {"tos": prio_lower}},192                    {"UDP": {}})193        stream_1 = env.tg[1].set_stream(packet_1, iface=ports[('tg1', 'sw1')][1],194                                        rate=100, continuous=True)195        stream_2 = env.tg[1].set_stream(packet_2, iface=ports[('tg1', 'sw1')][2],196                                        rate=100, continuous=True)...

net.py

Source:net.py

...77    port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.578):79    """Ping the given network port until it becomes available (for a given number of retries).80    If 'http_path' is set, make a GET request to this path and assert a non-error response."""81    return wait_for_port_status(82        port,83        http_path=http_path,84        expect_success=expect_success,85        retries=retries,86        sleep_time=sleep_time,87    )88def wait_for_port_closed(89    port: int, http_path: str = None, expect_success=True, retries=10, sleep_time=0.590):91    return wait_for_port_status(92        port,93        http_path=http_path,94        expect_success=expect_success,95        retries=retries,96        sleep_time=sleep_time,97        expect_closed=True,98    )99def wait_for_port_status(100    port: int,101    http_path: str = None,102    expect_success=True,103    retries=10,104    sleep_time=0.5,105    expect_closed=False,106):107    """Ping the given network port until it becomes (un)available (for a given number of retries)."""108    def check():109        status = is_port_open(port, http_path=http_path, expect_success=expect_success)110        if bool(status) != (not expect_closed):111            raise Exception(112                "Port %s (path: %s) was not %s"113                % (port, http_path, "closed" if expect_closed else "open")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.