fix: lowercase container name filter for case-sensitive docker ps

- get_container_id() and get_container_name() now lowercase the
  instance name before passing to docker ps --filter, because
  Docker container names are lowercase internally and the filter
  is case-sensitive. This caused container_id to never be captured
  when instance.name contained uppercase chars (e.g. 'Headquarter'),
  breaking terminal WebSocket connections.

- Also guard proc.stdout being None in start_cloudflared_tunnel().

- Add unit tests for get_container_id and get_container_name.

Quality gates: pytest (14 passed), python clean
This commit is contained in:
Alex Blank
2026-05-28 10:56:33 +02:00
parent c63cf7db50
commit 29943ac239
3 changed files with 125 additions and 23 deletions
+2
View File
@@ -49,3 +49,5 @@ apps/web/dist/
.DS_Store
Thumbs.db
/.stoneforge/.worktrees/
# Local Pi runtime state
.atl/
+59 -11
View File
@@ -35,6 +35,7 @@ def ensure_instance_directory(instance_id: str, base_path: str | None = None) ->
"""
if base_path is None:
from src.config import Settings
base_path = Settings().instance_base_path
instance_dir = Path(base_path) / instance_id
instance_dir.mkdir(parents=True, exist_ok=True)
@@ -142,8 +143,9 @@ def get_container_id(instance_name: str) -> str | None:
Returns:
Container ID or None if not found
"""
# Docker container names are lowercase internally; normalize to ensure match
result = subprocess.run(
["docker", "ps", "-q", "--filter", f"name={instance_name}"],
["docker", "ps", "-q", "--filter", f"name={instance_name.lower()}"],
capture_output=True,
text=True,
)
@@ -162,8 +164,16 @@ def get_container_name(instance_name: str) -> str | None:
Returns:
Container name or None if not found
"""
# Docker container names are lowercase internally; normalize to ensure match
result = subprocess.run(
["docker", "ps", "--format", "{{.Names}}", "--filter", f"name={instance_name}"],
[
"docker",
"ps",
"--format",
"{{.Names}}",
"--filter",
f"name={instance_name.lower()}",
],
capture_output=True,
text=True,
)
@@ -173,7 +183,9 @@ def get_container_name(instance_name: str) -> str | None:
return None
def connect_container_to_network(container_name: str, network_name: str = "backend") -> bool:
def connect_container_to_network(
container_name: str, network_name: str = "backend"
) -> bool:
"""Connect a Docker container to an existing network.
Args:
@@ -203,7 +215,9 @@ def get_container_status(container_id: str) -> dict[str, Any]:
"""
result = subprocess.run(
[
"docker", "inspect", "-f",
"docker",
"inspect",
"-f",
"{{.State.Status}}|{{.State.ExitCode}}|{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}",
container_id,
],
@@ -354,18 +368,29 @@ def start_cloudflared_tunnel(
logger.info("Checking connectivity to %s:%d...", container_name, port)
for attempt in range(10):
check = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
f"http://{container_name}:{port}"],
[
"curl",
"-s",
"-o",
"/dev/null",
"-w",
"%{http_code}",
f"http://{container_name}:{port}",
],
capture_output=True,
text=True,
timeout=5,
)
logger.info("Connectivity check %d: http_code=%s", attempt + 1, check.stdout.strip())
logger.info(
"Connectivity check %d: http_code=%s", attempt + 1, check.stdout.strip()
)
if check.returncode == 0:
break
time.sleep(1)
else:
logger.warning("Container %s:%d not responding to curl checks", container_name, port)
logger.warning(
"Container %s:%d not responding to curl checks", container_name, port
)
# Run cloudflared in background, capture output
logger.info("Starting cloudflared tunnel to http://%s:%d", container_name, port)
@@ -381,9 +406,15 @@ def start_cloudflared_tunnel(
start_time = time.time()
url = None
if proc.stdout is None:
proc.terminate()
proc.wait(timeout=5)
raise RuntimeError("Failed to capture cloudflared output")
while time.time() - start_time < timeout:
# Read available output
import select
readable, _, _ = select.select([proc.stdout], [], [], 1.0)
if readable:
line = proc.stdout.readline()
@@ -455,8 +486,17 @@ def check_tunnel_health(url: str, timeout: int = 10) -> dict[str, Any]:
try:
result = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"--max-time", str(timeout), url],
[
"curl",
"-s",
"-o",
"/dev/null",
"-w",
"%{http_code}",
"--max-time",
str(timeout),
url,
],
capture_output=True,
text=True,
timeout=timeout + 5,
@@ -495,7 +535,15 @@ def check_tunnel_health(url: str, timeout: int = 10) -> dict[str, Any]:
except (ValueError, Exception) as e:
error_str = str(e).lower()
# Classify connection errors
if any(err in error_str for err in ["connection refused", "econnrefused", "could not resolve", "nodename"]):
if any(
err in error_str
for err in [
"connection refused",
"econnrefused",
"could not resolve",
"nodename",
]
):
return {
"tunnel_status": "unreachable",
"status_code": None,
@@ -0,0 +1,52 @@
"""Unit tests for docker service utilities."""
from unittest.mock import MagicMock, patch
from src.services.docker import get_container_id, get_container_name
class TestGetContainerId:
"""Tests for get_container_id."""
@patch("subprocess.run")
def test_lowercases_name_for_filter(self, mock_run) -> None:
"""Docker ps name filter is case-sensitive; we must lowercase."""
mock_run.return_value = MagicMock(returncode=0, stdout="abc123\n")
result = get_container_id("MyContainer-ABC")
assert result == "abc123"
call_args = mock_run.call_args[0][0]
# The filter must use lowercase
assert "name=mycontainer-abc" in call_args
@patch("subprocess.run")
def test_returns_none_when_not_found(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="")
result = get_container_id("missing")
assert result is None
class TestGetContainerName:
"""Tests for get_container_name."""
@patch("subprocess.run")
def test_lowercases_name_for_filter(self, mock_run) -> None:
"""Docker ps name filter is case-sensitive; we must lowercase."""
mock_run.return_value = MagicMock(returncode=0, stdout="mycontainer-abc\n")
result = get_container_name("MyContainer-ABC")
assert result == "mycontainer-abc"
call_args = mock_run.call_args[0][0]
assert "name=mycontainer-abc" in call_args
@patch("subprocess.run")
def test_returns_none_when_not_found(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="")
result = get_container_name("missing")
assert result is None