test: Add lockingto pyright test to allow parall testion (from UX)

Signed-off-by: Jos Verlinde <[email protected]>

Author: Josverl

.gitignore

@@ -61,3 +61,4 @@ snippets/*/typings
 typings_test
 typings
 empty
+**/*_lock.file

poetry.lock

@@ -98,6 +98,7 @@ pytest-mock = "^3.10.0"
 sourcery-cli = "^1.0.3"
 mpremote = { git = "https://github.com/Josverl/mpremote", subdirectory = "tools/mpremote", optional = true }
 ipykernel = "^6.23.1"
+fasteners = "^0.19"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -13,8 +13,10 @@
 ow.select_rom(b"12345678")  # select a specific device by its ROM code
 
 assert_type(ow, onewire.OneWire)
-assert_type(ow.write(b"123"), None)
-assert_type(ow.select_rom(b"12345678"), None)
+
+# there was no onewire documatation before 1.19.1
+# assert_type(ow.write(b"123"), None)
+# assert_type(ow.select_rom(b"12345678"), None)
 
 # assert_type(ow.scan(), list)
 # assert_type(ow.reset(), None)

snippets/check_esp32/check_onewire.py

@@ -1,8 +1,29 @@
+"""Pytest configuration file for snippets tests.
+
+- snip_path
+  returns the path to the feature folder (feat_xxxx) or check folder (check_xxxx)
+
+- type_stub_cache_path
+  Is used to install the type stubs for the given portboard and version and cache it for 24 hours to speed up tests
+  Returns the path to the cache folder
+
+- install_stubs
+  is the function that does the actual pip install to a folder
+
+- copy_type_stubs
+  copies the type stubs from the cache to the feature folder
+
+- pytest_runtest_makereport 
+  is used to add the caplog to the test report to make it avaialble to VSCode test explorer
+
+"""
+
 import shutil
 import subprocess
 import time
 from pathlib import Path
 
+import fasteners
 import pytest
 from loguru import logger as log
 
@@ -11,7 +32,19 @@
 
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
-    # execute all other hooks to obtain the report object
+    """
+    pytest_runtest_makereport hook implementation.
+
+    Executes all other hooks to obtain the report object. Looks at actual failing test calls, not setup/teardown. Adds the caplog errors and warnings to the report.
+
+    Args:
+        item: The pytest Item object.
+        call: The pytest CallInfo object.
+
+    Returns:
+        The pytest Report object.
+
+    """
     outcome = yield
     report = outcome.get_result()
 
@@ -41,37 +74,67 @@ def type_stub_cache_path(
     request: pytest.FixtureRequest,
 ) -> Path:
     """
-    Installs a copy of the type stubs for the given portboard and version.
-    Returns the path to the cache folder
+    Installs a copy of the type stubs for the given portboard and version. Returns the path to the cache folder.
+
+    Args:
+        portboard: The portboard.
+        version: The version.
+        stub_source: The stub source.
+        pytestconfig: The pytest Config object.
+        request: The pytest FixtureRequest object.
+
+    Returns:
+        Path: The path to the cache folder.
     """
 
     log.trace(f"setup install type_stubs to cache: {stub_source}, {version}, {portboard}")
     flatversion = version.replace(".", "_")
     cache_key = f"stubber/{stub_source}/{version}/{portboard}"
     # cache_path = pytestconfig.rootpath / "snippets" / "typings_cache"
