Address linting test issues

Author: kevin-bates

jupyter_server/services/kernelspecs/kernelspec_cache.py

@@ -6,15 +6,15 @@
 import os
 import sys
 from abc import ABC, ABCMeta, abstractmethod
-from typing import Dict, Optional, Union
+from typing import Any, Dict, Optional, Union
 
 # See compatibility note on `group` keyword in https://docs.python.org/3/library/importlib.metadata.html#entry-points
 if sys.version_info < (3, 10):  # pragma: no cover
     from importlib_metadata import EntryPoint, entry_points
 else:  # pragma: no cover
     from importlib.metadata import EntryPoint, entry_points
 
-from jupyter_client.kernelspec import KernelSpec
+from jupyter_client.kernelspec import KernelSpec, KernelSpecManager
 from traitlets.config import LoggingConfigurable
 from traitlets.traitlets import CBool, Instance, Unicode, default
 
@@ -63,10 +63,10 @@ def _monitor_name_default(self):
 
     # The kernelspec cache consists of a dictionary mapping the kernel name to the actual
     # kernelspec data (CacheItemType).
-    cache_items: Dict
+    cache_items: Dict[str, CacheItemType]
     cache_misses: int
 
-    def __init__(self, kernel_spec_manager, **kwargs) -> None:
+    def __init__(self, kernel_spec_manager: KernelSpecManager, **kwargs: Any) -> None:
         """Initialize the cache."""
         super().__init__(**kwargs)
         self.kernel_spec_manager = kernel_spec_manager
@@ -86,7 +86,7 @@ def __del__(self):
             self.kernel_spec_monitor.destroy()
             self.kernel_spec_monitor = None
 
-    async def get_kernel_spec(self, kernel_name: str) -> KernelSpec:
+    async def get_kernel_spec(self, kernel_name: str) -> Optional[KernelSpec]:
         """Get the named kernel specification.
 
         This method is equivalent to calling KernelSpecManager.get_kernel_spec().  If
@@ -130,7 +130,7 @@ def get_item(self, kernel_name: str) -> Optional[KernelSpec]:
         If cache is disabled or the item is not in the cache, None is returned;
         otherwise, a KernelSpec instance of the item is returned.
         """
-        kernelspec = None
+        kernelspec: Optional[KernelSpec] = None
         if self.cache_enabled:
             cache_item = self.cache_items.get(kernel_name.lower())
             if cache_item:  # Convert to KernelSpec
@@ -149,7 +149,7 @@ def get_item(self, kernel_name: str) -> Optional[KernelSpec]:
                 )
         return kernelspec
 
