CI: update test cases to use different environment variables

change test environments
optimize asio udp server test
fix icmp echo test case
use ethernet_router env to run iperf test cases

.gitlab/ci/target-test.yml

@@ -11,8 +11,14 @@
     reports:
       junit: XUNIT_RESULT.xml
     expire_in: 1 week
+  variables:
+    GIT_DEPTH: 1
   script:
     - retry_failed git clone $KNOWN_FAILURE_CASES_REPO known_failure_cases
+    # get runner env config file
+    - retry_failed git clone $TEST_ENV_CONFIG_REPO
+    - python $CHECKOUT_REF_SCRIPT ci-test-runner-configs ci-test-runner-configs
+    - cp ./ci-test-runner-configs/${CI_RUNNER_DESCRIPTION}/EnvConfig.yml EnvConfig.yml || true
     # using runner tags as markers to filter the test cases
     # Runner tags are comma separated, replace the comma with " and " for markers
     - job_tags=$(python tools/ci/python_packages/gitlab_api.py get_job_tags $CI_PROJECT_ID --job_id $CI_JOB_ID)
@@ -80,13 +86,13 @@ example_test_pytest_esp32c3_flash_suspend:
     - build_pytest_examples_esp32c3
   tags: [ esp32c3, flash_suspend ]
 
-example_test_pytest_esp32_ethernet_ota:
+example_test_pytest_esp32_ethernet_router:
   extends:
     - .pytest_examples_dir_template
     - .rules:test:example_test-esp32-ota-related_changes
   needs:
     - build_pytest_examples_esp32
-  tags: [ esp32, ethernet_ota ]
+  tags: [ esp32, ethernet_router ]
 
 example_test_pytest_esp32_wifi_ota:
   extends:
@@ -128,21 +134,29 @@ example_test_pytest_esp32_8mb_flash:
     - build_pytest_examples_esp32
   tags: [ esp32, ethernet_flash_8m ]
 
-example_test_pytest_esp32_wifi:
+example_test_pytest_esp32_wifi_nearby:
   extends:
     - .pytest_examples_dir_template
     - .rules:test:example_test-esp32-wifi-related_changes
   needs:
     - build_pytest_examples_esp32
-  tags: [ esp32, wifi ]
+  tags: [ esp32, wifi_nearby ]
 
-example_test_pytest_esp32_wifi_bt:
+example_test_pytest_esp32_wifi_router:
   extends:
     - .pytest_examples_dir_template
     - .rules:test:example_test-esp32
   needs:
     - build_pytest_examples_esp32
-  tags: [ esp32, wifi_bt ]
+  tags: [ esp32, wifi_router ]
+
+example_test_pytest_esp32_wifi_wlan:
+  extends:
+    - .pytest_examples_dir_template
+    - .rules:test:example_test-esp32
+  needs:
+    - build_pytest_examples_esp32
+  tags: [ esp32, wifi_wlan ]
 
 example_test_pytest_esp32_ethernet_ip101:
   extends:
