CI: Use higher-level interaction with GDB in example tests and test apps

examples/storage/semihost_vfs/semihost_vfs_example_test.py

@@ -24,9 +24,9 @@ def test_examples_semihost_vfs(env, extra_data):
         temp_dir = tempfile.mkdtemp()
         host_file_path = os.path.join(proj_path, 'data', host_file_name)
         shutil.copyfile(host_file_path, os.path.join(temp_dir, host_file_name))
-        openocd_extra_args = '-c \'set ESP_SEMIHOST_BASEDIR {}\''.format(temp_dir)
+        cfg_cmds = ['set ESP_SEMIHOST_BASEDIR "{}"'.format(temp_dir)]
 
-        with ttfw_idf.OCDProcess(os.path.join(proj_path, 'openocd.log'), openocd_extra_args):
+        with ttfw_idf.OCDBackend(os.path.join(proj_path, 'openocd.log'), dut.app.target, cfg_cmds=cfg_cmds):
             dut.start_app()
             dut.expect_all('example: Switch to semihosted stdout',
                            'example: Switched back to UART stdout',

examples/system/app_trace_to_host/example_test.py

@@ -12,26 +12,22 @@ def test_examples_app_trace_to_host(env, extra_data):
     idf_path = dut.app.get_sdk_path()
     proj_path = os.path.join(idf_path, rel_project_path)
 
-    with ttfw_idf.OCDProcess(os.path.join(proj_path, 'openocd.log')):
-        with ttfw_idf.TelnetProcess(os.path.join(proj_path, 'telnet.log')) as telnet_p:
-            dut.start_app()
-            dut.expect_all('example: Enabling ADC1 on channel 6 / GPIO34.',
-                           'example: Enabling CW generator on DAC channel 1',
-                           'example: Custom divider of RTC 8 MHz clock has been set.',
-                           'example: Sampling ADC and sending data to the host...',
-                           re.compile(r'example: Collected \d+ samples in 20 ms.'),
-                           'example: Sampling ADC and sending data to the UART...',
-                           re.compile(r'example: Sample:\d, Value:\d+'),
-                           re.compile(r'example: Collected \d+ samples in 20 ms.'),
-                           timeout=20)
-
-            telnet_p.pexpect_proc.sendline('esp apptrace start file://adc.log 0 9000 5 0 0')
-            telnet_p.pexpect_proc.expect_exact('App trace params: from 2 cores, size 9000 bytes, '
-                                               'stop_tmo 5 s, poll period 0 ms, wait_rst 0, skip 0 bytes')
-            telnet_p.pexpect_proc.expect_exact('Targets connected.')
-            telnet_p.pexpect_proc.expect_exact('Targets disconnected.')
-            telnet_p.pexpect_proc.expect_exact('Tracing is STOPPED. Size is 9000 of 9000 @')
-            telnet_p.pexpect_proc.expect_exact('Data: blocks incomplete 0, lost bytes: 0')
+    with ttfw_idf.OCDBackend(os.path.join(proj_path, 'openocd.log'), dut.app.target) as ocd:
+        dut.start_app()
+        dut.expect_all('example: Enabling ADC1 on channel 6 / GPIO34.',
+                       'example: Enabling CW generator on DAC channel 1',
+                       'example: Custom divider of RTC 8 MHz clock has been set.',
+                       'example: Sampling ADC and sending data to the host...',
+                       re.compile(r'example: Collected \d+ samples in 20 ms.'),
+                       'example: Sampling ADC and sending data to the UART...',
+                       re.compile(r'example: Sample:\d, Value:\d+'),
+                       re.compile(r'example: Collected \d+ samples in 20 ms.'),
+                       timeout=20)
+
+        response = ocd.cmd_exec('esp apptrace start file://adc.log 0 9000 5 0 0')
+        with open(os.path.join(proj_path, 'telnet.log'), 'w') as f:
+            f.write(response)
+        assert('Data: blocks incomplete 0, lost bytes: 0' in response)
 
     with ttfw_idf.CustomProcess(' '.join([os.path.join(idf_path, 'tools/esp_app_trace/logtrace_proc.py'),
                                           'adc.log',

examples/system/gcov/example_test.py

@@ -1,5 +1,4 @@
 from __future__ import unicode_literals
-from pexpect import TIMEOUT
 from ttfw_idf import Utility
 import os
 import ttfw_idf
@@ -12,41 +11,46 @@ def test_examples_gcov(env, extra_data):
     dut = env.get_dut('gcov', rel_project_path)
     idf_path = dut.app.get_sdk_path()
     proj_path = os.path.join(idf_path, rel_project_path)
-
-    with ttfw_idf.OCDProcess(os.path.join(proj_path, 'openocd.log')):
-        with ttfw_idf.TelnetProcess(os.path.join(proj_path, 'telnet.log')) as telnet_p:
-            dut.start_app()
-
-            def expect_counter_output(loop, timeout=10):
-                dut.expect_all('blink_dummy_func: Counter = {}'.format(loop),
-                               'some_dummy_func: Counter = {}'.format(loop * 2),
-                               timeout=timeout)
-
-            expect_counter_output(0, timeout=20)
-            dut.expect('Ready to dump GCOV data...', timeout=5)
-
-            def dump_coverage():
-                try:
-                    telnet_p.pexpect_proc.sendline('esp gcov dump')
-                    telnet_p.pexpect_proc.expect_exact('Targets connected.')
-                    telnet_p.pexpect_proc.expect_exact('gcov_example_main.c.gcda')
-                    telnet_p.pexpect_proc.expect_exact('gcov_example_func.c.gcda')
-                    telnet_p.pexpect_proc.expect_exact('some_funcs.c.gcda')
-                    telnet_p.pexpect_proc.expect_exact('Targets disconnected.')
-                except TIMEOUT:
-                    # Print what is happening with DUT. Limit the size if it is in loop and generating output.
-                    Utility.console_log(dut.read(size=1000))
-                    raise
-
-            dump_coverage()
-            dut.expect('GCOV data have been dumped.', timeout=5)
-            expect_counter_output(1)
-            dut.expect('Ready to dump GCOV data...', timeout=5)
-            dump_coverage()
-            dut.expect('GCOV data have been dumped.', timeout=5)
-
-            for i in range(2, 6):
-                expect_counter_output(i)
+    openocd_cmd_log = os.path.join(proj_path, 'openocd_cmd.log')
+
+    with ttfw_idf.OCDBackend(os.path.join(proj_path, 'openocd.log'), dut.app.target) as oocd:
+        dut.start_app()
+
+        def expect_counter_output(loop, timeout=10):
+            dut.expect_all('blink_dummy_func: Counter = {}'.format(loop),
+                           'some_dummy_func: Counter = {}'.format(loop * 2),
+                           timeout=timeout)
+
+        expect_counter_output(0, timeout=20)
+        dut.expect('Ready to dump GCOV data...', timeout=5)
+
+        def dump_coverage():
+            try:
+                response = oocd.cmd_exec('esp gcov dump')
+                with open(openocd_cmd_log, 'a') as f:
+                    f.write(response)
+
+                assert all(x in response for x in ['Targets connected.',
+                                                   'gcov_example_main.c.gcda',
+                                                   'gcov_example_func.c.gcda',
+                                                   'some_funcs.c.gcda',
+                                                   'Targets disconnected.',
+                                                   ])
+
+            except AssertionError:
+                # Print what is happening with DUT. Limit the size if it is in loop and generating output.
+                Utility.console_log(dut.read(size=1000))
+                raise
+
+        dump_coverage()
+        dut.expect('GCOV data have been dumped.', timeout=5)
+        expect_counter_output(1)
+        dut.expect('Ready to dump GCOV data...', timeout=5)
+        dump_coverage()
+        dut.expect('GCOV data have been dumped.', timeout=5)
+
+        for i in range(2, 6):
+            expect_counter_output(i)
 
 
 if __name__ == '__main__':

examples/system/sysview_tracing/example_test.py

@@ -1,8 +1,10 @@
 from __future__ import unicode_literals
 from io import open
+import debug_backend
 import os
 import re
 import tempfile
+import time
 import ttfw_idf
 
 
@@ -29,7 +31,7 @@ def test_examples_sysview_tracing(env, extra_data):
             new_content = new_content.replace('file:///tmp/sysview_example.svdat', 'file://{}'.format(tempfiles[1]), 1)
             f_out.write(new_content)
 
-        with ttfw_idf.OCDProcess(os.path.join(proj_path, 'openocd.log')):
+        with ttfw_idf.OCDBackend(os.path.join(proj_path, 'openocd.log'), dut.app.target) as oocd:
             dut.start_app()
 
             def dut_expect_task_event():
@@ -37,17 +39,20 @@ def test_examples_sysview_tracing(env, extra_data):
 
             dut_expect_task_event()
 
-            gdb_args = '-x {} --directory={}'.format(tempfiles[0], os.path.join(proj_path, 'main'))
-            with ttfw_idf.GDBProcess(os.path.join(proj_path, 'gdb.log'), elf_path, dut.app.target, gdb_args) as gdb:
-                gdb.pexpect_proc.expect_exact('Thread 1 hit Breakpoint 1, app_main ()')
-                gdb.pexpect_proc.expect_exact('Targets connected.')
-                gdb.pexpect_proc.expect(re.compile(r'\d+'))
+            gdb_log = os.path.join(proj_path, 'gdb.log')
+            gdb_workdir = os.path.join(proj_path, 'main')
+            with ttfw_idf.GDBBackend(gdb_log, elf_path, dut.app.target, tempfiles[0], gdb_workdir) as p:
+                p.gdb.wait_target_state(debug_backend.TARGET_STATE_RUNNING)
+                stop_reason = p.gdb.wait_target_state(debug_backend.TARGET_STATE_STOPPED)
+                assert stop_reason == debug_backend.TARGET_STOP_REASON_BP, 'STOP reason: {}'.format(stop_reason)
 
                 dut.expect('example: Created task')  # dut has been restarted by gdb since the last dut.expect()
                 dut_expect_task_event()
 
-                gdb.pexpect_proc.sendcontrol('c')
-                gdb.pexpect_proc.expect_exact('(gdb)')
+                # Do a sleep while sysview samples are captured.
+                time.sleep(3)
+                # GDBMI isn't responding now to any commands, therefore, the following command is issued to openocd
+                oocd.cmd_exec('esp sysview stop')
     finally:
         for x in tempfiles:
             try:

examples/system/sysview_tracing_heap_log/example_test.py

@@ -1,5 +1,6 @@
 from __future__ import unicode_literals
 from io import open
+import debug_backend
 import os
 import re
 import tempfile
@@ -29,14 +30,17 @@ def test_examples_sysview_tracing_heap_log(env, extra_data):
             new_content = new_content.replace('file:///tmp/heap_log.svdat', 'file://{}'.format(tempfiles[1]), 1)
             f_out.write(new_content)
 
-        with ttfw_idf.OCDProcess(os.path.join(proj_path, 'openocd.log')):
+        with ttfw_idf.OCDBackend(os.path.join(proj_path, 'openocd.log'), dut.app.target):
             dut.start_app()
             dut.expect('esp_apptrace: Initialized TRAX on CPU0')
 
-            gdb_args = '-x {} --directory={}'.format(tempfiles[0], os.path.join(proj_path, 'main'))
-            with ttfw_idf.GDBProcess(os.path.join(proj_path, 'gdb.log'), elf_path, dut.app.target, gdb_args) as gdb:
-                gdb.pexpect_proc.expect_exact('Thread 1 hit Temporary breakpoint 2, heap_trace_stop ()')
-                gdb.pexpect_proc.expect_exact('(gdb)')
+            gdb_log = os.path.join(proj_path, 'gdb.log')
+            gdb_workdir = os.path.join(proj_path, 'main')
+            with ttfw_idf.GDBBackend(gdb_log, elf_path, dut.app.target, tempfiles[0], gdb_workdir) as p:
+                for _ in range(2):  # There are two breakpoints
+                    p.gdb.wait_target_state(debug_backend.TARGET_STATE_RUNNING)
+                    stop_reason = p.gdb.wait_target_state(debug_backend.TARGET_STATE_STOPPED)
+                    assert stop_reason == debug_backend.TARGET_STOP_REASON_BP, 'STOP reason: {}'.format(stop_reason)
 
                 # dut has been restarted by gdb since the last dut.expect()
                 dut.expect('esp_apptrace: Initialized TRAX on CPU0')

tools/ci/config/target-test.yml

@@ -298,6 +298,7 @@ example_test_009:
       expire_in: 1 week
   variables:
     SETUP_TOOLS: "1"
+    PYTHON_VER: 3
 
 example_test_010:
   extends: .example_test_template
@@ -361,6 +362,7 @@ test_app_test_001:
       expire_in: 1 week
   variables:
     SETUP_TOOLS: "1"
+    PYTHON_VER: 3
 
 test_app_test_002:
   extends: .test_app_template

tools/ci/python_packages/ttfw_idf/DebugUtils.py

@@ -15,7 +15,10 @@
 from __future__ import unicode_literals
 from io import open
 from tiny_test_fw import Utility
+import debug_backend
+import logging
 import pexpect
+import pygdbmi.gdbcontroller
 
 
 class CustomProcess(object):
@@ -37,63 +40,68 @@ class CustomProcess(object):
         self.f.close()
 
 
-class OCDProcess(CustomProcess):
-    def __init__(self, logfile_path, extra_args='', verbose=True):
+class OCDBackend(object):
+    def __init__(self, logfile_path, target, cfg_cmds=[], extra_args=[]):
         # TODO Use configuration file implied by the test environment (board)
-        cmd = 'openocd {} -f board/esp32-wrover-kit-3.3v.cfg'.format(extra_args)
-        super(OCDProcess, self).__init__(cmd, logfile_path, verbose)
-        patterns = ['Info : Listening on port 3333 for gdb connections']
+        self.oocd = debug_backend.create_oocd(chip_name=target,
+                                              oocd_exec='openocd',
+                                              oocd_scripts=None,
+                                              oocd_cfg_files=['board/esp32-wrover-kit-3.3v.cfg'],
+                                              oocd_cfg_cmds=cfg_cmds,
+                                              oocd_debug=2,
+                                              oocd_args=extra_args,
+                                              host='localhost',
+                                              log_level=logging.DEBUG,
+                                              log_stream_handler=None,
+                                              log_file_handler=logging.FileHandler(logfile_path, 'w'),
+                                              scope=None)
+        self.oocd.start()
 
-        try:
-            while True:
-                i = self.pexpect_proc.expect_exact(patterns, timeout=30)
-                # TIMEOUT or EOF exceptions will be thrown upon other errors
-                if i == 0:
-                    if self.verbose:
-                        Utility.console_log('openocd is listening for gdb connections')
-                    break  # success
-        except Exception:
-            if self.verbose:
-                Utility.console_log('openocd initialization has failed', 'R')
-            raise
-
-    def close(self):
-        try:
-            self.pexpect_proc.sendcontrol('c')
-            self.pexpect_proc.expect_exact('shutdown command invoked')
-        except Exception:
-            if self.verbose:
-                Utility.console_log('openocd needs to be killed', 'O')
-            super(OCDProcess, self).close()
-
-
-class GDBProcess(CustomProcess):
-    def __init__(self, logfile_path, elffile_path, target, extra_args='', verbose=True):
-        cmd = 'xtensa-{}-elf-gdb {} {}'.format(target, extra_args, elffile_path)
-        super(GDBProcess, self).__init__(cmd, logfile_path, verbose)
-
-    def close(self):
-        try:
-            self.pexpect_proc.sendline('q')
-            self.pexpect_proc.expect_exact('Quit anyway? (y or n)')
-            self.pexpect_proc.sendline('y')
-            self.pexpect_proc.expect_exact('Ending remote debugging.')
-        except Exception:
-            if self.verbose:
-                Utility.console_log('gdb needs to be killed', 'O')
-            super(GDBProcess, self).close()
+    def __enter__(self):
+        return self
 
+    def __exit__(self, type, value, traceback):
+        self.oocd.stop()
+
+    def cmd_exec(self, cmd):
+        return self.oocd.cmd_exec(cmd)
+
+
+class GDBBackend(object):
+    def __init__(self, logfile_path, elffile_path, target, gdbinit_path=None, working_dir=None):
+        self.gdb = debug_backend.create_gdb(chip_name=target,
+                                            gdb_path='xtensa-{}-elf-gdb'.format(target),
+                                            remote_target=None,
+                                            extended_remote_mode=False,
+                                            gdb_log_file=logfile_path,
+                                            log_level=None,
+                                            log_stream_handler=None,
+                                            log_file_handler=None,
+                                            scope=None)
+        if working_dir:
+            self.gdb.console_cmd_run('directory {}'.format(working_dir))
+        self.gdb.exec_file_set(elffile_path)
+        if gdbinit_path:
+            try:
+                self.gdb.console_cmd_run('source {}'.format(gdbinit_path))
+            except debug_backend.defs.DebuggerTargetStateTimeoutError:
+                # The internal timeout is not enough on RPI for more time consuming operations, e.g. "load".
+                # So lets try to apply the commands one-by-one:
+                with open(gdbinit_path, 'r') as f:
+                    for line in f:
+                        line = line.strip()
+                        if len(line) > 0 and not line.startswith('#'):
+                            self.gdb.console_cmd_run(line)
+                            # Note that some commands cannot be applied with console_cmd_run, e.g. "commands"
 
-class TelnetProcess(CustomProcess):
-    def __init__(self, logfile_path, host='localhost', port=4444, verbose=True):
-        cmd = 'telnet {} {}'.format(host, port)
-        super(TelnetProcess, self).__init__(cmd, logfile_path, verbose)
+    def __enter__(self):
+        return self
 
-    def close(self):
+    def __exit__(self, type, value, traceback):
         try:
-            self.pexpect_proc.sendline('exit')
-            self.pexpect_proc.expect_exact('Connection closed by foreign host.')
-        except Exception:
-            if self.verbose:
-                Utility.console_log('telnet needs to be killed', 'O')
-            super(TelnetProcess, self).close()
+            self.gdb.gdb_exit()
+        except pygdbmi.gdbcontroller.NoGdbProcessError as e:
+            #  the debug backend can fail on gdb exit when it tries to read the response after issuing the exit command.
+            Utility.console_log('Ignoring exception: {}'.format(e), 'O')
+        except debug_backend.defs.DebuggerTargetStateTimeoutError:
+            Utility.console_log('Ignoring timeout exception for GDB exit', 'O')

tools/ci/python_packages/ttfw_idf/__init__.py

@@ -20,7 +20,7 @@ import re
 from tiny_test_fw import TinyFW, Utility
 from .IDFApp import IDFApp, Example, LoadableElfTestApp, UT, TestApp  # noqa: export all Apps for users
 from .IDFDUT import IDFDUT, ESP32DUT, ESP32S2DUT, ESP8266DUT, ESP32QEMUDUT  # noqa: export DUTs for users
-from .DebugUtils import OCDProcess, GDBProcess, TelnetProcess, CustomProcess  # noqa: export DebugUtils for users
+from .DebugUtils import OCDBackend, GDBBackend, CustomProcess  # noqa: export DebugUtils for users
 
 # pass TARGET_DUT_CLS_DICT to Env.py to avoid circular dependency issue.
 TARGET_DUT_CLS_DICT = {

tools/test_apps/system/gdb_loadable_elf/app_test.py

@@ -1,12 +1,11 @@
 from __future__ import unicode_literals
+from tiny_test_fw import Utility
+import debug_backend
 import os
-import threading
-import time
-
 import pexpect
 import serial
-
-from tiny_test_fw import Utility
+import threading
+import time
 import ttfw_idf
 
 
@@ -48,26 +47,22 @@ def test_app_loadable_elf(env, extra_data):
     with SerialThread(esp_log_path):
         openocd_log = os.path.join(proj_path, 'openocd.log')
         gdb_log = os.path.join(proj_path, 'gdb.log')
-        gdb_args = '-x {} --directory={}'.format(os.path.join(proj_path, '.gdbinit.ci'),
-                                                 os.path.join(proj_path, 'main'))
-
-        with ttfw_idf.OCDProcess(openocd_log), ttfw_idf.GDBProcess(gdb_log, elf_path, app.target, gdb_args) as gdb:
-            i = gdb.pexpect_proc.expect_exact(['Thread 1 hit Temporary breakpoint 2, app_main ()',
-                                               'Load failed'])
-            if i == 0:
-                Utility.console_log('gdb is at breakpoint')
-            elif i == 1:
-                raise RuntimeError('Load has failed. Please examine the logs.')
-            else:
-                Utility.console_log('i = {}'.format(i))
-                Utility.console_log(str(gdb.pexpect_proc))
-                # This really should not happen. TIMEOUT and EOF failures are exceptions.
-                raise RuntimeError('An unknown error has occurred. Please examine the logs.')
-
-            gdb.pexpect_proc.expect_exact('(gdb)')
-            gdb.pexpect_proc.sendline('b esp_restart')
-            gdb.pexpect_proc.sendline('c')
-            gdb.pexpect_proc.expect_exact('Thread 1 hit Breakpoint 3, esp_restart ()')
+        gdb_init = os.path.join(proj_path, 'gdbinit')
+        gdb_dir = os.path.join(proj_path, 'main')
+
+        with ttfw_idf.OCDBackend(openocd_log, app.target):
+            with ttfw_idf.GDBBackend(gdb_log, elf_path, app.target, gdb_init, gdb_dir) as p:
+                def wait_for_breakpoint():
+                    p.gdb.wait_target_state(debug_backend.TARGET_STATE_RUNNING)
+                    stop_reason = p.gdb.wait_target_state(debug_backend.TARGET_STATE_STOPPED)
+                    assert stop_reason == debug_backend.TARGET_STOP_REASON_BP, 'STOP reason: {}'.format(stop_reason)
+
+                wait_for_breakpoint()
+
+                p.gdb.add_bp('esp_restart')
+                p.gdb.exec_continue()
+
+                wait_for_breakpoint()
 
     if pexpect.run('grep "Restarting now." {}'.format(esp_log_path), withexitstatus=True)[1]:
         raise RuntimeError('Expected output from ESP was not received')