Merge branch 'bugfix/python3_target_test_4.2' into 'release/v4.2'

ci: run target test with python3

See merge request espressif/esp-idf!12755

examples/bluetooth/nimble/bleprph/bleprph_test.py

@@ -89,7 +89,7 @@ def bleprph_client_task(prph_obj, dut, dut_addr):
     - write 'A' to characteristic with write permission
     '''
     chars_ret_on_write = {}
-    chars_ret_on_write = ble_client_obj.write_chars('A')
+    chars_ret_on_write = ble_client_obj.write_chars(b'A')
     if chars_ret_on_write:
         Utility.console_log("\nCharacteristics after write operation")
         for path, props in chars_ret_on_write.items():

examples/protocols/http_server/simple/http_server_simple_test.py

@@ -100,12 +100,6 @@ def test_examples_protocol_http_server_simple(env, extra_data):
         raise RuntimeError
     dut1.expect("Found URL query => " + query, timeout=30)
 
-    query = "abcd\nyz"
-    Utility.console_log("Test /hello with invalid query")
-    if client.test_custom_uri_query(got_ip, got_port, query):
-        raise RuntimeError
-    dut1.expect("400 Bad Request - Server unable to understand request due to invalid syntax", timeout=30)
-
 
 if __name__ == '__main__':
     test_examples_protocol_http_server_simple()

examples/protocols/sockets/tcp_server/example_test.py

@@ -36,13 +36,13 @@ def tcp_client(address, payload):
         print('Could not open socket: ', msg)
         sock.close()
         raise
-    sock.sendall(payload)
+    sock.sendall(payload.encode())
     data = sock.recv(1024)
     if not data:
         return
     print('Reply : ' + data.decode())
     sock.close()
-    return data
+    return data.decode()
 
 
 @ttfw_idf.idf_example_test(env_tag="Example_WIFI")

examples/protocols/sockets/udp_server/example_test.py

@@ -31,7 +31,7 @@ def udp_client(address, payload):
         print('Could not create socket: ' + str(msg[0]) + ': ' + msg[1])
         raise
     try:
-        sock.sendto(payload, addr)
+        sock.sendto(payload.encode(), addr)
         reply, addr = sock.recvfrom(128)
         if not reply:
             return
@@ -40,7 +40,7 @@ def udp_client(address, payload):
         print('Error Code : ' + str(msg[0]) + ' Message: ' + msg[1])
         sock.close()
         raise
-    return reply
+    return reply.decode()
 
 
 @ttfw_idf.idf_example_test(env_tag="Example_WIFI")

examples/system/ota/advanced_https_ota/example_test.py

@@ -1,5 +1,6 @@
 import re
 import os
+import struct
 import socket
 from threading import Thread
 import ssl
@@ -241,8 +242,8 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
     truncated_bin_size = 64000
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, bin_name)
-    f = open(binary_file, "r+")
-    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "w+")
+    f = open(binary_file, "rb+")
+    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
     fo.write(f.read(truncated_bin_size))
     fo.close()
     f.close()
@@ -292,8 +293,8 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr
     truncated_bin_size = 180
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, bin_name)
-    f = open(binary_file, "r+")
-    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "w+")
+    f = open(binary_file, "rb+")
+    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
     fo.write(f.read(truncated_bin_size))
     fo.close()
     f.close()
@@ -341,12 +342,12 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
     random_bin_size = 32000
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
-    fo = open(binary_file, "w+")
+    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(str(0))
+    fo.write(struct.pack("B", 0))
     for i in range(random_bin_size - 1):
-        fo.write(str(random.randrange(0,255,1)))
+        fo.write(struct.pack("B", random.randrange(0,255,1)))
     fo.close()
     bin_size = os.path.getsize(binary_file)
     ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))

examples/system/ota/native_ota_example/example_test.py

@@ -1,5 +1,6 @@
 import re
 import os
+import struct
 import socket
 from threading import Thread
 import ssl
@@ -207,8 +208,8 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
     truncated_bin_size = 64000
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, bin_name)
-    f = open(binary_file, "r+")
-    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "w+")
+    f = open(binary_file, "rb+")
+    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
     fo.write(f.read(truncated_bin_size))
     fo.close()
     f.close()
@@ -258,8 +259,8 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
     truncated_bin_size = 180
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, bin_name)
-    f = open(binary_file, "r+")
-    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "w+")
+    f = open(binary_file, "rb+")
+    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
     fo.write(f.read(truncated_bin_size))
     fo.close()
     f.close()
@@ -307,12 +308,12 @@ def test_examples_protocol_native_ota_example_random(env, extra_data):
     random_bin_size = 32000
     # check and log bin size
     binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
-    fo = open(binary_file, "w+")
+    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(str(0))
+    fo.write(struct.pack("B", 0))
     for i in range(random_bin_size - 1):
-        fo.write(str(random.randrange(0,255,1)))
+        fo.write(struct.pack("B", random.randrange(0,255,1)))
     fo.close()
     bin_size = os.path.getsize(binary_file)
     ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))

tools/ci/config/target-test.yml

@@ -34,6 +34,7 @@
     LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS"
     ENV_FILE: "$CI_PROJECT_DIR/ci-test-runner-configs/$CI_RUNNER_DESCRIPTION/EnvConfig.yml"
     SUBMODULES_TO_FETCH: "components/esptool_py/esptool"
+    PYTHON_VER: 3
   script:
     - *define_config_file_name
     # first test if config file exists, if not exist, exit 0

tools/ci/setup_python.sh

@@ -1,14 +1,7 @@
 #!/usr/bin/env bash
 
-# Regexp for matching job names which are incompatible with Python 3
-# - UT_009_ - multi-device tests are not compatible
-# - UT_014_ - multi-device tests are not compatible
-# - UT_017_ - multi-device tests are not compatible
-py3_incomp='UT_009_|UT_013_|UT_014_|UT_017_'
-
-if [ -z ${PYTHON_VER+x} ] || [[ $CI_JOB_NAME =~ $py3_incomp ]]; then
-    # Use this version of the Python interpreter if it was not defined before or
-    # the given job is not compatible with Python 3
+if [ -z ${PYTHON_VER+x} ]; then
+    # Use this version of the Python interpreter if it was not defined before
     PYTHON_VER=2.7.15
 fi
 

tools/esp_prov/__init__.py

@@ -1 +1 @@
-from esp_prov import *  # noqa: export esp_prov module to users
+from .esp_prov import *  # noqa: export esp_prov module to users

tools/esp_prov/esp_prov.py

@@ -16,7 +16,7 @@
 #
 
 from __future__ import print_function
-from builtins import input
+from builtins import input as binput
 import argparse
 import textwrap
 import time
@@ -471,7 +471,7 @@ if __name__ == '__main__':
 
             while True:
                 try:
-                    select = int(input("Select AP by number (0 to rescan) : "))
+                    select = int(binput("Select AP by number (0 to rescan) : "))
                     if select < 0 or select > len(APs):
                         raise ValueError
                     break

tools/esp_prov/prov/custom_prov.py

@@ -48,7 +48,7 @@ def custom_config_response(security_ctx, response_data):
 
 def custom_data_request(security_ctx, data):
     # Encrypt the custom data
-    enc_cmd = security_ctx.encrypt_data(data)
+    enc_cmd = security_ctx.encrypt_data(tobytes(data))
     print_verbose(security_ctx, "Client -> Device (CustomData cmd) " + utils.str_to_hexstr(enc_cmd))
     return enc_cmd
 

tools/esp_prov/security/security1.py

@@ -167,7 +167,7 @@ class Security1(Security):
             return -1
 
     def encrypt_data(self, data):
-        return self.cipher.update(data)
+        return self.cipher.update(tobytes(data))
 
     def decrypt_data(self, data):
-        return self.cipher.update(data)
+        return self.cipher.update(tobytes(data))

tools/esp_prov/transport/ble_cli.py

@@ -234,6 +234,8 @@ class BLE_Bluez_Client:
 
         try:
             path.WriteValue([ord(c) for c in data], {}, dbus_interface='org.bluez.GattCharacteristic1')
+        except TypeError:  # python3 compatible
+            path.WriteValue([c for c in data], {}, dbus_interface='org.bluez.GattCharacteristic1')
         except dbus.exceptions.DBusException as e:
             raise RuntimeError("Failed to write value to characteristic " + characteristic_uuid + ": " + str(e))
 

tools/esp_prov/utils/convenience.py

@@ -14,12 +14,14 @@
 #
 
 # Convenience functions for commonly used data type conversions
+import binascii
+from future.utils import tobytes
 
 
 def str_to_hexstr(string):
     # Form hexstr by appending ASCII codes (in hex) corresponding to
     # each character in the input string
-    return ''.join('{:02x}'.format(ord(c)) for c in string)
+    return binascii.hexlify(tobytes(string)).decode()
 
 
 def hexstr_to_str(hexstr):
@@ -28,4 +30,4 @@ def hexstr_to_str(hexstr):
         hexstr = '0' + hexstr
     # Interpret consecutive pairs of hex characters as 8 bit ASCII codes
     # and append characters corresponding to each code to form the string
-    return ''.join(chr(int(hexstr[2 * i: 2 * i + 2], 16)) for i in range(len(hexstr) // 2))
+    return binascii.unhexlify(tobytes(hexstr)).decode()