@@ -340,6 +354,7 @@ test_app_test_pytest_esp32s2_usb_host:
       junit: $LOG_PATH/*/XUNIT_RESULT.xml
     expire_in: 1 week
   variables:
+    GIT_DEPTH: 1
     TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw"
     LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS"
     ENV_FILE: "$CI_PROJECT_DIR/ci-test-runner-configs/$CI_RUNNER_DESCRIPTION/EnvConfig.yml"
@@ -434,7 +449,7 @@ example_test_protocols:
   extends: .example_test_esp32_template
   tags:
     - ESP32
-    - Example_WIFI_Protocols
+    - wifi_router
 
 example_test_002:
   extends: .example_test_esp32_template
@@ -446,7 +461,7 @@ example_test_ethernet:
   extends: .example_test_esp32_template
   tags:
     - ESP32
-    - Example_Ethernet
+    - ethernet_router
 
 .example_test_003:
   extends: .example_test_esp32_template
@@ -559,6 +574,7 @@ example_test_ESP32C3_SDSPI:
   needs:
     - assign_custom_test
   variables:
+    GIT_DEPTH: 1
     TEST_CASE_PATH: "$CI_PROJECT_DIR/tools/test_apps"
     CONFIG_FILE_PATH: "${CI_PROJECT_DIR}/tools/test_apps/test_configs"
 
@@ -605,7 +621,7 @@ test_app_test_eth:
   extends: .test_app_esp32_template
   tags:
     - ESP32
-    - Example_EthKitV1
+    - ethernet_router
 
 test_app_test_003:
   extends: .test_app_esp32_template
@@ -666,6 +682,7 @@ test_app_test_flash_psram_f8r8:
   needs: # the assign already needs all the build jobs
     - assign_unit_test
   variables:
+    GIT_DEPTH: 1
     TEST_CASE_PATH: "$CI_PROJECT_DIR/tools/unit-test-app"
     CONFIG_FILE_PATH: "${CI_PROJECT_DIR}/components/idf_test/unit_test/test_configs"
 

examples/ethernet/iperf/iperf_test.py

@@ -14,7 +14,9 @@ import re
 import subprocess
 import time
 
+import netifaces
 import ttfw_idf
+from common_test_methods import get_env_config, get_my_ip_by_interface
 from idf_iperf_test_util import IperfUtility
 from tiny_test_fw import TinyFW
 
@@ -53,14 +55,15 @@ class IperfTestUtilityEth(IperfUtility.IperfTestUtility):
         return dut_ip, rssi
 
 
-@ttfw_idf.idf_example_test(env_tag='Example_Ethernet')
+@ttfw_idf.idf_example_test(env_tag='ethernet_router')
 def test_ethernet_throughput_basic(env, _):  # type: (Any, Any) -> None
     """
     steps: |
       1. test TCP tx rx and UDP tx rx throughput
       2. compare with the pre-defined pass standard
     """
-    pc_nic_ip = env.get_pc_nic_info('pc_nic', 'ipv4')['addr']
+    pc_nic = get_env_config('wifi_router').get('pc_nic', 'eth1')
+    pc_nic_ip = get_my_ip_by_interface(pc_nic, netifaces.AF_INET)
     pc_iperf_log_file = os.path.join(env.log_path, 'pc_iperf_log.md')
 
     # 1. get DUT

examples/network/simple_sniffer/example_test.py

@@ -1,52 +0,0 @@
-# SPDX-FileCopyrightText: 2021 Espressif Systems (Shanghai) CO LTD
-# SPDX-License-Identifier: Apache-2.0
-from __future__ import unicode_literals
-
-import re
-from typing import Any
-
-import ttfw_idf
-
-
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
-def test_examples_simple_sniffer(env, _):  # type: (Any, Any) -> None
-
-    dut = env.get_dut('simple_sniffer', 'examples/network/simple_sniffer', app_config_name='mem')
-    dut.start_app()
-
-    dut.expect('sniffer>')
-    dut.write('pcap --open -f simple-sniffer')
-    dut.expect('cmd_pcap: open file successfully')
-    dut.write('sniffer -i wlan -c 2 -n 10')
-    dut.expect('cmd_sniffer: 10 packages will be captured')
-    dut.expect('cmd_sniffer: start WiFi promiscuous ok')
-    dut.expect('cmd_sniffer: stop promiscuous ok')
-    dut.write('pcap --summary -f simple-sniffer')
-    dut.expect('cmd_pcap: Memory is to be parsed')
-    dut.expect('Pcap packet Head:')
-    dut.expect('Magic Number: a1b2c3d4')
-    dut.expect(re.compile(r'Major Version: [0-9]*'))
-    dut.expect(re.compile(r'Minor Version: [0-9]*'))
-    dut.expect(re.compile(r'SnapLen: [0-9]*'))
-    dut.expect(re.compile(r'LinkType: [0-9]*'))
-    for i in range(0, 10):
-        dut.expect('Packet ' + str(i) + ':')
-        dut.expect(re.compile(r'Timestamp \(Seconds\): [0-9]*'))
-        dut.expect(re.compile(r'Timestamp \(Microseconds\): [0-9]*'))
-        dut.expect(re.compile(r'Capture Length: [0-9]*'))
-        dut.expect(re.compile(r'Packet Length: [0-9]*'))
-        dut.expect(re.compile(r'Frame Type: .*'))
-        dut.expect(re.compile(r'Frame Subtype: .*'))
-        dut.expect(re.compile(r'Destination: .*'))
-        dut.expect(re.compile(r'Source: .*'))
-    dut.expect('Pcap packet Number: 10')
-    dut.write('pcap --close -f simple-sniffer')
-    dut.expect('cmd_pcap: free memory successfully')
-    dut.expect('cmd_pcap: .pcap file close done')
-
-    dut.write('')
-    dut.expect('sniffer>')
-
-
-if __name__ == '__main__':
-    test_examples_simple_sniffer()

examples/network/simple_sniffer/pytest_simple_sniffer.py

@@ -0,0 +1,47 @@
+# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+from pytest_embedded import Dut
+
+
+@pytest.mark.esp32
+@pytest.mark.wifi_nearby
+@pytest.mark.parametrize('config', [
+    'mem',
+], indirect=True)
+def test_examples_simple_sniffer(dut: Dut) -> None:
+    sniffer_num = 10
+    dut.expect('sniffer>')
+    dut.write('pcap --open -f simple-sniffer')
+    dut.expect('cmd_pcap: open file successfully')
+    dut.write(f'sniffer -i wlan -c 2 -n {sniffer_num}')
+    dut.expect(f'cmd_sniffer: {sniffer_num} packages will be captured')
+    dut.expect('cmd_sniffer: start WiFi promiscuous ok')
+    dut.expect('cmd_sniffer: stop promiscuous ok')
+    dut.write('pcap --summary -f simple-sniffer')
+    dut.expect('cmd_pcap: Memory is to be parsed')
+    dut.expect('Pcap packet Head:')
+    dut.expect('Magic Number: a1b2c3d4')
+    dut.expect(r'Major Version: [0-9]+')
+    dut.expect(r'Minor Version: [0-9]+')
+    dut.expect(r'SnapLen: [0-9]+')
+    dut.expect(r'LinkType: [0-9]+')
+    # Allow "save captured packet failed" once
+    for i in range(0, sniffer_num - 1):
+        dut.expect(f'Packet {i}:')
+        dut.expect(r'Timestamp \(Seconds\): [0-9]+')
+        dut.expect(r'Timestamp \(Microseconds\): [0-9]+')
+        dut.expect(r'Capture Length: [0-9]+')
+        dut.expect(r'Packet Length: [0-9]+')
+        dut.expect(r'Frame Type:\s+\w+')
+        dut.expect(r'Frame Subtype:\s+\w+')
+        dut.expect(r'Destination:\s+\w+')
+        dut.expect(r'Source:\s+\w+')
+    dut.expect(r'Pcap packet Number: \d+')
+    dut.write('pcap --close -f simple-sniffer')
+    dut.expect('cmd_pcap: free memory successfully')
+    dut.expect('cmd_pcap: .pcap file close done')
+
+    dut.write('')
+    dut.expect('sniffer>')

examples/protocols/.build-test-rules.yml

@@ -108,7 +108,7 @@ examples/protocols/https_x509_bundle:
 
 examples/protocols/icmp_echo:
   disable_test:
-    - if: IDF_TARGET != "esp32"
+    - if: IDF_TARGET == "esp32c2"
       temporary: true
       reason: lack of runners
 

examples/protocols/cbor/pytest_cbor.py

@@ -9,7 +9,7 @@ from pytest_embedded import Dut
 
 
 @pytest.mark.supported_targets
-@pytest.mark.wifi
+@pytest.mark.generic
 def test_examples_cbor(dut: Dut) -> None:
 
     dut.expect(r'example: encoded buffer size \d+')

examples/protocols/esp_local_ctrl/pytest_esp_local_ctrl.py

@@ -39,7 +39,7 @@ class CustomProcess(object):
 
 
 @pytest.mark.supported_targets
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_esp_local_ctrl(dut: Dut) -> None:
 
     rel_project_path = os.path.join('examples', 'protocols', 'esp_local_ctrl')

examples/protocols/http_server/advanced_tests/pytest_http_server_advanced.py

@@ -35,7 +35,7 @@ from pytest_embedded import Dut
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_http_server_advanced(dut: Dut) -> None:
 
     # Get binary file

examples/protocols/http_server/captive_portal/pytest_captive_portal.py

@@ -72,7 +72,7 @@ def test_captive_page(ip: str, port: str, uri: str) -> bool:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi_bt
+@pytest.mark.wifi_wlan
 @pytest.mark.xfail(reason='Runner unable to connect to target over WiFi', run=False)
 def test_example_captive_portal(dut: Dut) -> None:
 

examples/protocols/http_server/persistent_sockets/pytest_http_server_persistence.py

@@ -30,7 +30,7 @@ from pytest_embedded import Dut
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_http_server_persistence(dut: Dut) -> None:
 
     # Get binary file

examples/protocols/http_server/simple/pytest_http_server_simple.py

@@ -62,7 +62,7 @@ class http_client_thread(threading.Thread):
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_http_server_simple(dut: Dut) -> None:
 
     # Get binary file
@@ -130,7 +130,7 @@ def test_examples_protocol_http_server_simple(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_http_server_lru_purge_enable(dut: Dut) -> None:
 
     # Get binary file

examples/protocols/http_server/ws_echo_server/pytest_ws_server_example.py

@@ -51,7 +51,7 @@ class WsClient:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_http_ws_echo_server(dut: Dut) -> None:
     # Get binary file
     binary_file = os.path.join(dut.app.binary_path, 'ws_echo_server.bin')

examples/protocols/https_request/pytest_https_request.py

@@ -10,19 +10,11 @@ from typing import Callable
 
 import pexpect
 import pytest
+from common_test_methods import get_my_ip4_by_dest_ip
 from pytest_embedded import Dut
 from RangeHTTPServer import RangeRequestHandler
 
 
-def get_my_ip() -> str:
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = ''
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def get_server_status(host_ip: str, port: int) -> bool:
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     server_status = sock.connect_ex((host_ip, port))
@@ -86,7 +78,15 @@ def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
     bin_size = os.path.getsize(binary_file)
     logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024))
     # start test
-    host_ip = get_my_ip()
+    dut.expect('Loaded app from partition at offset', timeout=30)
+    try:
+        ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=60)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
+
+    # start https server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
     server_port = 8070
     server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem')
     key_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_key.pem')
@@ -96,17 +96,9 @@ def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
         thread1.start()
     logging.info('The server started on {}:{}'.format(host_ip, server_port))
 
-    dut.expect('Loaded app from partition at offset', timeout=30)
-    try:
-        ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode()
-        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-    except pexpect.exceptions.TIMEOUT:
-        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-
     dut.expect('Start https_request example', timeout=30)
 
     print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port)))
-
     dut.write('https://' + host_ip + ':' + str(server_port))
     logging.info("Testing for \"https_request using saved session\"")
 

examples/protocols/https_server/simple/pytest_https_server_simple.py

@@ -93,7 +93,7 @@ success_response = '<h1>Hello Secure World!</h1>'
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_https_server_simple(dut: Dut) -> None:
     """
     steps: |
@@ -168,7 +168,7 @@ def test_examples_protocol_https_server_simple(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 @pytest.mark.parametrize('config', ['dynamic_buffer',], indirect=True)
 def test_examples_protocol_https_server_simple_dynamic_buffers(dut: Dut) -> None:
     # Test with mbedTLS dynamic buffer feature

examples/protocols/https_server/wss_server/pytest_https_wss_server.py

@@ -109,7 +109,7 @@ def test_multiple_client_keep_alive_and_async_response(ip, port, ca_file):  # ty
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_https_wss_server(dut: Dut) -> None:
 
     # Get binary file

examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py

@@ -11,7 +11,7 @@ from pytest_embedded import Dut
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 def test_examples_protocol_https_x509_bundle(dut: Dut) -> None:
     """
     steps: |
@@ -33,7 +33,7 @@ def test_examples_protocol_https_x509_bundle(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.wifi
+@pytest.mark.wifi_router
 @pytest.mark.parametrize('config', ['ssldyn',], indirect=True)
 def test_examples_protocol_https_x509_bundle_dynamic_buffer(dut: Dut) -> None:
     # test mbedtls dynamic resource

examples/protocols/icmp_echo/example_test.py

@@ -1,33 +0,0 @@
-from __future__ import unicode_literals
-
-import os
-import re
-
-import ttfw_idf
-
-
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
-def test_examples_icmp_echo(env, extra_data):
-
-    dut = env.get_dut('icmp_echo', 'examples/protocols/icmp_echo')
-    dut.start_app()
-
-    dut.expect('example_connect: Connected to')
-    dut.expect('esp>')
-
-    ping_dest = os.getenv('EXAMPLE_ICMP_SERVER', 'www.espressif.com')
-    dut.write('ping {}'.format(ping_dest))
-
-    ip_re = r'\.'.join((r'\d{1,3}',) * 4)
-    ip = dut.expect(re.compile(r'64 bytes from ({}) icmp_seq=1 ttl=\d+ time=\d+ ms'.format(ip_re)))[0]
-
-    # expect at least one more (there could be lost packets)
-    dut.expect(re.compile(r'64 bytes from {} icmp_seq=[2-5] ttl=\d+ time='.format(ip)))
-
-    dut.expect(re.compile(r'5 packets transmitted, [2-5] received, \d{1,3}% packet loss'))
-    dut.write('')
-    dut.expect('esp>')
-
-
-if __name__ == '__main__':
-    test_examples_icmp_echo()

examples/protocols/icmp_echo/main/echo_example_main.c

@@ -199,8 +199,6 @@ void app_main(void)
     ESP_ERROR_CHECK(nvs_flash_init());
     ESP_ERROR_CHECK(esp_netif_init());
     ESP_ERROR_CHECK(esp_event_loop_create_default());
-    /* wait for active network connection */
-    ESP_ERROR_CHECK(example_connect());
 
     esp_console_repl_config_t repl_config = ESP_CONSOLE_REPL_CONFIG_DEFAULT();
     // install console REPL environment