-    def get_all_items(self) -> Dict[str, CacheItemType]:
+    def get_all_items(self) -> Dict:  # type is Dict[str, CacheItemType]
         """Retrieves all kernel specification from the cache.
 
         If cache is disabled or no items are in the cache, an empty dictionary is returned;
@@ -175,9 +175,9 @@ def put_item(self, kernel_name: str, cache_item: Union[KernelSpec, CacheItemType
             self.log.info(f"KernelSpecCache: adding/updating kernelspec: {kernel_name}")
             if type(cache_item) is KernelSpec:
                 cache_item = KernelSpecCache.kernel_spec_to_cache_item(cache_item)
-            self.cache_items[kernel_name.lower()] = cache_item
+            self.cache_items[kernel_name.lower()] = cache_item  # type: ignore
 
-    def put_all_items(self, kernelspecs: Dict[str, CacheItemType]) -> None:
+    def put_all_items(self, kernelspecs: Dict[str, Union[KernelSpec, CacheItemType]]) -> None:
         """Adds or updates a dictionary of kernel specification in the cache."""
         for kernel_name, cache_item in kernelspecs.items():
             self.put_item(kernel_name, cache_item)
@@ -205,6 +205,7 @@ def kernel_spec_to_cache_item(kernelspec: KernelSpec) -> CacheItemType:
     @staticmethod
     def cache_item_to_kernel_spec(cache_item: CacheItemType) -> KernelSpec:
         """Converts a CacheItemType to a KernelSpec instance for user consumption."""
+        assert type(cache_item["spec"]) is dict
         kernel_spec = KernelSpec(resource_dir=cache_item["resource_dir"], **cache_item["spec"])
         return kernel_spec
 
@@ -220,7 +221,7 @@ class KernelSpecMonitorBase(  # type:ignore[misc]
 
     @classmethod
     def create_instance(
-        cls, kernel_spec_cache: KernelSpecCache, **kwargs
+        cls, kernel_spec_cache: KernelSpecCache, **kwargs: Any
     ) -> "KernelSpecMonitorBase":
         """Creates an instance of the monitor class configured on the KernelSpecCache instance."""
 

jupyter_server/services/kernelspecs/monitors/polling_monitor.py

@@ -4,15 +4,15 @@
 import json
 import os
 from hashlib import md5
-from typing import Dict
+from typing import Any, Dict
 
 from overrides import overrides
 from traitlets.traitlets import Float, default
 
 from ..kernelspec_cache import KernelSpecCache, KernelSpecMonitorBase
 
 
-class KernelSpecPollingMonitor(KernelSpecMonitorBase):
+class KernelSpecPollingMonitor(KernelSpecMonitorBase):  # type:ignore[misc]
     """Polling monitor that uses a periodic poll period to reload the kernelspec cache."""
 
     interval_env = "JUPYTER_POLLING_MONITOR_INTERVAL"
@@ -32,7 +32,7 @@ def _interval_default(self):
     # the churn and noise when publishing events
     hash_values: Dict[str, str]
 
-    def __init__(self, kernel_spec_cache: KernelSpecCache, **kwargs):
+    def __init__(self, kernel_spec_cache: KernelSpecCache, **kwargs: Any):
         """Initialize the handler."""
         super().__init__(**kwargs)
         self.kernel_spec_cache: KernelSpecCache = kernel_spec_cache

jupyter_server/services/kernelspecs/monitors/watchdog_monitor.py

@@ -2,24 +2,25 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 import os
+from typing import Any, Set, Tuple
 
 from overrides import overrides
-from watchdog.events import FileMovedEvent, FileSystemEventHandler
+from watchdog.events import FileSystemEventHandler
 from watchdog.observers import Observer
 
 from ..kernelspec_cache import KernelSpecCache, KernelSpecMonitorBase
 
 
-class KernelSpecWatchdogMonitor(KernelSpecMonitorBase):
+class KernelSpecWatchdogMonitor(KernelSpecMonitorBase):  # type:ignore[misc]
     """Watchdog handler that filters on specific files deemed representative of a kernel specification."""
 
-    def __init__(self, kernel_spec_cache: KernelSpecCache, **kwargs):
+    def __init__(self, kernel_spec_cache: KernelSpecCache, **kwargs: Any):
         """Initialize the handler."""
         super().__init__(**kwargs)
         self.kernel_spec_cache: KernelSpecCache = kernel_spec_cache
         self.kernel_spec_manager = self.kernel_spec_cache.kernel_spec_manager
-        self.observed_dirs = set()  # Tracks which directories are being watched
-        self.observer = None
+        self.observed_dirs: Set[str] = set()  # Tracks which directories are being watched
+        self.observer: Any = None
 
     @overrides
     def initialize(self) -> None:
@@ -58,7 +59,7 @@ class WatchDogHandler(FileSystemEventHandler):
     # files in the future) should be added to this list - at which time it should become configurable.
     watched_files = ["kernel.json"]
 
-    def __init__(self, monitor: "KernelSpecWatchdogMonitor", **kwargs):
+    def __init__(self, monitor: "KernelSpecWatchdogMonitor", **kwargs: Any):
         """Initialize the handler."""
         super().__init__(**kwargs)
         self.kernel_spec_cache = monitor.kernel_spec_cache
@@ -73,14 +74,6 @@ def dispatch(self, event):
         """
 
         if os.path.basename(event.src_path) in self.watched_files:
-            src_resource_dir = os.path.dirname(event.src_path)
-            event.src_resource_dir = src_resource_dir
-            event.src_kernel_name = os.path.basename(src_resource_dir)
-            if type(event) is FileMovedEvent:
-                dest_resource_dir = os.path.dirname(event.dest_path)
-                event.dest_resource_dir = dest_resource_dir
-                event.dest_kernel_name = os.path.basename(dest_resource_dir)
-
             super().dispatch(event)
 
     def on_created(self, event):