-    tsc_path = request.config.cache.makedir(
-        f"typings_{flatversion}_{portboard}_stub_{stub_source}"
+    tsc_path = Path(
+        request.config.cache.makedir(f"typings_{flatversion}_{portboard}_stub_{stub_source}")
     )
+    # prevent simultaneous updates to the cache
+    cache_lock = fasteners.InterProcessLock(tsc_path.parent / f"{tsc_path.name}.lock")
     # check if stubs are already installed to the cache
-    if (tsc_path / "micropython.pyi").exists():
-        # check if stubs are in the cache
-        timestamp = request.config.cache.get(cache_key, None)
-        # if timestamp is not older than 24 hours, use cache
-
-        if timestamp and timestamp > (time.time() - MAX_CACHE_AGE):
-            log.trace(f"Using cached type stubs for {portboard} {version}")
-            return tsc_path
-
-    ok = install_stubs(portboard, version, stub_source, pytestconfig, flatversion, tsc_path)
-    if not ok:
-        pytest.skip(f"Could not install stubs for {portboard} {version}")
-    # add the timestamp to the cache
-    request.config.cache.set(cache_key, time.time())
+    with cache_lock:
+        if (tsc_path / "micropython.pyi").exists():
+            # check if stubs are in the cache
+            timestamp = request.config.cache.get(cache_key, None)
+            # if timestamp is not older than 24 hours, use cache
+
+            if timestamp and timestamp > (time.time() - MAX_CACHE_AGE):
+                log.trace(f"Using cached type stubs for {portboard} {version}")
+                return tsc_path
+
+        ok = install_stubs(portboard, version, stub_source, pytestconfig, flatversion, tsc_path)
+        if not ok:
+            pytest.skip(f"Could not install stubs for {portboard} {version}")
+        # add the timestamp to the cache
+        request.config.cache.set(cache_key, time.time())
+
     return tsc_path
 
 
-def install_stubs(portboard, version, stub_source, pytestconfig, flatversion, tsc_path) -> bool:
-    "Expensive / Slow function"
+def install_stubs(
+    portboard, version, stub_source, pytestconfig, flatversion, tsc_path: Path
+) -> bool:
+    """
+    Cleans up prior install to avoid stale files.
+    Uses pip to install type stubs for the given portboard and version.
+
+    Args:
+        portboard: The portboard.
+        version: The version.
+        stub_source: The stub source.
+        pytestconfig: The pytest Config object.
+        flatversion: The flat version.
+        tsc_path: The path to the cache folder.
+
+    Returns:
+        bool: True if the installation was successful, False otherwise.
+    """
+
     # clean up prior install to avoid stale files
     if tsc_path.exists():
         shutil.rmtree(tsc_path, ignore_errors=True)
@@ -83,7 +146,10 @@ def install_stubs(portboard, version, stub_source, pytestconfig, flatversion, ts
     else:
         foldername = f"micropython-{flatversion}-{portboard}-stubs"
         stubsource = pytestconfig.inipath.parent / f"repos/micropython-stubs/publish/{foldername}"
+        if not stubsource.exists():
+            pytest.skip(f"Could not find stubs for {portboard} {version} at {stubsource}")
         cmd = f"pip install {stubsource} --target {tsc_path} --no-user"
+
     try:
         subprocess.run(cmd, shell=False, check=True, capture_output=True, text=True)
     except subprocess.CalledProcessError as e:
@@ -96,6 +162,17 @@ def install_stubs(portboard, version, stub_source, pytestconfig, flatversion, ts
 
 @pytest.fixture(scope="function")
 def snip_path(feature: str, pytestconfig) -> Path:
+    """
+    Get the path to the feat_ or check_ folder.
+
+    Args:
+        feature: The feature.
+        pytestconfig: The pytest Config object.
+
+    Returns:
+        Path: The path to the feature folder.
+    """
+
     snip_path = pytestconfig.inipath.parent / "snippets" / f"feat_{feature}"
     if not snip_path.exists():
         snip_path = pytestconfig.inipath.parent / "snippets" / f"check_{feature}"
@@ -107,15 +184,27 @@ def copy_type_stubs(
     portboard: str, version: str, feature: str, type_stub_cache_path: Path, snip_path: Path
 ):
     """
-    Copies installed/cached typestub fom cache to the feature folder
+    Copies installed/cached type stubs from the cache to the feature folder.
+
+    Args:
+        portboard: The portboard.
+        version: The version.
+        feature: The feature.
+        type_stub_cache_path: The path to the cache folder.
+        snip_path: The path to the feature folder.
     """
-    log.trace(f"- copy_type_stubs: {version}, {portboard} to {feature}")
-    print(f"\n - copy_type_stubs : {version}, {portboard} to {feature}")
-    if not snip_path or not snip_path.exists():
-        # skip if no feature folder
-        pytest.skip(f"no feature folder for {feature}")
-    typings_path = snip_path / "typings"
-    if typings_path.exists():
-        shutil.rmtree(typings_path, ignore_errors=True)
-    shutil.copytree(type_stub_cache_path, typings_path)
-    # time.sleep(0.2)
+    cache_lock = fasteners.InterProcessLock(
+        type_stub_cache_path.parent / f"{type_stub_cache_path.name}.lock"
+    )
+    typecheck_lock = fasteners.InterProcessLock(snip_path / "typecheck_lock.file")
+    with cache_lock:
+        with typecheck_lock:
+            log.trace(f"- copy_type_stubs: {version}, {portboard} to {feature}")
+            print(f"\n - copy_type_stubs : {version}, {portboard} to {feature}")
+            if not snip_path or not snip_path.exists():
+                # skip if no feature folder
+                pytest.skip(f"no feature folder for {feature}")
+            typings_path = snip_path / "typings"
+            if typings_path.exists():
+                shutil.rmtree(typings_path, ignore_errors=True)
+            shutil.copytree(type_stub_cache_path, typings_path)

snippets/conftest.py

@@ -13,10 +13,20 @@ You can update / install the type-stubs in the various typings folders by runnin
 
 ```powershell
 # Update the type stubs
+foreach ($version in @( "latest", "v1.20.0", "v1.19.1", "v1.18.0", "v1.17.0" )) {
+    stubber switch $version
+    stubber get-docstubs 
+    stubber merge --version $version
+    stubber build --version $version
+}
+
+```	
+
 stubber switch latest
 stubber get-docstubs 
 stubber merge --version latest
 stubber build --version latest
+
 .\snippets\install-stubs.ps1
 ```
 ## Test with pytest

snippets/readme.md

@@ -41,19 +41,22 @@
             "path": "feat_networking"
         },
         {
+            "name": "feat_espnow",
             "path": "feat_espnow"
         },
         {
+            "name": "wip_todo",
             "path": "wip_todo"
         },
         {
+            "name": "snippets",
             "path": "."
         }
     ],
     "settings": {
         "python.languageServer": "Pylance",
         "python.analysis.typeCheckingMode": "basic",
-        "python.analysis.diagnosticMode": "workspace",
+        "python.analysis.diagnosticMode": "openFilesOnly",
         "python.autoComplete.extraPaths": [
             "${workspacePath}/typings",
         ],
@@ -90,6 +93,8 @@
             "titleBar.inactiveForeground": "#e7e7e799"
         },
         "peacock.color": "#eb117e",
-        "testExplorer.useNativeTesting": true
+        "testExplorer.useNativeTesting": true,
+        "python.analysis.enableSyncServer": false,
+        "python.testing.autoTestDiscoverOnSaveEnabled": false
     },
 }

snippets/snippets.code-workspace

@@ -0,0 +1,24 @@
+import time
+
+import fasteners
+import pytest
+
+
+# create a test than can be run in parrallel
+@pytest.mark.parametrize("x", [1, 2, 3, 4, 5])
+def test_parallel(x):
+    id = x % 2
+    lock = fasteners.InterProcessLock(f"path/to/lock_{id}.file")
+    with lock:
+        time.sleep(x)
+        ...  # exclusive access
+
+
+# create a test than can be run in parrallel
+@pytest.mark.parametrize("x", [1, 2, 3, 4, 5])
+def test_slow(x):
+    id = x % 2
+    lock = fasteners.InterProcessLock(f"path/to/lock_{id}.file")
+    with lock:
+        time.sleep(x + 3)
+        ...  # exclusive access