@@ -215,6 +213,8 @@ void app_main(void)
     ESP_ERROR_CHECK(esp_console_new_repl_usb_serial_jtag(&usbjtag_config, &repl_config, &repl));
 #endif
 
+    /* register wifi connect commands */
+    register_wifi_connect_commands();
     /* register command `ping` */
     register_ping();
     /* register command `quit` */

examples/protocols/icmp_echo/pytest_icmp_echo.py

@@ -0,0 +1,36 @@
+# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+
+import pytest
+from common_test_methods import get_env_config
+from pytest_embedded import Dut
+
+
+@pytest.mark.esp32
+@pytest.mark.esp32c3
+@pytest.mark.esp32s2
+@pytest.mark.esp32s3
+@pytest.mark.wifi_nearby
+def test_protocols_icmp_echo(dut: Dut) -> None:
+    # get env config
+    env_config = get_env_config('wifi_nearby')
+    ap_ssid = env_config['ap_ssid']
+    ap_password = env_config['ap_password']
+    ap_channel = env_config.get('ap_channel', 0)
+
+    dut.expect('esp>')
+    dut.write(f'wifi_connect {ap_ssid} {ap_password} -n {ap_channel}')
+    dut.expect('Got IPv4 event:', timeout=30)
+
+    ping_dest = os.getenv('EXAMPLE_ICMP_SERVER', 'ci.espressif.cn')
+    dut.write('ping {} -c 5'.format(ping_dest))
+
+    # expect at least two packets (there could be lost packets)
+    ip = dut.expect(r'64 bytes from (\d+\.\d+\.\d+\.\d+) icmp_seq=\d ttl=\d+ time=\d+ ms')[1].decode()
+    dut.expect(fr'64 bytes from {ip} icmp_seq=[2-5] ttl=\d+ time=')
+
+    dut.expect(r'5 packets transmitted, [2-5] received, \d{1,3}% packet loss')
+    dut.write('')
+    dut.expect('esp>')

examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py

@@ -7,19 +7,12 @@ import time
 from threading import Thread
 
 import ttfw_idf
+from common_test_methods import get_my_ip4_by_dest_ip
 from tiny_test_fw import DUT
 
 msgid = -1
 
 
-def get_my_ip():
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def mqqt_server_sketch(my_ip, port):
     global msgid
     print('Starting the server on {}'.format(my_ip))
@@ -68,19 +61,20 @@ def test_examples_protocol_mqtt_qos1(env, extra_data):
     binary_file = os.path.join(dut1.app.binary_path, 'mqtt_tcp.bin')
     bin_size = os.path.getsize(binary_file)
     ttfw_idf.log_performance('mqtt_tcp_bin_size', '{}KB'.format(bin_size // 1024))
-    # 1. start mqtt broker sketch
-    host_ip = get_my_ip()
-    thread1 = Thread(target=mqqt_server_sketch, args=(host_ip,1883))
-    thread1.start()
-    # 2. start the dut test and wait till client gets IP address
+    # 1. start the dut test and wait till client gets IP address
     dut1.start_app()
     # waiting for getting the IP address
     try:
-        ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)
+        ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)[0]
         print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
     except DUT.ExpectTimeout:
         raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
 
+    # 2. start mqtt broker sketch
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = Thread(target=mqqt_server_sketch, args=(host_ip,1883))
+    thread1.start()
+
     print('writing to device: {}'.format('mqtt://' + host_ip + '\n'))
     dut1.write('mqtt://' + host_ip + '\n')
     thread1.join()

examples/protocols/sntp/example_test.py

@@ -1,51 +0,0 @@
-from __future__ import unicode_literals
-
-import datetime
-import re
-
-import ttfw_idf
-from tiny_test_fw import Utility
-
-
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
-def test_examples_sntp(env, extra_data):
-
-    dut = env.get_dut('sntp', 'examples/protocols/sntp')
-    dut.start_app()
-
-    dut.expect_all('Time is not set yet. Connecting to WiFi and getting time over NTP.',
-                   'Initializing SNTP',
-                   re.compile(r'Waiting for system time to be set... \(\d+/\d+\)'),
-                   'Notification of a time synchronization event',
-                   timeout=60)
-
-    TIME_FORMAT = '%a %b %d %H:%M:%S %Y'
-    TIME_FORMAT_REGEX = r'\w+\s+\w+\s+\d{1,2}\s+\d{2}:\d{2}:\d{2} \d{4}'
-    TIME_DIFF = datetime.timedelta(seconds=10 + 2)  # cpu spends 10 seconds in deep sleep
-    NY_time = None
-    SH_time = None
-
-    def check_time(prev_NY_time, prev_SH_time):
-        NY_str = dut.expect(re.compile(r'The current date/time in New York is: ({})'.format(TIME_FORMAT_REGEX)))[0]
-        SH_str = dut.expect(re.compile(r'The current date/time in Shanghai is: ({})'.format(TIME_FORMAT_REGEX)))[0]
-        Utility.console_log('New York: "{}"'.format(NY_str))
-        Utility.console_log('Shanghai: "{}"'.format(SH_str))
-        dut.expect('Entering deep sleep for 10 seconds')
-        Utility.console_log('Sleeping...')
-        new_NY_time = datetime.datetime.strptime(NY_str, TIME_FORMAT)
-        new_SH_time = datetime.datetime.strptime(SH_str, TIME_FORMAT)
-
-        # The initial time is not checked because datetime has problems with timezones
-        assert prev_NY_time is None or new_NY_time - prev_NY_time < TIME_DIFF
-        assert prev_SH_time is None or new_SH_time - prev_SH_time < TIME_DIFF
-
-        return (new_NY_time, new_SH_time)
-
-    NY_time, SH_time = check_time(NY_time, SH_time)
-    for i in range(2, 4):
-        dut.expect('example: Boot count: {}'.format(i), timeout=30)
-        NY_time, SH_time = check_time(NY_time, SH_time)
-
-
-if __name__ == '__main__':
-    test_examples_sntp()

examples/protocols/sntp/pytest_sntp.py

@@ -0,0 +1,55 @@
+# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+import logging
+from typing import Any, Tuple
+
+import pytest
+from common_test_methods import get_env_config
+from pytest_embedded import Dut
+
+
+@pytest.mark.esp32
+@pytest.mark.wifi_nearby
+def test_get_time_from_sntp_server(dut: Dut) -> None:
+    dut.expect('Time is not set yet. Connecting to WiFi and getting time over NTP.')
+    if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
+        # get env config
+        env_config = get_env_config('wifi_nearby')
+        ap_ssid = env_config['ap_ssid']
+        ap_password = env_config['ap_password']
+        dut.expect('Please input ssid password:')
+        dut.write(' '.join([ap_ssid, ap_password]))
+    dut.expect('IPv4 address:')
+
+    dut.expect('Initializing SNTP')
+    dut.expect(r'Waiting for system time to be set... \(\d+/\d+\)')
+    dut.expect('Notification of a time synchronization event')
+
+    TIME_FORMAT = '%a %b %d %H:%M:%S %Y'
+    TIME_FORMAT_REGEX = r'\w+\s+\w+\s+\d{1,2}\s+\d{2}:\d{2}:\d{2} \d{4}'
+    TIME_DIFF = datetime.timedelta(seconds=10 + 2)  # cpu spends 10 seconds in deep sleep
+    NY_time = None
+    SH_time = None
+
+    def check_time(prev_NY_time: Any, prev_SH_time: Any) -> Tuple[Any, Any]:
+        NY_str = dut.expect(r'The current date/time in New York is: ({})'.format(TIME_FORMAT_REGEX))[1].decode()
+        SH_str = dut.expect(r'The current date/time in Shanghai is: ({})'.format(TIME_FORMAT_REGEX))[1].decode()
+        logging.info('New York: "{}"'.format(NY_str))
+        logging.info('Shanghai: "{}"'.format(SH_str))
+        dut.expect('Entering deep sleep for 10 seconds')
+        logging.info('Sleeping...')
+        new_NY_time = datetime.datetime.strptime(NY_str, TIME_FORMAT)
+        new_SH_time = datetime.datetime.strptime(SH_str, TIME_FORMAT)
+
+        # The initial time is not checked because datetime has problems with timezones
+        assert not prev_NY_time or new_NY_time - prev_NY_time < TIME_DIFF
+        assert not prev_SH_time or new_SH_time - prev_SH_time < TIME_DIFF
+
+        return (new_NY_time, new_SH_time)
+
+    NY_time, SH_time = check_time(NY_time, SH_time)
+    for i in range(2, 4):
+        dut.expect('example: Boot count: {}'.format(i), timeout=30)
+        NY_time, SH_time = check_time(NY_time, SH_time)

