Merge branch 'feat/pytest_monitor_ide_integration' into 'master'

CI: Migrate monitor IDE integration test to pytest

Closes IDFCI-1383

See merge request espressif/esp-idf!19224

.gitlab/ci/target-test.yml

@@ -340,6 +340,16 @@ test_app_test_pytest_esp32_generic:
   variables:
     SETUP_TOOLS: "1"  # need gdb
 
+test_app_test_pytest_esp32_jtag:
+  extends:
+    - .pytest_test_apps_dir_template
+    - .rules:test:custom_test-esp32
+  needs:
+    - build_pytest_test_apps_esp32
+  tags: [ esp32, test_jtag_arm]
+  variables:
+    SETUP_TOOLS: "1"  # need gdb
+
 test_app_test_pytest_esp32s2_generic:
   extends:
     - .pytest_test_apps_dir_template

pytest.ini

@@ -60,6 +60,7 @@ markers =
   MSPI_F8R8: runner with Octal Flash and Octal PSRAM
   MSPI_F4R8: runner with Quad Flash and Octal PSRAM
   MSPI_F4R4: runner with Quad Flash and Quad PSRAM
+  test_jtag_arm: runner where the chip is accessible through JTAG as well
 
   # multi-dut markers
   multi_dut_generic: tests should be run on generic runners, at least have two duts connected.

tools/requirements/requirements.ci.txt

@@ -7,3 +7,4 @@ idf-build-apps
 junit_xml
 python-gitlab
 pyyaml
+SimpleWebSocketServer

tools/test_apps/system/monitor_ide_integration/app_test.py

@@ -1,93 +0,0 @@
-from __future__ import unicode_literals
-
-import glob
-import json
-import os
-import re
-import threading
-
-import ttfw_idf
-from SimpleWebSocketServer import SimpleWebSocketServer, WebSocket
-from tiny_test_fw import Utility
-
-
-class IDEWSProtocol(WebSocket):
-
-    def handleMessage(self):
-        try:
-            j = json.loads(self.data)
-        except Exception as e:
-            Utility.console_log('Server ignores error: {}'.format(e), 'orange')
-            return
-        event = j.get('event')
-        if event and 'prog' in j and ((event == 'gdb_stub' and 'port' in j) or
-                                      (event == 'coredump' and 'file' in j)):
-            payload = {'event': 'debug_finished'}
-            self.sendMessage(json.dumps(payload))
-            Utility.console_log('Server sent: {}'.format(payload))
-        else:
-            Utility.console_log('Server received: {}'.format(j), 'orange')
-
-    def handleConnected(self):
-        Utility.console_log('{} connected to server'.format(self.address))
-
-    def handleClose(self):
-        Utility.console_log('{} closed the connection'.format(self.address))
-
-
-class WebSocketServer(object):
-    HOST = '127.0.0.1'
-    PORT = 1123
-
-    def run(self):
-        server = SimpleWebSocketServer(self.HOST, self.PORT, IDEWSProtocol)
-        while not self.exit_event.is_set():
-            server.serveonce()
-
-    def __init__(self):
-        self.exit_event = threading.Event()
-        self.thread = threading.Thread(target=self.run)
-        self.thread.start()
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.exit_event.set()
-        self.thread.join(10)
-        if self.thread.is_alive():
-            Utility.console_log('Thread cannot be joined', 'orange')
-
-
-@ttfw_idf.idf_custom_test(env_tag='test_jtag_arm', group='test-apps')
-def test_monitor_ide_integration(env, extra_data):
-    config_files = glob.glob(os.path.join(os.path.dirname(__file__), 'sdkconfig.ci.*'))
-    config_names = [os.path.basename(s).replace('sdkconfig.ci.', '') for s in config_files]
-    rel_proj_path = 'tools/test_apps/system/monitor_ide_integration'
-    for name in config_names:
-        Utility.console_log('Checking config "{}"... '.format(name), 'green', end='')
-        dut = env.get_dut('panic', rel_proj_path, app_config_name=name)
-        monitor_path = os.path.join(dut.app.idf_path, 'tools/idf_monitor.py')
-        elf_path = os.path.join(dut.app.binary_path, 'panic.elf')
-        dut.start_app()
-        # Closing the DUT because we will reconnect with IDF Monitor
-        env.close_dut(dut.name)
-
-        with WebSocketServer(), ttfw_idf.CustomProcess(' '.join([monitor_path,
-                                                                 elf_path,
-                                                                 '--port', str(dut.port),
-                                                                 '--ws', 'ws://{}:{}'.format(WebSocketServer.HOST,
-                                                                                             WebSocketServer.PORT)]),
-                                                       logfile='monitor_{}.log'.format(name)) as p:
-            p.pexpect_proc.expect(re.compile(r'Guru Meditation Error'), timeout=10)
-            p.pexpect_proc.expect_exact('Communicating through WebSocket', timeout=5)
-            # "u?" is for Python 2 only in the following regular expressions.
-            # The elements of dictionary can be printed in different order depending on the Python version.
-            p.pexpect_proc.expect(re.compile(r"WebSocket sent: \{u?.*'event': u?'" + name + "'"), timeout=5)
-            p.pexpect_proc.expect_exact('Waiting for debug finished event', timeout=5)
-            p.pexpect_proc.expect(re.compile(r"WebSocket received: \{u?'event': u?'debug_finished'\}"), timeout=5)
-            p.pexpect_proc.expect_exact('Communications through WebSocket is finished', timeout=5)
-
-
-if __name__ == '__main__':
-    test_monitor_ide_integration()

