Coverage for cli/_container_runtime.py: 98%

1""" 

2Container runtime detection (Docker, Finch, Podman) — shared helper. 

3 

4Originally part of cli/stacks.py for CDK asset bundling; extracted so 

5the new cli/images.py ImageManager can reuse the cached detection 

6without duplicating the probe logic. 

7 

8CDK requires a container runtime to build Lambda function assets, and 

9the image registry uses the same runtime for ``docker build`` / 

10``docker push`` calls. This module checks for available runtimes in 

11order of preference and verifies they are actually running (not just 

12installed). 

13 

14Priority order: docker > finch > podman. 

15 

16If the ``CDK_DOCKER`` environment variable is set, that value is 

17returned without checking if the runtime is available. 

18""" 

19 

20from __future__ import annotations 

21 

22import logging 

23import os 

24import shutil 

25import subprocess 

26 

27# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

28# Flowchart(s) generated from this file: 

29# * ``detect_container_runtime`` -> ``diagrams/code_diagrams/cli/_container_runtime.detect_container_runtime.html`` 

30# (PNG: ``diagrams/code_diagrams/cli/_container_runtime.detect_container_runtime.png``) 

31# Regenerate with ``python diagrams/code_diagrams/generate.py``. 

32# <pyflowchart-code-diagram> END 

33 

34 

35logger = logging.getLogger(__name__) 

36 

37# Cached result for container runtime detection. 

38# 

39# Sentinel pattern: ``_UNCHECKED`` means the probe has not run yet; 

40# any other value (including ``None``, which means "no runtime found") 

41# is the cached result of the last probe. Using a single sentinel 

42# instead of two separate ``_cache`` / ``_checked`` globals keeps the 

43# cache state idempotent under concurrent first-callers and avoids 

44# the static-analysis false positive on a stand-alone bool flag. 

45_UNCHECKED: object = object() 

46_container_runtime_cache: str | None | object = _UNCHECKED 

47 

48 

49def detect_container_runtime() -> str | None: 

50 """ 

51 Detect available container runtime (cached). 

52 

53 Returns: 

54 Runtime name (``"docker"``, ``"finch"``, or ``"podman"``) if a 

55 runtime is found and running, ``None`` if nothing is available. 

56 

57 Note: 

58 If the ``CDK_DOCKER`` environment variable is set, that value 

59 is returned without checking if the runtime is available. 

60 """ 

61 global _container_runtime_cache 

62 if _container_runtime_cache is not _UNCHECKED: 

63 # ``_container_runtime_cache`` is narrowed to ``str | None`` once 

64 # past the sentinel check, but mypy can't infer that across the 

65 # ``object`` union. The runtime cast is explicit. 

66 return _container_runtime_cache # type: ignore[return-value] 

67 

68 result = _detect_container_runtime_uncached() 

69 _container_runtime_cache = result 

70 return result 

71 

72 

73def _detect_container_runtime_uncached() -> str | None: 

74 """Uncached implementation of container runtime detection.""" 

75 # Check if CDK_DOCKER is already set 

76 if os.environ.get("CDK_DOCKER"): 

77 return os.environ["CDK_DOCKER"] 

78 

79 # Try docker first 

80 if shutil.which("docker"): 

81 # Verify docker is actually running 

82 try: 

83 result = subprocess.run( 

84 ["docker", "info"], 

85 capture_output=True, 

86 timeout=5, 

87 ) 

88 if result.returncode == 0: 

89 return "docker" 

90 except Exception as e: 

91 logger.debug("docker info check failed: %s", e) 

92 

93 # Try finch as fallback 

94 if shutil.which("finch"): 

95 try: 

96 result = subprocess.run( 

97 ["finch", "info"], 

98 capture_output=True, 

99 timeout=5, 

100 ) 

101 if result.returncode == 0: 

102 return "finch" 

103 except Exception as e: 

104 logger.debug("finch info check failed: %s", e) 

105 

106 # Try podman as last resort 

107 if shutil.which("podman"): 

108 try: 

109 result = subprocess.run( 

110 ["podman", "info"], 

111 capture_output=True, 

112 timeout=5, 

113 ) 

114 if result.returncode == 0: 114 ↛ 119line 114 didn't jump to line 119 because the condition on line 114 was always true

115 return "podman" 

116 except Exception as e: 

117 logger.debug("podman info check failed: %s", e) 

118 

119 return None