examples/protocols/sntp/sdkconfig.ci

@@ -1,2 +1,3 @@
 CONFIG_SNTP_TIME_SERVER="time.windows.com"
 CONFIG_LWIP_SNTP_MAX_SERVERS=2
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y

examples/protocols/sockets/tcp_client/example_test.py

@@ -17,18 +17,13 @@ from threading import Event, Thread
 
 import netifaces
 import ttfw_idf
+from common_test_methods import get_env_config, get_my_interface_by_dest_ip, get_my_ip_by_interface
 
 # -----------  Config  ----------
 PORT = 3333
-INTERFACE = 'eth0'
 # -------------------------------
 
 
-def get_my_ip(type):
-    for i in netifaces.ifaddresses(INTERFACE)[type]:
-        return i['addr'].replace('%{}'.format(INTERFACE), '')
-
-
 class TcpServer:
 
     def __init__(self, port, family_addr, persist=False):
@@ -85,7 +80,7 @@ class TcpServer:
                 break
 
 
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
+@ttfw_idf.idf_example_test(env_tag='wifi_router')
 def test_examples_protocol_socket_tcpclient(env, extra_data):
     """
     steps:
@@ -93,6 +88,11 @@ def test_examples_protocol_socket_tcpclient(env, extra_data):
       2. have the board connect to the server
       3. send and receive data
     """
+    # get env config
+    env_config = get_env_config('wifi_router')
+    ap_ssid = env_config['ap_ssid']
+    ap_password = env_config['ap_password']
+
     dut1 = env.get_dut('tcp_client', 'examples/protocols/sockets/tcp_client', dut_class=ttfw_idf.ESP32DUT)
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, 'tcp_client.bin')
@@ -101,21 +101,24 @@ def test_examples_protocol_socket_tcpclient(env, extra_data):
 
     # start test
     dut1.start_app()
+    dut1.expect('Please input ssid password:')
+    dut1.write(' '.join([ap_ssid, ap_password]))
 
     ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
     ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8)    # expect all 8 octets from IPv6 (assumes it's printed in the long form)
     ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
     print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
 
+    my_interface = get_my_interface_by_dest_ip(ipv4)
     # test IPv4
     with TcpServer(PORT, socket.AF_INET):
-        server_ip = get_my_ip(netifaces.AF_INET)
+        server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET)
         print('Connect tcp client to server IP={}'.format(server_ip))
         dut1.write(server_ip)
         dut1.expect(re.compile(r'OK: Message from ESP32'))
     # test IPv6
     with TcpServer(PORT, socket.AF_INET6):
-        server_ip = get_my_ip(netifaces.AF_INET6)
+        server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET6)
         print('Connect tcp client to server IP={}'.format(server_ip))
         dut1.write(server_ip)
         dut1.expect(re.compile(r'OK: Message from ESP32'))

examples/protocols/sockets/tcp_client/sdkconfig.ci

@@ -1 +1,2 @@
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
 CONFIG_EXAMPLE_SOCKET_IP_INPUT_STDIN=y

examples/protocols/sockets/tcp_server/example_test.py

@@ -14,10 +14,10 @@ import socket
 import sys
 
 import ttfw_idf
+from common_test_methods import get_env_config, get_my_interface_by_dest_ip
 
 # -----------  Config  ----------
 PORT = 3333
-INTERFACE = 'eth0'
 # -------------------------------
 
 
@@ -46,7 +46,7 @@ def tcp_client(address, payload):
     return data.decode()
 
 
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
+@ttfw_idf.idf_example_test(env_tag='wifi_router')
 def test_examples_protocol_socket_tcpserver(env, extra_data):
     MESSAGE = 'Data to ESP'
     """
@@ -55,6 +55,11 @@ def test_examples_protocol_socket_tcpserver(env, extra_data):
       2. have the board connect to the server
       3. send and receive data
     """
+    # get env config
+    env_config = get_env_config('wifi_router')
+    ap_ssid = env_config['ap_ssid']
+    ap_password = env_config['ap_password']
+
     dut1 = env.get_dut('tcp_client', 'examples/protocols/sockets/tcp_server', dut_class=ttfw_idf.ESP32DUT)
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, 'tcp_server.bin')
@@ -63,19 +68,22 @@ def test_examples_protocol_socket_tcpserver(env, extra_data):
 
     # start test
     dut1.start_app()
+    dut1.expect('Please input ssid password:')
+    dut1.write(' '.join([ap_ssid, ap_password]))
 
     ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
     ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8)    # expect all 8 octets from IPv6 (assumes it's printed in the long form)
     ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
     print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
 
+    interface = get_my_interface_by_dest_ip(ipv4)
     # test IPv4
     received = tcp_client(ipv4, MESSAGE)
     if not received == MESSAGE:
         raise
     dut1.expect(MESSAGE)
     # test IPv6
-    received = tcp_client('{}%{}'.format(ipv6, INTERFACE), MESSAGE)
+    received = tcp_client('{}%{}'.format(ipv6, interface), MESSAGE)
     if not received == MESSAGE:
         raise
     dut1.expect(MESSAGE)

examples/protocols/sockets/tcp_server/sdkconfig.ci

@@ -1,2 +1,3 @@
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
 CONFIG_EXAMPLE_IPV4=y
 CONFIG_EXAMPLE_IPV6=y

examples/protocols/sockets/tcp_server/sdkconfig.ci.1

@@ -1,3 +1,4 @@
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
 CONFIG_EXAMPLE_IPV4=y
 CONFIG_EXAMPLE_IPV6=n
 CONFIG_EXAMPLE_CONNECT_IPV6=n

examples/protocols/sockets/udp_client/example_test.py

@@ -17,18 +17,13 @@ from threading import Event, Thread
 
 import netifaces
 import ttfw_idf
+from common_test_methods import get_env_config, get_my_interface_by_dest_ip, get_my_ip_by_interface
 
 # -----------  Config  ----------
 PORT = 3333
-INTERFACE = 'eth0'
 # -------------------------------
 
 
-def get_my_ip(type):
-    for i in netifaces.ifaddresses(INTERFACE)[type]:
-        return i['addr'].replace('%{}'.format(INTERFACE), '')
-
-
 class UdpServer:
 
     def __init__(self, port, family_addr, persist=False):
@@ -78,7 +73,7 @@ class UdpServer:
                 break
 
 
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
+@ttfw_idf.idf_example_test(env_tag='wifi_router')
 def test_examples_protocol_socket_udpclient(env, extra_data):
     """
     steps:
@@ -86,6 +81,11 @@ def test_examples_protocol_socket_udpclient(env, extra_data):
       2. have the board connect to the server
       3. send and receive data
     """
+    # get env config
+    env_config = get_env_config('wifi_router')
+    ap_ssid = env_config['ap_ssid']
+    ap_password = env_config['ap_password']
+
     dut1 = env.get_dut('udp_client', 'examples/protocols/sockets/udp_client', dut_class=ttfw_idf.ESP32DUT)
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, 'udp_client.bin')
@@ -94,21 +94,24 @@ def test_examples_protocol_socket_udpclient(env, extra_data):
 
     # start test
     dut1.start_app()
+    dut1.expect('Please input ssid password:')
+    dut1.write(' '.join([ap_ssid, ap_password]))
 
     ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
     ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8)    # expect all 8 octets from IPv6 (assumes it's printed in the long form)
     ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
     print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
 