@@ -89,19 +82,18 @@ def on_created(self, event):
         This will trigger a call to the configured KernelSpecManager to fetch the instance
         associated with the created file, which is then added to the cache.
         """
-        kernel_name = event.src_kernel_name
+        resource_dir, kernel_name = WatchDogHandler._extract_info(event.src_path)
         try:
             kernelspec = self.kernel_spec_cache.kernel_spec_manager.get_kernel_spec(kernel_name)
             self.kernel_spec_cache.put_item(kernel_name, kernelspec)
         except Exception as e:
             self.log.warning(
-                "The following exception occurred creating cache entry for: {src_resource_dir} "
-                "- continuing...  ({e})".format(src_resource_dir=event.src_resource_dir, e=e)
+                f"The following exception occurred creating cache entry for: {resource_dir} - continuing...  ({e})"
             )
 
     def on_deleted(self, event):
         """Fires when a watched file is deleted, triggering a removal of the corresponding item from the cache."""
-        kernel_name = event.src_kernel_name
+        _, kernel_name = WatchDogHandler._extract_info(event.src_path)
         self.kernel_spec_cache.remove_item(kernel_name)
 
     def on_modified(self, event):
@@ -110,14 +102,13 @@ def on_modified(self, event):
         This will trigger a call to the configured KernelSpecManager to fetch the instance
         associated with the modified file, which is then replaced in the cache.
         """
-        kernel_name = event.src_kernel_name
+        resource_dir, kernel_name = WatchDogHandler._extract_info(event.src_path)
         try:
             kernelspec = self.kernel_spec_cache.kernel_spec_manager.get_kernel_spec(kernel_name)
             self.kernel_spec_cache.put_item(kernel_name, kernelspec)
         except Exception as e:
             self.log.warning(
-                "The following exception occurred updating cache entry for: {src_resource_dir} "
-                "- continuing...  ({e})".format(src_resource_dir=event.src_resource_dir, e=e)
+                f"The following exception occurred updating cache entry for: {resource_dir} - continuing...  ({e})"
             )
 
     def on_moved(self, event):
