Merge pull request #221 from horw/feat/issue_182

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

@@ -63,7 +63,7 @@ def __init__(
     ) -> None:
         self._meta = meta
 
-        esptool_target = beta_target or target
+        esptool_target = beta_target or target or 'auto'
         if port is None:
             available_ports = esptool.get_port_list()
             ports = list(set(available_ports) - set(self.occupied_ports.keys()))

pytest-embedded-serial-esp/tests/test_esp.py

@@ -1,5 +1,7 @@
 import re
 
+import pytest
+
 
 def test_detect_port(testdir):
     testdir.makepyfile(
@@ -55,3 +57,42 @@ def test_detect_port_again(dut):
     first_index_of_messages(
         re.compile('^hit port-target cache: .+ - esp32$', re.MULTILINE), caplog.messages, esp32s2_hit_cache_index + 1
     )
+
+
+def test_detect_port_with_local_cache(testdir):
+    pytest.global_port_target_cache = {}
+
+    testdir.makepyfile(r"""
+    import pytest
+
+    def test_empty_port_target_cache_before_init_devices(port_target_cache):
+        assert isinstance(port_target_cache, dict)
+        assert port_target_cache == pytest.global_port_target_cache
+
+    def test_init_devices(dut, port_target_cache):
+        pytest.global_port_target_cache = port_target_cache
+
+    def test_empty_port_target_cache_after_init_devices(port_target_cache):
+        assert port_target_cache == pytest.global_port_target_cache
+    """)
+    result = testdir.runpytest(
+        '-s',
+        '--embedded-services', 'esp',
+        '--cache-dir', './cache-test',
+        '--count', '2',
+    )
+    result.assert_outcomes(passed=3)
+
+    testdir.makepyfile("""
+    import pytest
+
+    def test_load_local_saved_port_target_cache(port_target_cache):
+        assert port_target_cache == pytest.global_port_target_cache
+    """)
+    result = testdir.runpytest(
+        '-s',
+        '--embedded-services', 'esp',
+        '--cache-dir', './cache-test',
+        '--count', '2',
+    )
+    result.assert_outcomes(passed=1)

pytest-embedded/pytest_embedded/plugin.py

@@ -1,11 +1,13 @@
 import contextlib
 import datetime
+import dbm
 import functools
 import importlib
 import io
 import logging
 import multiprocessing
 import os
+import shelve
 import subprocess
 import sys
 import tempfile
@@ -102,6 +104,9 @@ def pytest_addoption(parser):
 
     # supports parametrization
     base_group.addoption('--root-logdir', help='set session-based root log dir. (Default: system temp folder)')
+    base_group.addoption(
+        '--cache-dir', help='set root cache-dir for storing cache files. \n' '(Default: system temp folder)'
+    )
     base_group.addoption(
         '--embedded-services',
         default='',
@@ -528,9 +533,32 @@ def session_tempdir(session_root_logdir) -> str:
 
 
 @pytest.fixture(scope='session')
-def port_target_cache() -> t.Dict[str, str]:
+def cache_dir(request: FixtureRequest) -> str:
+    """Cache dir for pytest-embedded"""
+    _cache_root_dir = os.path.realpath(
+        _request_param_or_config_option_or_default(request, 'cache_dir', tempfile.gettempdir())
+    )
+    _cache_work_dir = os.path.join(_cache_root_dir, 'pytest-embedded', 'pytest-embedded-cache')
+    os.makedirs(_cache_work_dir, exist_ok=True)
+    return _cache_work_dir
+
+
+@pytest.fixture(scope='session')
+def port_target_cache(cache_dir) -> t.Dict[str, str]:
     """Session scoped port-target cache, for esp only"""
-    return {}
+    _cache_file_path = os.path.join(cache_dir, 'port_target_cache')
+    resp: t.Dict[str, str] = {}
+    try:
+        with shelve.open(_cache_file_path) as f:
+            resp = dict(f)
+    except dbm.error:
+        os.remove(_cache_file_path)
+
+    yield resp
+
+    with shelve.open(_cache_file_path) as f:
+        for k, v in resp.items():
+            f[k] = v
 
 
 @pytest.fixture(scope='session')
@@ -1298,7 +1326,7 @@ def dut(
     app: App,
     serial: t.Optional[t.Union['Serial', 'LinuxSerial']],
     qemu: t.Optional['Qemu'],
-) -> Dut:
+) -> t.Union[Dut, t.List[Dut]]:
     """
     A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect
     process, and run `expect()` via its pexpect process.