Fix stub loader can never read serial issue

pytest-embedded-idf/tests/test_idf.py

@@ -46,11 +46,11 @@ def test_idf_app(app, dut):
     result.assert_outcomes(passed=1)
 
 
-def test_multi_count_app(testdir):
+def test_multi_dut_app(testdir):
     testdir.makepyfile("""
         import pytest
 
-        def test_multi_count_app(app, dut):
+        def test_multi_dut_app(app, dut):
             assert len(app[0].flash_files) == 3
             assert app[0].target == 'esp32'
 
@@ -74,12 +74,12 @@ def test_multi_count_app(app, dut):
     result.assert_outcomes(passed=1)
 
 
-def test_multi_count_autoflash(testdir):
+def test_multi_dut_autoflash(testdir):
     testdir.makepyfile("""
         import pytest
         import pexpect
 
-        def test_multi_count_autoflash(app, dut):
+        def test_multi_dut_autoflash(app, dut):
             skip_dut = dut[0]
             auto_dut = dut[1]
             with pytest.raises(pexpect.TIMEOUT):
@@ -182,7 +182,7 @@ def test_different_build_dir(testdir):
     testdir.makepyfile("""
         import pytest
 
-        def test_multi_count_app(app, dut):
+        def test_multi_dut_app(app, dut):
             assert app.target == 'esp32'
             assert app.binary_path.endswith('test_new_name')
     """)
@@ -204,3 +204,35 @@ def test_multi_count_app(app, dut):
     )
 
     result.assert_outcomes(passed=1)
+
+
+def test_multi_dut_read_flash(testdir):
+    testdir.makepyfile(r"""
+        import pytest
+        import pexpect
+
+        def test_multi_dut_read_flash(app, serial, dut):
+            dut[0].expect('Hash of data verified.', timeout=5)
+            dut[0].expect_exact('Hello world!', timeout=5)
+
+            dut[1].expect('Hash of data verified.', timeout=5)
+            dut[1].expect_exact('Hello world!', timeout=5)
+
+            serial[0].dump_flash('./test.bin', partition='phy_init')
+            serial[1].dump_flash('./test.bin', partition='phy_init')
+
+            dut[0].expect_exact('Hello world!', timeout=5)
+            dut[1].expect_exact('Hello world!', timeout=5)
+    """)
+
+    result = testdir.runpytest(
+        '-s',
+        '--count', 2,
+        '--app-path', f'{os.path.join(testdir.tmpdir, "hello_world_esp32")}'
+                      f'|'
+                      f'{os.path.join(testdir.tmpdir, "hello_world_esp32c3")}',
+        '--embedded-services', 'esp,idf',
+        '--part-tool', os.path.join(testdir.tmpdir, 'gen_esp32part.py'),
+    )
+
+    result.assert_outcomes(passed=1)

pytest-embedded-serial-esp/pytest_embedded_serial_esp/serial.py

@@ -79,21 +79,33 @@ def _post_init(self):
 
     def use_esptool(func):
         """
-        1. call to the decorated function, could use `self.stub` as the stubbed loader
-        2. call `hard_reset()`
-        3. create the `self.forward_io` thread again.
+        1. close the port and open the port to kill the `self._forward_io` thread
+        2. call `run_stub()`
+        3. call to the decorated function, could use `self.stub` as the stubbed loader
+        4. call `hard_reset()`
+        5. create the `self.forward_io` thread again.
         """
 
         @functools.wraps(func)
         def wrapper(self, *args, **kwargs):
-            settings = self.proc.get_settings()
+            killed = False
+            if self._forward_io_thread and self._forward_io_thread.is_alive():
+                self.proc.close()
+                self.proc.open()  # to kill the redirect stdout thread
+                killed = True
 
+            settings = self.proc.get_settings()
             try:
                 with DuplicateStdout(self.pexpect_proc):
+                    if killed:
+                        self.esp.connect('hard_reset')
+                        self.stub = self.esp.run_stub()
                     ret = func(self, *args, **kwargs)
                     self.stub.hard_reset()
             finally:
                 self.proc.apply_settings(settings)
+                if killed:
+                    self.create_forward_io_thread(self.pexpect_proc)
 
             return ret