@@ -126,8 +117,15 @@ def on_moved(self, event):
         This will trigger the update of the existing cached item, replacing its resource_dir entry
         with that of the new destination.
         """
-        src_kernel_name = event.src_kernel_name
-        dest_kernel_name = event.dest_kernel_name
+        _, src_kernel_name = WatchDogHandler._extract_info(event.src_path)
+        dest_resource_dir, dest_kernel_name = WatchDogHandler._extract_info(event.dest_path)
         cache_item = self.kernel_spec_cache.remove_item(src_kernel_name)
-        cache_item["resource_dir"] = event.dest_resource_dir
-        self.kernel_spec_cache.put_item(dest_kernel_name, cache_item)
+        if cache_item is not None:
+            cache_item["resource_dir"] = dest_resource_dir
+            self.kernel_spec_cache.put_item(dest_kernel_name, cache_item)
+
+    @staticmethod
+    def _extract_info(dir_name: str) -> Tuple[str, str]:
+        """Extracts the resource directory and kernel_name from the given dir_name."""
+        resource_dir: str = os.path.dirname(dir_name)  # includes kernel_name
+        return resource_dir, os.path.basename(resource_dir)

tests/services/kernelspecs/test_kernelspec_cache.py

@@ -4,12 +4,11 @@
 
 import asyncio
 import json
-import os
 import shutil
+from pathlib import Path
 
 import pytest
 from jupyter_client.kernelspec import NoSuchKernel
-from pytest_jupyter.utils import mkdir
 from traitlets.config import Config
 
 from jupyter_server.services.kernelspecs.kernelspec_cache import KernelSpecCache
@@ -20,38 +19,39 @@
 }
 
 
-def _install_kernelspec(kernels_dir, kernel_name):
+def _install_kernelspec(kernels_dir: Path, kernel_name: str) -> Path:
     """install a sample kernel in a kernels directory"""
-    kernelspec_dir = os.path.join(kernels_dir, kernel_name)
-    os.makedirs(kernelspec_dir)
-    json_file = os.path.join(kernelspec_dir, "kernel.json")
+    kernelspec_dir = kernels_dir / kernel_name
+    kernelspec_dir.mkdir(parents=True)
+    json_file = Path(kernelspec_dir) / "kernel.json"
     named_json = kernelspec_json.copy()
-    named_json["display_name"] = named_json["display_name"].format(kernel_name=kernel_name)
-    with open(json_file, "w") as f:
+    named_json["display_name"] = str(named_json["display_name"]).format(kernel_name=kernel_name)
+    with open(str(json_file), "w") as f:
         json.dump(named_json, f)
     return kernelspec_dir
 
 
-def _modify_kernelspec(kernelspec_dir, kernel_name):
-    json_file = os.path.join(kernelspec_dir, "kernel.json")
+def _modify_kernelspec(kernelspec_dir: str, kernel_name: str) -> None:
+    json_file = Path(kernelspec_dir) / "kernel.json"
     kernel_json = kernelspec_json.copy()
     kernel_json["display_name"] = f"{kernel_name} modified!"
-    with open(json_file, "w") as f:
+    with open(str(json_file), "w") as f:
         json.dump(kernel_json, f)
 
 
-kernelspec_location = pytest.fixture(lambda jp_data_dir: mkdir(jp_data_dir, "kernels"))
-other_kernelspec_location = pytest.fixture(
-    lambda jp_env_jupyter_path: mkdir(jp_env_jupyter_path, "kernels")
-)
+@pytest.fixture
+def other_kernelspec_location(jp_env_jupyter_path):
+    other_location = Path(jp_env_jupyter_path) / "kernels"
+    other_location.mkdir()
+    return other_location
 
 
 @pytest.fixture
-def setup_kernelspecs(jp_environ, kernelspec_location):
+def setup_kernelspecs(jp_environ, jp_kernel_dir):
     # Only populate factory info
-    _install_kernelspec(str(kernelspec_location), "test1")
-    _install_kernelspec(str(kernelspec_location), "test2")
-    _install_kernelspec(str(kernelspec_location), "test3")
+    _install_kernelspec(jp_kernel_dir, "test1")
+    _install_kernelspec(jp_kernel_dir, "test2")
+    _install_kernelspec(jp_kernel_dir, "test3")
 
 
 MONITORS = ["watchdog-monitor", "polling-monitor"]
@@ -83,7 +83,7 @@ def kernel_spec_cache(
     app.clear_instance()
 
 
-def get_delay_factor(kernel_spec_cache: KernelSpecCache):
+def get_delay_factor(kernel_spec_cache: KernelSpecCache) -> float:
     if kernel_spec_cache.cache_enabled:
         if kernel_spec_cache.monitor_name == "polling-monitor":
             return 2.0
@@ -117,11 +117,11 @@ async def test_get_modified_spec(kernel_spec_cache):
     assert kspec.display_name == "test2 modified!"
 
 
-async def test_add_spec(kernel_spec_cache, kernelspec_location, other_kernelspec_location):
+async def test_add_spec(kernel_spec_cache, jp_kernel_dir, other_kernelspec_location):
     with pytest.raises(NoSuchKernel):
         await kernel_spec_cache.get_kernel_spec("added")  # this will increment cache_miss
 
-    _install_kernelspec(str(other_kernelspec_location), "added")
+    _install_kernelspec(other_kernelspec_location, "added")
     # this will increment cache_miss prior to load
     kspec = await kernel_spec_cache.get_kernel_spec("added")
 
@@ -130,7 +130,7 @@ async def test_add_spec(kernel_spec_cache, kernelspec_location, other_kernelspec
     assert kernel_spec_cache.cache_misses == (2 if kernel_spec_cache.cache_enabled else 0)
 
     # Add another to an existing observed directory, no cache miss here
-    _install_kernelspec(str(kernelspec_location), "added2")
+    _install_kernelspec(jp_kernel_dir, "added2")
     await asyncio.sleep(
         get_delay_factor(kernel_spec_cache)
     )  # sleep to allow cache to add item (no cache miss in this case)