Coverage for cuda / pathfinder / _dynamic_libs / subprocess_protocol.py: 93.33%

1# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. 

2# SPDX-License-Identifier: Apache-2.0 

3 

4from __future__ import annotations (empty)

5 

6import json (empty)

7import sys (empty)

8from dataclasses import dataclass (empty)

9from pathlib import Path (empty)

10from typing import Literal (empty)

11 

12MODE_CANARY: Literal["canary"] = "canary" (empty)

13MODE_LOAD: Literal["load"] = "load" (empty)

14VALID_MODES: tuple[Literal["canary"], Literal["load"]] = (MODE_CANARY, MODE_LOAD) (empty)

15 

16STATUS_OK: Literal["ok"] = "ok" (empty)

17STATUS_NOT_FOUND: Literal["not-found"] = "not-found" (empty)

18 

19DYNAMIC_LIB_SUBPROCESS_MODULE = "cuda.pathfinder._dynamic_libs.dynamic_lib_subprocess" (empty)

20DYNAMIC_LIB_SUBPROCESS_CWD = Path(__file__).resolve().parents[3] (empty)

21 

22 

23@dataclass(frozen=True) (empty)

24class DynamicLibSubprocessPayload: (empty)

25 status: Literal["ok", "not-found"] (empty)

26 abs_path: str | None (empty)

27 

28 

29def format_dynamic_lib_subprocess_payload( (empty)

30 status: Literal["ok", "not-found"], 

31 abs_path: str | None, 

32 *, 

33 error: dict[str, str] | None = None, 

34) -> str: 

35 payload: dict[str, object] = {"status": status, "abs_path": abs_path} 6 ctx1567123

36 if error is not None: 6 ctx1567123

37 payload["error"] = error 3 ctx1123

38 return json.dumps(payload) 6 ctx1567123

39 

40 

41def build_dynamic_lib_subprocess_command(mode: str, libname: str) -> list[str]: (empty)

42 return [sys.executable, "-m", DYNAMIC_LIB_SUBPROCESS_MODULE, mode, libname] 57 ctx18409KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ!#

43 

44 

45def parse_dynamic_lib_subprocess_payload( (empty)

46 stdout: str, 

47 *, 

48 libname: str, 

49 error_label: str, 

50) -> DynamicLibSubprocessPayload: 

51 # Use the final non-empty line in case earlier output lines are emitted. 

52 lines = [line for line in stdout.splitlines() if line.strip()] 53 ctx140KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

53 if not lines: 53 ctx140KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

54 raise RuntimeError(f"{error_label} produced no stdout payload for {libname!r}") 1 ctx14

55 try: 52 ctx10KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

56 payload = json.loads(lines[-1]) 52 ctx10KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

57 except json.JSONDecodeError: 1 ctx10

58 raise RuntimeError(f"{error_label} emitted invalid JSON payload for {libname!r}: {lines[-1]!r}") from None 1 ctx10

59 if not isinstance(payload, dict): 51 ctx1KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

60 raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") 

61 status = payload.get("status") 51 ctx1KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

62 abs_path = payload.get("abs_path") 51 ctx1KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

63 if status == STATUS_OK: 51 ctx1KLdMNOPQbRefghSTijklUcmnopqrstuvwxyzABCDEFVGWHXIYJZ

64 if not isinstance(abs_path, str): 17 ctx1LMNOPQbRSTUcVWXYZ

65 raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") 

66 return DynamicLibSubprocessPayload(status=STATUS_OK, abs_path=abs_path) 17 ctx1LMNOPQbRSTUcVWXYZ

67 if status == STATUS_NOT_FOUND: 36 ctx1KdbefghijklcmnopqrstuvwxyzABCDEFGHIJ

68 if abs_path is not None: 35 ctx1dbefghijklcmnopqrstuvwxyzABCDEFGHIJ

69 raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") 

70 return DynamicLibSubprocessPayload(status=STATUS_NOT_FOUND, abs_path=None) 35 ctx1dbefghijklcmnopqrstuvwxyzABCDEFGHIJ

71 raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") 1 ctx1K