pre-commit mypy-pathfinder cleanup

Author: rwgk

cuda_pathfinder/cuda/pathfinder/_dynamic_libs/dynamic_lib_subprocess.py

@@ -58,7 +58,10 @@ def _load_nvidia_dynamic_lib_for_test(libname: str) -> str:
     if loaded_dl_fresh.was_already_loaded_from_elsewhere:
         raise RuntimeError("loaded_dl_fresh.was_already_loaded_from_elsewhere")
 
-    _validate_abs_path(loaded_dl_fresh.abs_path)
+    abs_path = loaded_dl_fresh.abs_path
+    if not isinstance(abs_path, str):
+        raise RuntimeError(f"loaded_dl_fresh.abs_path is not a string: {abs_path!r}")
+    _validate_abs_path(abs_path)
     assert loaded_dl_fresh.found_via is not None
 
     loaded_dl_from_cache = load_nvidia_dynamic_lib(libname)
@@ -69,10 +72,13 @@ def _load_nvidia_dynamic_lib_for_test(libname: str) -> str:
     supported_libs = SUPPORTED_WINDOWS_DLLS if IS_WINDOWS else SUPPORTED_LINUX_SONAMES
     if not loaded_dl_no_cache.was_already_loaded_from_elsewhere and libname in supported_libs:
         raise RuntimeError("not loaded_dl_no_cache.was_already_loaded_from_elsewhere")
-    if not os.path.samefile(loaded_dl_no_cache.abs_path, loaded_dl_fresh.abs_path):
-        raise RuntimeError(f"not os.path.samefile({loaded_dl_no_cache.abs_path=!r}, {loaded_dl_fresh.abs_path=!r})")
-    _validate_abs_path(loaded_dl_no_cache.abs_path)
-    return loaded_dl_fresh.abs_path
+    abs_path_no_cache = loaded_dl_no_cache.abs_path
+    if not isinstance(abs_path_no_cache, str):
+        raise RuntimeError(f"loaded_dl_no_cache.abs_path is not a string: {abs_path_no_cache!r}")
+    if not os.path.samefile(abs_path_no_cache, abs_path):
+        raise RuntimeError(f"not os.path.samefile({abs_path_no_cache=!r}, {abs_path=!r})")
+    _validate_abs_path(abs_path_no_cache)
+    return abs_path
 
 
 def probe_dynamic_lib_and_print_json(libname: str, mode: str) -> None:

cuda_pathfinder/cuda/pathfinder/_dynamic_libs/load_nvidia_dynamic_lib.py

@@ -31,6 +31,7 @@
     MODE_CANARY,
     STATUS_NOT_FOUND,
     STATUS_OK,
+    DynamicLibSubprocessPayload,
     build_dynamic_lib_subprocess_command,
     parse_dynamic_lib_subprocess_payload,
 )
@@ -116,13 +117,16 @@ def _resolve_system_loaded_abs_path_in_subprocess(libname: str) -> str | None:
     if result.returncode != 0:
         _raise_canary_probe_child_process_error(returncode=result.returncode, stderr=result.stderr)
 
-    payload = parse_dynamic_lib_subprocess_payload(
+    payload: DynamicLibSubprocessPayload = parse_dynamic_lib_subprocess_payload(
         result.stdout,
         libname=libname,
         error_label="Canary probe child process",
     )
+    abs_path: str | None = payload.abs_path
     if payload.status == STATUS_OK:
-        return payload.abs_path
+        if abs_path is None:
+            raise RuntimeError(f"Canary probe child process emitted unexpected payload for {libname!r}: {payload!r}")
+        return abs_path
     if payload.status == STATUS_NOT_FOUND:
         return None
     raise RuntimeError(f"Canary probe child process emitted unexpected payload for {libname!r}: {payload!r}")

cuda_pathfinder/cuda/pathfinder/_dynamic_libs/subprocess_protocol.py

@@ -9,13 +9,13 @@
 from pathlib import Path
 from typing import Literal
 
-MODE_CANARY = "canary"
-MODE_LOAD = "load"
-VALID_MODES: tuple[str, ...] = (MODE_CANARY, MODE_LOAD)
+MODE_CANARY: Literal["canary"] = "canary"
+MODE_LOAD: Literal["load"] = "load"
+VALID_MODES: tuple[Literal["canary"], Literal["load"]] = (MODE_CANARY, MODE_LOAD)
 
-STATUS_OK = "ok"
-STATUS_NOT_FOUND = "not-found"
-VALID_STATUSES: tuple[str, ...] = (STATUS_OK, STATUS_NOT_FOUND)
+STATUS_OK: Literal["ok"] = "ok"
+STATUS_NOT_FOUND: Literal["not-found"] = "not-found"
+VALID_STATUSES: tuple[Literal["ok"], Literal["not-found"]] = (STATUS_OK, STATUS_NOT_FOUND)
 
 DYNAMIC_LIB_SUBPROCESS_MODULE = "cuda.pathfinder._dynamic_libs.dynamic_lib_subprocess"
 DYNAMIC_LIB_SUBPROCESS_CWD = Path(__file__).resolve().parents[3]
@@ -27,7 +27,7 @@ class DynamicLibSubprocessPayload:
     abs_path: str | None
 
 
-def format_dynamic_lib_subprocess_payload(status: str, abs_path: str | None) -> str:
+def format_dynamic_lib_subprocess_payload(status: Literal["ok", "not-found"], abs_path: str | None) -> str:
     if status not in VALID_STATUSES:
         raise ValueError(f"Unsupported subprocess payload status: {status!r}")
     return json.dumps({"status": status, "abs_path": abs_path})