Coverage for cli / _container_runtime.py: 100.00%
54 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Container runtime detection (Docker, Finch, Podman) — shared helper.
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.
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).
14Priority order: docker > finch > podman.
16If the ``CDK_DOCKER`` environment variable is set, that value is
17returned without checking if the runtime is available.
18"""
20from __future__ import annotations
22import logging
23import os
24import shutil
25import subprocess
27# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
28# Generated at (UTC): 2026-09-01T14:42:56Z
29# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
30# Flowchart(s) generated from this file:
31# * ``detect_container_runtime`` -> ``diagrams/code_diagrams/cli/_container_runtime.detect_container_runtime.html``
32# (PNG: ``diagrams/code_diagrams/cli/_container_runtime.detect_container_runtime.png``)
33# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
34# <pyflowchart-code-diagram> END
37logger = logging.getLogger(__name__)
39# Cached result for container runtime detection.
40#
41# Sentinel pattern: ``_UNCHECKED`` means the probe has not run yet;
42# any other value (including ``None``, which means "no runtime found")
43# is the cached result of the last probe. Using a single sentinel
44# instead of two separate ``_cache`` / ``_checked`` globals keeps the
45# cache state idempotent under concurrent first-callers and avoids
46# the static-analysis false positive on a stand-alone bool flag.
47_UNCHECKED: object = object()
48_container_runtime_cache: str | None | object = _UNCHECKED
51def detect_container_runtime() -> str | None:
52 """
53 Detect available container runtime (cached).
55 Returns:
56 Runtime name (``"docker"``, ``"finch"``, or ``"podman"``) if a
57 runtime is found and running, ``None`` if nothing is available.
59 Note:
60 If the ``CDK_DOCKER`` environment variable is set, that value
61 is returned without checking if the runtime is available.
62 """
63 global _container_runtime_cache
64 if _container_runtime_cache is not _UNCHECKED:
65 # ``_container_runtime_cache`` is narrowed to ``str | None`` once
66 # past the sentinel check, but mypy can't infer that across the
67 # ``object`` union. The runtime cast is explicit.
68 return _container_runtime_cache # type: ignore[return-value]
70 result = _detect_container_runtime_uncached()
71 _container_runtime_cache = result
72 return result
75def _detect_container_runtime_uncached() -> str | None:
76 """Uncached implementation of container runtime detection."""
77 # Check if CDK_DOCKER is already set
78 if os.environ.get("CDK_DOCKER"):
79 return os.environ["CDK_DOCKER"]
81 # Try docker first
82 if shutil.which("docker"):
83 # Verify docker is actually running
84 try:
85 result = subprocess.run(
86 ["docker", "info"],
87 capture_output=True,
88 timeout=5,
89 )
90 if result.returncode == 0:
91 return "docker"
92 except Exception as e:
93 logger.debug("docker info check failed: %s", e)
95 # Try finch as fallback
96 if shutil.which("finch"):
97 try:
98 result = subprocess.run(
99 ["finch", "info"],
100 capture_output=True,
101 timeout=5,
102 )
103 if result.returncode == 0:
104 return "finch"
105 except Exception as e:
106 logger.debug("finch info check failed: %s", e)
108 # Try podman as last resort
109 if shutil.which("podman"):
110 try:
111 result = subprocess.run(
112 ["podman", "info"],
113 capture_output=True,
114 timeout=5,
115 )
116 if result.returncode == 0:
117 return "podman"
118 except Exception as e:
119 logger.debug("podman info check failed: %s", e)
121 return None
124def container_runtime_error_message(*, allow_cdk_docker: bool = False) -> str:
125 """Build an actionable "no runtime" message for the situation on this host.
127 Detection requires a runtime that is *running*, not merely installed, so the
128 two failure modes need different advice and conflating them wastes real time:
129 a stopped Finch VM was reported as "please install Finch" during a live
130 release-validation run on 2026-08-26, on a machine that already had it. The
131 install hint sent the reader to a page they had already followed.
133 So: if a runtime binary is on PATH but did not answer, say so and give the
134 command that starts it. Only suggest installing when nothing is present.
135 """
136 import shutil
138 installed = [name for name in ("docker", "finch", "podman") if shutil.which(name)]
140 if installed:
141 start_hints = {
142 "docker": "start Docker Desktop (or `open -a Docker` on macOS)",
143 "finch": "finch vm start (first time: finch vm init)",
144 "podman": "podman machine start (first time: podman machine init)",
145 }
146 lines = [
147 "No container runtime is running. These are installed but did not "
148 f"respond: {', '.join(installed)}.",
149 "",
150 "Start one and retry:",
151 ]
152 lines += [f" - {name}: {start_hints[name]}" for name in installed]
153 lines += [
154 "",
155 "Detection runs `<runtime> info` with a 5s timeout, so a runtime whose "
156 "VM is still booting also reports as unavailable — give it a moment "
157 "and retry.",
158 ]
159 if allow_cdk_docker:
160 lines.append("Alternatively set CDK_DOCKER=<path> to a runtime binary.")
161 return "\n".join(lines)
163 lines = [
164 "No container runtime found. Install Docker, Finch, or Podman.",
165 " - Docker: https://docs.docker.com/get-docker/",
166 " - Finch: brew install finch && finch vm init && finch vm start",
167 " - Podman: https://podman.io/getting-started/installation",
168 ]
169 if allow_cdk_docker:
170 lines.append("Alternatively set CDK_DOCKER=<path> to a runtime binary.")
171 return "\n".join(lines)