tools/test_apps/system/monitor_ide_integration/pytest_monitor_ide_integration.py

@@ -0,0 +1,92 @@
+# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Unlicense OR CC0-1.0
+import json
+import logging
+import multiprocessing
+import os
+import re
+import sys
+from types import TracebackType
+from typing import Optional, Type, TypeVar
+
+import pexpect
+import pytest
+from pytest_embedded import Dut
+from SimpleWebSocketServer import SimpleWebSocketServer, WebSocket
+
+WebSocketServerType = TypeVar('WebSocketServerType')
+
+
+class IDEWSProtocol(WebSocket):
+
+    def handleMessage(self) -> None:
+        try:
+            j = json.loads(self.data)
+        except Exception as e:
+            logging.info(f'Server ignores error: {e}')
+            return
+        event = j.get('event')
+        if event and 'prog' in j and ((event == 'gdb_stub' and 'port' in j) or
+                                      (event == 'coredump' and 'file' in j)):
+            payload = {'event': 'debug_finished'}
+            self.sendMessage(json.dumps(payload))
+            logging.info(f'Server sent: {payload}')
+        else:
+            logging.info(f'Server received: {j}')
+
+    def handleConnected(self) -> None:
+        logging.info(f'{self.address} connected to server')
+
+    def handleClose(self) -> None:
+        logging.info(f'{self.address} closed the connection')
+
+
+class WebSocketServer(object):
+    HOST = '127.0.0.1'
+    PORT = 1123
+
+    def run(self) -> None:
+        server = SimpleWebSocketServer(self.HOST, self.PORT, IDEWSProtocol)
+        while not self.exit_event.is_set():
+            server.serveonce()
+
+    def __init__(self) -> None:
+        self.exit_event = multiprocessing.Event()
+        self.proc = multiprocessing.Process(target=self.run)
+        self.proc.start()
+
+    def __enter__(self: WebSocketServerType) -> WebSocketServerType:
+        return self
+
+    def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], traceback:
+                 Optional[TracebackType]) -> None:
+        self.exit_event.set()
+        self.proc.join(10)
+        if self.proc.is_alive():
+            logging.info('Process cannot be joined')
+
+
+@pytest.mark.esp32
+@pytest.mark.test_jtag_arm
+@pytest.mark.parametrize('config', ['gdb_stub', 'coredump'], indirect=True)
+def test_monitor_ide_integration(config: str, dut: Dut) -> None:
+    # The port needs to be closed because idf_monitor.py will connect to it
+    dut.serial.stop_redirect_thread()
+
+    monitor_py = os.path.join(os.environ['IDF_PATH'], 'tools', 'idf_monitor.py')
+    with open(f'monitor_{config}.log', 'w') as log:
+        monitor_cmd = ' '.join([sys.executable, monitor_py, os.path.join(dut.app.binary_path, 'panic.elf'),
+                                '--port', str(dut.serial.port),
+                                '--ws', f'ws://{WebSocketServer.HOST}:{WebSocketServer.PORT}'])
+        with WebSocketServer(), pexpect.spawn(monitor_cmd,
+                                              logfile=log,
+                                              timeout=60,
+                                              encoding='utf-8',
+                                              codec_errors='ignore') as p:
+            p.expect(re.compile(r'Guru Meditation Error'), timeout=10)
+            p.expect_exact('Communicating through WebSocket', timeout=5)
+            # The elements of dictionary can be printed in different order depending on the Python version.
+            p.expect(re.compile(r"WebSocket sent: \{.*'event': '" + config + "'"), timeout=5)
+            p.expect_exact('Waiting for debug finished event', timeout=5)
+            p.expect(re.compile(r"WebSocket received: \{'event': 'debug_finished'\}"), timeout=5)
+            p.expect_exact('Communications through WebSocket is finished', timeout=5)