+    my_interface = get_my_interface_by_dest_ip(ipv4)
     # test IPv4
     with UdpServer(PORT, socket.AF_INET):
-        server_ip = get_my_ip(netifaces.AF_INET)
+        server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET)
         print('Connect udp client to server IP={}'.format(server_ip))
         dut1.write(server_ip)
         dut1.expect(re.compile(r'OK: Message from ESP32'))
     # test IPv6
     with UdpServer(PORT, socket.AF_INET6):
-        server_ip = get_my_ip(netifaces.AF_INET6)
+        server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET6)
         print('Connect udp client to server IP={}'.format(server_ip))
         dut1.write(server_ip)
         dut1.expect(re.compile(r'OK: Message from ESP32'))

examples/protocols/sockets/udp_client/sdkconfig.ci

@@ -1 +1,2 @@
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
 CONFIG_EXAMPLE_SOCKET_IP_INPUT_STDIN=y

examples/protocols/sockets/udp_server/example_test.py

@@ -14,10 +14,10 @@ import socket
 import sys
 
 import ttfw_idf
+from common_test_methods import get_env_config, get_my_interface_by_dest_ip
 
 # -----------  Config  ----------
 PORT = 3333
-INTERFACE = 'eth0'
 # -------------------------------
 
 
@@ -49,7 +49,7 @@ def udp_client(address, payload):
     return reply.decode()
 
 
-@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
+@ttfw_idf.idf_example_test(env_tag='wifi_router')
 def test_examples_protocol_socket_udpserver(env, extra_data):
     MESSAGE = 'Data to ESP'
     MAX_RETRIES = 3
@@ -59,6 +59,11 @@ def test_examples_protocol_socket_udpserver(env, extra_data):
       2. have the board connect to the server
       3. send and receive data
     """
+    # get env config
+    env_config = get_env_config('wifi_router')
+    ap_ssid = env_config['ap_ssid']
+    ap_password = env_config['ap_password']
+
     dut1 = env.get_dut('udp_server', 'examples/protocols/sockets/udp_server', dut_class=ttfw_idf.ESP32DUT)
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, 'udp_server.bin')
@@ -67,6 +72,8 @@ def test_examples_protocol_socket_udpserver(env, extra_data):
 
     # start test
     dut1.start_app()
+    dut1.expect('Please input ssid password:')
+    dut1.write(' '.join([ap_ssid, ap_password]))
 
     ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
     ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8)    # expect all 8 octets from IPv6 (assumes it's printed in the long form)
@@ -74,6 +81,7 @@ def test_examples_protocol_socket_udpserver(env, extra_data):
     print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
     dut1.expect(re.compile(r'Waiting for data'), timeout=10)
 
+    interface = get_my_interface_by_dest_ip(ipv4)
     # test IPv4
     for _ in range(MAX_RETRIES):
         print('Testing UDP on IPv4...')
@@ -88,7 +96,7 @@ def test_examples_protocol_socket_udpserver(env, extra_data):
     # test IPv6
     for _ in range(MAX_RETRIES):
         print('Testing UDP on IPv6...')
-        received = udp_client('{}%{}'.format(ipv6, INTERFACE), MESSAGE)
+        received = udp_client('{}%{}'.format(ipv6, interface), MESSAGE)
         if received == MESSAGE:
             print('OK')
             break

examples/protocols/sockets/udp_server/sdkconfig.ci

@@ -1,2 +1,3 @@
+CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
 CONFIG_EXAMPLE_IPV4=y
 CONFIG_EXAMPLE_IPV6=y

examples/system/ota/advanced_https_ota/pytest_advanced_ota.py

@@ -1,5 +1,6 @@
 # SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Unlicense OR CC0-1.0
+from ast import Try
 import http.server
 import multiprocessing
 import os
@@ -12,6 +13,7 @@ from typing import Callable
 
 import pexpect
 import pytest
+from common_test_methods import get_my_ip4_by_dest_ip
 from pytest_embedded import Dut
 from RangeHTTPServer import RangeRequestHandler
 
@@ -19,15 +21,6 @@ server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_cer
 key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_key.pem')
 
 
-def get_my_ip() -> str:
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = ''
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
     """
     Returns a request handler class that handles broken pipe exception
@@ -104,86 +97,86 @@ def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int,
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example(dut: Dut) -> None:
     """
     This is a positive test case, which downloads complete binary file multiple number of times.
     Number of iterations can be specified in variable iterations.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
-    try:
-        # Number of iterations to validate OTA
-        iterations = 3
-        server_port = 8001
-        bin_name = 'advanced_https_ota.bin'
-        # start test
-        host_ip = get_my_ip()
+    # Number of iterations to validate OTA
+    iterations = 3
+    server_port = 8001
+    bin_name = 'advanced_https_ota.bin'
+    # start test
+    for _ in range(iterations):
+        dut.expect('Loaded app from partition at offset', timeout=30)
+        try:
+            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+        except pexpect.exceptions.TIMEOUT:
+            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        dut.expect('Starting Advanced OTA example', timeout=30)
+
+        host_ip = get_my_ip4_by_dest_ip(ip_address)
         thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
         thread1.daemon = True
         thread1.start()
-        for i in range(iterations):
-            dut.expect('Loaded app from partition at offset', timeout=30)
-            try:
-                ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-                print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-            except pexpect.exceptions.TIMEOUT:
-                thread1.terminate()
-                raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-            dut.expect('Starting Advanced OTA example', timeout=30)
-
+        try:
             print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
             dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
-    finally:
-        thread1.terminate()
+        finally:
+            thread1.terminate()
 
 
 @pytest.mark.esp32
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) -> None:
     """
     Working of OTA if binary file is truncated is validated in this test case.
     Application should return with error message in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate truncated binary file
       3. Fetch OTA image over HTTPS
       4. Check working of code if bin is truncated
     """
+    server_port = 8001
+    # Original binary file generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # Truncated binary file to be generated from original binary file
+    truncated_bin_name = 'truncated.bin'
+    # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
+    # truncated_bin_size is set to 64000 to reduce consumed by the test case
+    truncated_bin_size = 64000
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    with open(binary_file, 'rb+') as f:
+        with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
+            fo.write(f.read(truncated_bin_size))
+
+    binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # Original binary file generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # Truncated binary file to be generated from original binary file
-        truncated_bin_name = 'truncated.bin'
-        # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
-        # truncated_bin_size is set to 64000 to reduce consumed by the test case
-        truncated_bin_size = 64000
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        with open(binary_file, 'rb+') as f:
-            with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
-                fo.write(f.read(truncated_bin_size))
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    dut.expect('Starting Advanced OTA example', timeout=30)
 
-        binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting Advanced OTA example', timeout=30)
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
         dut.expect('Image validation failed, image is corrupted', timeout=30)
@@ -199,46 +192,46 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) ->
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut) -> None:
     """
     Working of OTA if headers of binary file are truncated is vaildated in this test case.
     Application should return with error message in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate binary file with truncated headers
       3. Fetch OTA image over HTTPS
       4. Check working of code if headers are not sent completely
     """
+    server_port = 8001
+    # Original binary file generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # Truncated binary file to be generated from original binary file
+    truncated_bin_name = 'truncated_header.bin'
+    # Size of truncated file to be generated. This value should be less than 288 bytes (Image header size)
+    truncated_bin_size = 180
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    with open(binary_file, 'rb+') as f:
+        with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
+            fo.write(f.read(truncated_bin_size))
+
+    binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # Original binary file generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # Truncated binary file to be generated from original binary file
-        truncated_bin_name = 'truncated_header.bin'
-        # Size of truncated file to be generated. This value should be less than 288 bytes (Image header size)
-        truncated_bin_size = 180
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        with open(binary_file, 'rb+') as f:
-            with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
-                fo.write(f.read(truncated_bin_size))
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
-        binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
         dut.expect('advanced_https_ota_example: esp_https_ota_read_img_desc failed', timeout=30)
@@ -254,46 +247,46 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut)
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
     """
     Working of OTA if random data is added in binary file are validated in this test case.
     Magic byte verification should fail in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate random binary image
       3. Fetch OTA image over HTTPS
       4. Check working of code for random binary file
     """
+    server_port = 8001
+    # Random binary file to be generated
+    random_bin_name = 'random.bin'
+    # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
+    random_bin_size = 32000
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, random_bin_name)
+    with open(binary_file, 'wb+') as fo:
+        # First byte of binary file is always set to zero. If first byte is generated randomly,
+        # in some cases it may generate 0xE9 which will result in failure of testcase.
+        fo.write(struct.pack('B', 0))
+        for i in range(random_bin_size - 1):
+            fo.write(struct.pack('B', random.randrange(0,255,1)))
+
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # Random binary file to be generated
-        random_bin_name = 'random.bin'
-        # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
-        random_bin_size = 32000
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, random_bin_name)
-        with open(binary_file, 'wb+') as fo:
-            # First byte of binary file is always set to zero. If first byte is generated randomly,
-            # in some cases it may generate 0xE9 which will result in failure of testcase.
-            fo.write(struct.pack('B', 0))
-            for i in range(random_bin_size - 1):
-                fo.write(struct.pack('B', random.randrange(0,255,1)))
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
         dut.expect(r'esp_https_ota: Incorrect app descriptor magic', timeout=10)
@@ -309,48 +302,48 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut) -> None:
     """
     Working of OTA if binary file have invalid chip id is validated in this test case.
     Chip id verification should fail in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate binary image with invalid chip id
       3. Fetch OTA image over HTTPS
       4. Check working of code for random binary file
     """
+    server_port = 8001
+    bin_name = 'advanced_https_ota.bin'
+    # Random binary file to be generated
+    random_bin_name = 'random.bin'
+    random_binary_file = os.path.join(dut.app.binary_path, random_bin_name)
+    # Size of random binary file. 2000 is choosen, to reduce the time required to run the test-case
+    random_bin_size = 2000
+
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    with open(binary_file, 'rb+') as f:
+        data = list(f.read(random_bin_size))
+    # Changing Chip id
+    data[13] = 0xfe
+    with open(random_binary_file, 'wb+') as fo:
+        fo.write(bytearray(data))
+
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        bin_name = 'advanced_https_ota.bin'
-        # Random binary file to be generated
-        random_bin_name = 'random.bin'
-        random_binary_file = os.path.join(dut.app.binary_path, random_bin_name)
-        # Size of random binary file. 2000 is choosen, to reduce the time required to run the test-case
-        random_bin_size = 2000
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        with open(binary_file, 'rb+') as f:
-            data = list(f.read(random_bin_size))
-        # Changing Chip id
-        data[13] = 0xfe
-        with open(random_binary_file, 'wb+') as fo:
-            fo.write(bytearray(data))
-
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
         dut.expect(r'esp_https_ota: Mismatch chip id, expected 0, found \d', timeout=10)
@@ -366,31 +359,31 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut)
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
     """
     This is a positive test case, which downloads complete binary file multiple number of times.
     Number of iterations can be specified in variable iterations.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
     # File to be downloaded. This file is generated after compilation
     bin_name = 'advanced_https_ota.bin'
     # start test
