Coverage for cli/_container_runtime.py: 98.18%

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# Generated at (UTC): 2026-07-18T01:03:40Z 

29# Flowchart(s) generated from this file: 

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

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

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

33# <pyflowchart-code-diagram> END 

34 

35 

36logger = logging.getLogger(__name__) 

37 

38# Cached result for container runtime detection. 

39# 

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

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

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

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

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

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

46_UNCHECKED: object = object() 

47_container_runtime_cache: str | None | object = _UNCHECKED 

48 

49 

50def detect_container_runtime() -> str | None: 

51 """ 

52 Detect available container runtime (cached). 

53 

54 Returns: 

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

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

57 

58 Note: 

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

60 is returned without checking if the runtime is available. 

61 """ 

62 global _container_runtime_cache 

63 if _container_runtime_cache is not _UNCHECKED: 

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

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

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

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

68 

69 result = _detect_container_runtime_uncached() 

70 _container_runtime_cache = result 

71 return result 

72 

73 

74def _detect_container_runtime_uncached() -> str | None: 

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

76 # Check if CDK_DOCKER is already set 

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

78 return os.environ["CDK_DOCKER"] 

79 

80 # Try docker first 

81 if shutil.which("docker"): 

82 # Verify docker is actually running 

83 try: 

84 result = subprocess.run( 

85 ["docker", "info"], 

86 capture_output=True, 

87 timeout=5, 

88 ) 

89 if result.returncode == 0: 

90 return "docker" 

91 except Exception as e: 

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

93 

94 # Try finch as fallback 

95 if shutil.which("finch"): 

96 try: 

97 result = subprocess.run( 

98 ["finch", "info"], 

99 capture_output=True, 

100 timeout=5, 

101 ) 

102 if result.returncode == 0: 

103 return "finch" 

104 except Exception as e: 

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

106 

107 # Try podman as last resort 

108 if shutil.which("podman"): 

109 try: 

110 result = subprocess.run( 

111 ["podman", "info"], 

112 capture_output=True, 

113 timeout=5, 

114 ) 

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

116 return "podman" 

117 except Exception as e: 

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

119 

120 return None