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 ctx1!#$678

36 if error is not None: 6 ctx1!#$678

37 payload["error"] = error 3 ctx1678

38 return json.dumps(payload) 6 ctx1!#$678

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] 62 ctx1%95'QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4()

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()] 58 ctx195QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

53 if not lines: 58 ctx195QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

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

55 try: 57 ctx15QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

56 payload = json.loads(lines[-1]) 57 ctx15QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

57 except json.JSONDecodeError: 1 ctx15

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

59 if not isinstance(payload, dict): 56 ctx1QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

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

61 status = payload.get("status") 56 ctx1QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

62 abs_path = payload.get("abs_path") 56 ctx1QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

63 if status == STATUS_OK: 56 ctx1QRdSeTfUbgVhWijkXYlmnopZcqrstuvwxyzABCDEFGHIJK0LM1N2O3P4

64 if not isinstance(abs_path, str): 16 ctx1RSTUbVWXYZc01234

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

66 return DynamicLibSubprocessPayload(status=STATUS_OK, abs_path=abs_path) 16 ctx1RSTUbVWXYZc01234

67 if status == STATUS_NOT_FOUND: 42 ctx1QdefbghijklmnopcqrstuvwxyzABCDEFGHIJKLMNOP

68 if abs_path is not None: 41 ctx1defbghijklmnopcqrstuvwxyzABCDEFGHIJKLMNOP

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

70 return DynamicLibSubprocessPayload(status=STATUS_NOT_FOUND, abs_path=None) 41 ctx1defbghijklmnopcqrstuvwxyzABCDEFGHIJKLMNOP

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