-    host_ip = get_my_ip()
-    chunked_server = start_chunked_server(dut.app.binary_path, 8070)
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    chunked_server = start_chunked_server(dut.app.binary_path, 8070)
 
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
         dut.write('https://' + host_ip + ':8070/' + bin_name)
         dut.expect('Loaded app from partition at offset', timeout=60)
@@ -403,46 +396,45 @@ def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) -> None:
     """
     This is a positive test case, which starts a server and a redirection server.
     Redirection server redirects http_request to different port
     Number of iterations can be specified in variable iterations.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    server_port = 8001
+    # Port to which the request should be redirected
+    redirection_server_port = 8081
+    redirection_server_port1 = 8082
+    # File to be downloaded. This file is generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # Port to which the request should be redirected
-        redirection_server_port = 8081
-        redirection_server_port1 = 8082
-        # File to be downloaded. This file is generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port, redirection_server_port1))
-        thread2.daemon = True
-        thread2.start()
-        thread3 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port1, server_port))
-        thread3.daemon = True
-        thread3.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            thread2.terminate()
-            thread3.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting Advanced OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    dut.expect('Starting Advanced OTA example', timeout=30)
 
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port, redirection_server_port1))
+    thread2.daemon = True
+    thread3 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port1, server_port))
+    thread3.daemon = True
+
+    try:
+        thread1.start()
+        thread2.start()
+        thread3.start()
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name))
         dut.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
         dut.expect('Loaded app from partition at offset', timeout=60)
@@ -465,49 +457,49 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
     Working of OTA when anti_rollback is enabled and security version of new image is less than current one.
     Application should return with error message in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate binary file with lower security version
       3. Fetch OTA image over HTTPS
       4. Check working of anti_rollback feature
     """
+    dut.serial.erase_flash()
+    dut.serial.flash()
+    server_port = 8001
+    # Original binary file generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # Modified firmware image to lower security version in its header. This is to enable negative test case
+    anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin'
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    file_size = os.path.getsize(binary_file)
+    with open(binary_file, 'rb+') as f:
+        with open(os.path.join(dut.app.binary_path, anti_rollback_bin_name), 'wb+') as fo:
+            fo.write(f.read(file_size))
+            # Change security_version to 0 for negative test case
+            fo.seek(36)
+            fo.write(b'\x00')
+    binary_file = os.path.join(dut.app.binary_path, anti_rollback_bin_name)
+    # start test
+    # Positive Case
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        dut.serial.erase_flash()
-        dut.serial.flash()
-        server_port = 8001
-        # Original binary file generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # Modified firmware image to lower security version in its header. This is to enable negative test case
-        anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin'
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        file_size = os.path.getsize(binary_file)
-        with open(binary_file, 'rb+') as f:
-            with open(os.path.join(dut.app.binary_path, anti_rollback_bin_name), 'wb+') as fo:
-                fo.write(f.read(file_size))
-                # Change security_version to 0 for negative test case
-                fo.seek(36)
-                fo.write(b'\x00')
-        binary_file = os.path.join(dut.app.binary_path, anti_rollback_bin_name)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        # Positive Case
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting Advanced OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
+        dut.expect('Starting Advanced OTA example', timeout=30)
         # Use originally generated image with secure_version=1
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
         dut.expect('Loaded app from partition at offset', timeout=60)
-        dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
+        dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
         dut.expect(r'App is valid, rollback cancelled successfully', timeout=30)
 
         # Negative Case
@@ -528,47 +520,47 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 @pytest.mark.parametrize('config', ['partial_download',], indirect=True)
 def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut) -> None:
     """
     This is a positive test case, to test OTA workflow with Range HTTP header.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    server_port = 8001
+    # Size of partial HTTP request
+    request_size = 16384
+    # File to be downloaded. This file is generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    bin_size = os.path.getsize(binary_file)
+    http_requests = int((bin_size / request_size) - 1)
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # Size of partial HTTP request
-        request_size = 16384
-        # File to be downloaded. This file is generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        bin_size = os.path.getsize(binary_file)
-        http_requests = int((bin_size / request_size) - 1)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            print('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-            thread1.terminate()
-            raise
-        dut.expect('Starting Advanced OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        print('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        raise
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
+        dut.expect('Starting Advanced OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
         for _ in range(http_requests):
             dut.expect('Connection closed', timeout=60)
         dut.expect('Loaded app from partition at offset', timeout=60)
         dut.expect('Starting Advanced OTA example', timeout=30)
-    finally:
+    except:
         thread1.terminate()
 
 
@@ -583,27 +575,29 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) ->
     """
     Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using NimBLE Host stack.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Run BLE advertise and then GATT server.
       3. Fetch OTA image over HTTPS
       4. Reboot with the new OTA image
     """
+    server_port = 8001
+    # File to be downloaded. This file is generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # File to be downloaded. This file is generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
 
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
+
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
         print('Started GAP advertising.')
@@ -611,7 +605,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) ->
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
         dut.expect('Loaded app from partition at offset', timeout=60)
         dut.expect('Starting Advanced OTA example', timeout=30)
-    finally:
+    except:
         thread1.terminate()
 
 
@@ -626,28 +620,29 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut)
     """
     Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using Bluedroid Host stack.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Run BLE advertise and then GATT server.
       3. Fetch OTA image over HTTPS
       4. Reboot with the new OTA image
     """
+    server_port = 8001
+    # File to be downloaded. This file is generated after compilation
+    bin_name = 'advanced_https_ota.bin'
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # File to be downloaded. This file is generated after compilation
-        bin_name = 'advanced_https_ota.bin'
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
 
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
+
+    try:
         dut.expect('Starting Advanced OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
         dut.expect('Started advertising.', timeout=30)
@@ -664,12 +659,12 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut)
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: Dut) -> None:
     """
     This is a test case for esp_http_client_read with binary size multiple of 289 bytes
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
@@ -689,18 +684,18 @@ def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: D
             for _ in range(dummy_data_size):
                 fo.write(struct.pack('B', random.randrange(0,255,1)))
     # start test
-    host_ip = get_my_ip()
+    dut.expect('Loaded app from partition at offset', timeout=30)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
     chunked_server = start_chunked_server(dut.app.binary_path, 8070)
     try:
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-
         dut.expect('Starting Advanced OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8070/' + aligned_bin_name))
         dut.write('https://' + host_ip + ':8070/' + aligned_bin_name)
         dut.expect('Loaded app from partition at offset', timeout=60)

examples/system/ota/native_ota_example/pytest_native_ota.py

@@ -12,6 +12,7 @@ from typing import Callable, Tuple
 
 import pexpect
 import pytest
+from common_test_methods import get_my_ip4_by_dest_ip
 from pytest_embedded import Dut
 
 server_cert = '-----BEGIN CERTIFICATE-----\n' \
@@ -65,15 +66,6 @@ server_key = '-----BEGIN PRIVATE KEY-----\n'\
              '-----END PRIVATE KEY-----\n'
 
 
-def get_my_ip() -> str:
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = ''
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def create_file(server_file: str, file_data: str) -> None:
     with open(server_file, 'w+') as file:
         file.write(file_data)
@@ -130,86 +122,84 @@ def start_chunked_server(ota_image_dir: str, server_port: int) -> subprocess.Pop
 
 
 @pytest.mark.supported_targets
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_native_ota_example(dut: Dut) -> None:
     """
     This is a positive test case, which downloads complete binary file multiple number of times.
     Number of iterations can be specified in variable iterations.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
-    try:
-        server_port = 8002
-        # No. of times working of application to be validated
-        iterations = 3
-        # File to be downloaded. This file is generated after compilation
-        bin_name = 'native_ota.bin'
-        # start test
-        host_ip = get_my_ip()
+    server_port = 8002
+    # No. of times working of application to be validated
+    iterations = 3
+    # File to be downloaded. This file is generated after compilation
+    bin_name = 'native_ota.bin'
+    # start test
+    for _ in range(iterations):
+        dut.expect('Loaded app from partition at offset', timeout=60)
+        try:
+            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+        except pexpect.exceptions.TIMEOUT:
+            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        # Start server
+        host_ip = get_my_ip4_by_dest_ip(ip_address)
         thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
         thread1.daemon = True
         thread1.start()
-        for i in range(iterations):
-            dut.expect('Loaded app from partition at offset', timeout=60)
-            try:
-                ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-                print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-            except pexpect.exceptions.TIMEOUT:
-                thread1.terminate()
-                raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+        try:
             dut.expect('Starting OTA example', timeout=30)
-
             print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
             dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
-    finally:
-        thread1.terminate()
+        finally:
+            thread1.terminate()
 
 
 @pytest.mark.supported_targets
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
     """
     Working of OTA if binary file is truncated is validated in this test case.
     Application should return with error message in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate truncated binary file
       3. Fetch OTA image over HTTPS
       4. Check working of code if bin is truncated
     """
+    server_port = 8002
+    # Original binary file generated after compilation
+    bin_name = 'native_ota.bin'
+    # Truncated binary file to be generated from original binary file
+    truncated_bin_name = 'truncated.bin'
+    # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
+    # truncated_bin_size is set to 64000 to reduce consumed by the test case
+    truncated_bin_size = 64000
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    f = open(binary_file, 'rb+')
+    fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
+    fo.write(f.read(truncated_bin_size))
+    fo.close()
+    f.close()
+    binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
     try:
-        server_port = 8002
-        # Original binary file generated after compilation
-        bin_name = 'native_ota.bin'
-        # Truncated binary file to be generated from original binary file
-        truncated_bin_name = 'truncated.bin'
-        # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
-        # truncated_bin_size is set to 64000 to reduce consumed by the test case
-        truncated_bin_size = 64000
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        f = open(binary_file, 'rb+')
-        fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
-        fo.write(f.read(truncated_bin_size))
-        fo.close()
-        f.close()
-        binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
         dut.expect('native_ota_example: Image validation failed, image is corrupted', timeout=20)
@@ -219,47 +209,46 @@ def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
 
 
 @pytest.mark.supported_targets
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None:
     """
     Working of OTA if headers of binary file are truncated is vaildated in this test case.
     Application should return with error message in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate binary file with truncated headers
       3. Fetch OTA image over HTTPS
       4. Check working of code if headers are not sent completely
     """
+    server_port = 8002
+    # Original binary file generated after compilation
+    bin_name = 'native_ota.bin'
+    # Truncated binary file to be generated from original binary file
+    truncated_bin_name = 'truncated_header.bin'
+    # Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
+    truncated_bin_size = 180
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, bin_name)
+    f = open(binary_file, 'rb+')
+    fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
+    fo.write(f.read(truncated_bin_size))
+    fo.close()
+    f.close()
+    binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
     try:
-        server_port = 8002
-        # Original binary file generated after compilation
-        bin_name = 'native_ota.bin'
-        # Truncated binary file to be generated from original binary file
-        truncated_bin_name = 'truncated_header.bin'
-        # Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
-        truncated_bin_size = 180
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, bin_name)
-        f = open(binary_file, 'rb+')
-        fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
-        fo.write(f.read(truncated_bin_size))
-        fo.close()
-        f.close()
-        binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
         dut.expect('native_ota_example: received package is not fit len', timeout=20)
@@ -269,46 +258,46 @@ def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None
 
 
 @pytest.mark.supported_targets
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
     """
     Working of OTA if random data is added in binary file are validated in this test case.
     Magic byte verification should fail in this case.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Generate random binary image
       3. Fetch OTA image over HTTPS
       4. Check working of code for random binary file
     """
+    server_port = 8002
+    # Random binary file to be generated
+    random_bin_name = 'random.bin'
+    # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
+    random_bin_size = 32000
+    # check and log bin size
+    binary_file = os.path.join(dut.app.binary_path, random_bin_name)
+    fo = open(binary_file, 'wb+')
+    # First byte of binary file is always set to zero. If first byte is generated randomly,
+    # in some cases it may generate 0xE9 which will result in failure of testcase.
+    fo.write(struct.pack('B', 0))
+    for i in range(random_bin_size - 1):
+        fo.write(struct.pack('B', random.randrange(0,255,1)))
+    fo.close()
+    # start test
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8002
-        # Random binary file to be generated
-        random_bin_name = 'random.bin'
-        # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
-        random_bin_size = 32000
-        # check and log bin size
-        binary_file = os.path.join(dut.app.binary_path, random_bin_name)
-        fo = open(binary_file, 'wb+')
-        # First byte of binary file is always set to zero. If first byte is generated randomly,
-        # in some cases it may generate 0xE9 which will result in failure of testcase.
-        fo.write(struct.pack('B', 0))
-        for i in range(random_bin_size - 1):
-            fo.write(struct.pack('B', random.randrange(0,255,1)))
-        fo.close()
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
+        dut.expect('Starting OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
         dut.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=20)
@@ -318,29 +307,29 @@ def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
 
 
 @pytest.mark.supported_targets
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_native_ota_example_chunked(dut: Dut) -> None:
     """
     This is a positive test case, which downloads complete binary file multiple number of times.
     Number of iterations can be specified in variable iterations.
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
     # File to be downloaded. This file is generated after compilation
     bin_name = 'native_ota.bin'
     # start test
-    host_ip = get_my_ip()
+    dut.expect('Loaded app from partition at offset', timeout=30)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
     chunked_server = start_chunked_server(dut.app.binary_path, 8070)
     try:
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-
         dut.expect('Starting OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
         dut.write('https://' + host_ip + ':8070/' + bin_name)

examples/system/ota/pre_encrypted_ota/pytest_pre_encrypted_ota.py

@@ -9,6 +9,7 @@ from typing import Callable
 
 import pexpect
 import pytest
+from common_test_methods import get_my_ip4_by_dest_ip
 from pytest_embedded import Dut
 from RangeHTTPServer import RangeRequestHandler
 
@@ -17,15 +18,6 @@ key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_cert
 enc_bin_name = 'pre_encrypted_ota_secure.bin'
 
 
-def get_my_ip() -> str:
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = ''
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
     """
     Returns a request handler class that handles broken pipe exception
@@ -64,25 +56,24 @@ def start_https_server(ota_image_dir: str, server_ip: str, server_port: int) ->
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 def test_examples_protocol_pre_encrypted_ota_example(dut: Dut) -> None:
+    server_port = 8001
+    dut.expect('Loaded app from partition at offset', timeout=30)
     try:
-        server_port = 8001
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
-        thread1.daemon = True
-        thread1.start()
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
+    thread1.daemon = True
+    thread1.start()
 
-        dut.expect('Loaded app from partition at offset', timeout=30)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    try:
         dut.expect('Starting Pre Encrypted OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name))
         dut.write('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name)
         dut.expect('Magic Verified', timeout=30)

examples/system/ota/simple_ota_example/pytest_simple_ota.py

@@ -10,6 +10,7 @@ from typing import Tuple
 
 import pexpect
 import pytest
+from common_test_methods import get_my_ip4_by_dest_ip
 from pytest_embedded import Dut
 
 server_cert = '-----BEGIN CERTIFICATE-----\n' \
@@ -63,15 +64,6 @@ server_key = '-----BEGIN PRIVATE KEY-----\n'\
              '-----END PRIVATE KEY-----\n'
 
 
-def get_my_ip() -> str:
-    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s1.connect(('8.8.8.8', 80))
-    my_ip = ''
-    my_ip = s1.getsockname()[0]
-    s1.close()
-    return my_ip
-
-
 def start_https_server(ota_image_dir: str, server_ip: str, server_port: int, server_file: str = None, key_file: str = None) -> None:
     os.chdir(ota_image_dir)
 
@@ -123,28 +115,28 @@ def calc_all_sha256(dut: Dut) -> Tuple[str, str]:
 def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    sha256_bootloader, sha256_app = calc_all_sha256(dut)
+    # start test
+    dut.expect('Loaded app from partition at offset 0x10000', timeout=30)
+    check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
+    check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
     try:
-        sha256_bootloader, sha256_app = calc_all_sha256(dut)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x10000', timeout=30)
-        check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
-        check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
@@ -157,30 +149,30 @@ def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 @pytest.mark.parametrize('config', ['spiram',], indirect=True)
 def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    # start test
+    dut.expect('Loaded app from partition at offset 0x10000', timeout=30))
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
     try:
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x10000', timeout=30))
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
@@ -199,29 +191,29 @@ def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(dut: D
 def test_examples_protocol_simple_ota_example_with_flash_encryption(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    # Erase flash
+    dut.serial.erase_flash()
+    dut.serial.flash()
+    # start test
+    dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
+    dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
     try:
-        # Erase flash
-        dut.serial.erase_flash()
-        dut.serial.flash()
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
-        dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
@@ -239,29 +231,29 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(dut: Dut) ->
 def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    # start test
+    # Erase flash
+    dut.serial.erase_flash()
+    dut.serial.flash()
+    dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
+    dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
+    try:
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
     try:
-        # start test
-        # Erase flash
-        dut.serial.erase_flash()
-        dut.serial.flash()
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
-        dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
         dut.expect('Starting OTA example', timeout=30)
-
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
@@ -275,39 +267,38 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(dut: Du
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet_ota
+@pytest.mark.ethernet_router
 @pytest.mark.parametrize('config', ['on_update_no_sb_ecdsa',], indirect=True)
 def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_ecdsa(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    sha256_bootloader, sha256_app = calc_all_sha256(dut)
+    # start test
+    dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
+    check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
+    check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
     try:
-        sha256_bootloader, sha256_app = calc_all_sha256(dut)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
-        check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
-        check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
+        dut.expect('Starting OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)
-
         dut.expect('Verifying image signature...', timeout=60)
-
         dut.expect('Loaded app from partition at offset 0x120000', timeout=20)
         dut.expect('Starting OTA example', timeout=30)
     finally:
@@ -318,33 +309,34 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat
 @pytest.mark.esp32c3
 @pytest.mark.esp32s2
 @pytest.mark.esp32s3
-@pytest.mark.ethernet
+@pytest.mark.ethernet_router
 @pytest.mark.parametrize('config', ['on_update_no_sb_rsa',], indirect=True)
 def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_rsa(dut: Dut) -> None:
     """
     steps: |
-      1. join AP
+      1. join AP/Ethernet
       2. Fetch OTA image over HTTPS
       3. Reboot with the new OTA image
     """
+    sha256_bootloader, sha256_app = calc_all_sha256(dut)
+    # start test
+    dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
+    check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
+    check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
     try:
-        sha256_bootloader, sha256_app = calc_all_sha256(dut)
-        # start test
-        host_ip = get_my_ip()
-        thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
-        thread1.daemon = True
-        thread1.start()
-        dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
-        check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
-        check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
-        try:
-            ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
-            print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
-        except pexpect.exceptions.TIMEOUT:
-            thread1.terminate()
-            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
-        dut.expect('Starting OTA example', timeout=30)
+        ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
+        print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
+    except pexpect.exceptions.TIMEOUT:
+        thread1.terminate()
+        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+    # Start server
+    host_ip = get_my_ip4_by_dest_ip(ip_address)
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
+    thread1.daemon = True
+    thread1.start()
 
+    try:
+        dut.expect('Starting OTA example', timeout=30)
         print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
         dut.write('https://' + host_ip + ':8000/simple_ota.bin')
         dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)

pytest.ini

@@ -36,7 +36,6 @@ markers =
   quad_psram: runners with quad psram
   octal_psram: runners with octal psram
   usb_host: usb host runners
-  ethernet_ota: ethernet OTA runners
   flash_encryption: Flash Encryption runners
   ir_transceiver: runners with a pair of IR transmitter and receiver
   wifi_ota: wifi OTA runners
@@ -44,15 +43,17 @@ markers =
   flash_encryption_wifi_ota: flash encryprion ota wifi runner
   ethernet: ethernet runner
   ethernet_flash_8m: ethernet runner with 8mb flash
-  wifi: wifi runner
-  wifi_bt: wifi runner with bluetooth
+  ethernet_router: both the runner and dut connect to the same router
+  wifi_nearby: runner with a wifi AP nearby
+  wifi_router: runner can connect to the dut by a wifi router
+  wifi_wlan: wifi runner with a wireless NIC
   deepsleep_temp_tag: temporary env for running potentially harmfull deepsleep related tests
 
   # multi-dut markers
   multi_dut_generic: tests should be run on generic runners, at least have two duts connected.
 
   # host_test markers
-  host_test: tests which shouldn't be built at the build stage, and instead built in host_test stage.
+  host_test: tests which shouldn not be built at the build stage, and instead built in host_test stage.
   qemu: build and test using qemu-system-xtensa, not real target.
 
 # log related

tools/test_apps/protocols/mqtt/publish_connect_test/app_test.py

@@ -15,6 +15,7 @@ from threading import Event, Lock, Thread
 
 import paho.mqtt.client as mqtt
 import ttfw_idf
+from common_test_methods import get_my_ip4_by_dest_ip
 
 DEFAULT_MSG_SIZE = 16
 
@@ -33,19 +34,6 @@ def set_server_cert_cn(ip):
             raise('openssl command {} failed'.format(args))
 
 
-def get_my_ip():
-    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    try:
-        # doesn't even have to be reachable
-        s.connect(('10.255.255.255', 1))
-        IP = s.getsockname()[0]
-    except Exception:
-        IP = '127.0.0.1'
-    finally:
-        s.close()
-    return IP
-
-
 # Publisher class creating a python client to send/receive published data from esp-mqtt client
 class MqttPublisher:
 
@@ -247,8 +235,8 @@ class TlsServer:
             self.shutdown.set()
 
 
-def connection_tests(dut, cases):
-    ip = get_my_ip()
+def connection_tests(dut, cases, dut_ip):
+    ip = get_my_ip4_by_dest_ip(dut_ip)
     set_server_cert_cn(ip)
     server_port = 2222
 
@@ -314,7 +302,7 @@ def connection_tests(dut, cases):
     teardown_connection_suite()
 
 
-@ttfw_idf.idf_custom_test(env_tag='Example_EthKitV1', group='test-apps')
+@ttfw_idf.idf_custom_test(env_tag='ethernet_router', group='test-apps')
 def test_app_protocol_mqtt_publish_connect(env, extra_data):
     """
     steps:
@@ -348,11 +336,11 @@ def test_app_protocol_mqtt_publish_connect(env, extra_data):
         raise
 
     dut1.start_app()
-    esp_ip = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)
-    print('Got IP={}'.format(esp_ip[0]))
+    esp_ip = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
+    print('Got IP={}'.format(esp_ip))
 
     if not os.getenv('MQTT_SKIP_CONNECT_TEST'):
-        connection_tests(dut1,cases)
+        connection_tests(dut1,cases,esp_ip)
 
     #
     # start publish tests only if enabled in the environment (for weekend tests only)